From e3432dbfaa3bfa0997c5691ee614bda5573b0231 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Mon, 7 Sep 2026 13:29:03 -0400 Subject: [PATCH] Add conversational orchestration plan editing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- application/single_app/config.py | 2 +- .../functions_orchestration_context.py | 21 + .../functions_orchestration_plan_editing.py | 365 ++++++ .../functions_orchestration_plan_revisions.py | 784 ++++++++++++ .../functions_orchestration_planner.py | 89 +- .../functions_orchestration_runs.py | 145 ++- .../functions_orchestration_schema.py | 26 +- .../single_app/route_backend_orchestration.py | 289 ++++- .../src/components/chat/ElicitationCard.tsx | 67 +- .../components/chat/OrchestrationPlanCard.tsx | 48 +- .../chat/OrchestrationPlanEditor.tsx | 413 +++++++ .../chat/OrchestrationPlanPanel.tsx | 78 +- .../components/chat/OrchestrationRunView.tsx | 31 +- application/v2_ui/src/lib/orchestration.ts | 181 ++- .../v2_ui/src/lib/orchestrationController.ts | 415 ++++++- .../v2_ui/src/lib/orchestrationPlan.ts | 1 + .../v2_ui/src/lib/orchestrationResume.ts | 7 + application/v2_ui/src/pages/ChatPage.tsx | 2 + .../v2_ui/src/stores/orchestrationStore.ts | 222 +++- docs/admin/orchestration.md | 7 + .../features/CHAT_ORCHESTRATION.md | 36 +- .../features/V2_ORCHESTRATION_PLAN_EDITING.md | 152 +++ .../review-and-edit-orchestration-plans.md | 92 ++ docs/reference/chat-controls.md | 14 + .../test_route_blueprint_policy_inventory.py | 22 +- ...chestration_conversation_context_routes.py | 85 +- ...est_orchestration_plan_revision_planner.py | 130 ++ ...test_orchestration_plan_revision_routes.py | 427 +++++++ .../test_orchestration_plan_revision_store.py | 841 +++++++++++++ .../test_support/orchestration_revisions.py | 204 ++++ .../fixtures/orchestration/harness_entry.tsx | 28 +- ui_tests/test_v2_orchestration_plan_card.py | 6 +- .../test_v2_orchestration_plan_editing.py | 13 +- ui_tests/test_v2_orchestration_plan_editor.py | 1069 +++++++++++++++++ ...st_v2_orchestration_plan_editor_backend.py | 213 ++++ 35 files changed, 6281 insertions(+), 244 deletions(-) create mode 100644 application/single_app/functions_orchestration_plan_editing.py create mode 100644 application/single_app/functions_orchestration_plan_revisions.py create mode 100644 application/v2_ui/src/components/chat/OrchestrationPlanEditor.tsx create mode 100644 docs/explanation/features/V2_ORCHESTRATION_PLAN_EDITING.md create mode 100644 docs/guides/review-and-edit-orchestration-plans.md create mode 100644 functional_tests/test_orchestration_plan_revision_planner.py create mode 100644 functional_tests/test_orchestration_plan_revision_routes.py create mode 100644 functional_tests/test_orchestration_plan_revision_store.py create mode 100644 functional_tests/test_support/orchestration_revisions.py create mode 100644 ui_tests/test_v2_orchestration_plan_editor.py create mode 100644 ui_tests/test_v2_orchestration_plan_editor_backend.py diff --git a/application/single_app/config.py b/application/single_app/config.py index 1f873d83d..69a549ed4 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -97,7 +97,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.261.101" +VERSION = "0.261.102" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_orchestration_context.py b/application/single_app/functions_orchestration_context.py index f9bff5717..e02513b0e 100644 --- a/application/single_app/functions_orchestration_context.py +++ b/application/single_app/functions_orchestration_context.py @@ -1292,6 +1292,27 @@ def conversation_user_urls(user_message, snapshot=None, message_ids=None, answer return _string_list(urls, limit=8) +def build_capability_request_context( + user_id, identity, user_message, agent_catalog, action_catalog=None, *, allowed_user_urls=None, +): + """Apply the same caller-specific capability gates to planning, revisions, and execution.""" + identity = identity or {} + urls = ( + list(allowed_user_urls) if allowed_user_urls is not None + else conversation_user_urls(user_message) + ) + return { + 'user_id': user_id, + 'user_message': user_message or '', + 'message_urls': urls, + 'user_roles': identity.get('user_roles') or [], + 'user_email': identity.get('user_email'), + 'user_enable_agents': identity.get('user_enable_agents', True), + 'agent_catalog': list(agent_catalog or ()), + 'action_catalog': list(action_catalog or ()), + } + + def build_conversation_signals(messages, user_message, *, truncated=False, message_ids=None): """Project already bounded history for the planner; the route owns loading it.""" allowed = set(message_ids) if message_ids is not None else None diff --git a/application/single_app/functions_orchestration_plan_editing.py b/application/single_app/functions_orchestration_plan_editing.py new file mode 100644 index 000000000..dc8db3cb1 --- /dev/null +++ b/application/single_app/functions_orchestration_plan_editing.py @@ -0,0 +1,365 @@ +# functions_orchestration_plan_editing.py +""" +Planner-assisted changes to a saved, unexecuted plan. + +The revision store owns concurrency and publication. This module prepares the scoped +request, reuses the planner and source authorization boundaries, and never executes work +or writes conversation messages. + +Version: 0.261.102 +""" + +import json +from copy import deepcopy +from datetime import datetime, timezone + +from functions_mixed_source_orchestration import resolve_authorized_source_manifest +from functions_orchestration_context import ( + build_capability_request_context, + build_conversation_signals, + build_elicitation_user_request, + build_planner_context, + conversation_user_urls, + merge_elicitation_context, + normalize_elicitation_answer, + resolve_action_catalog, + resolve_agent_catalog, + resolve_candidate_documents, + resolve_elicitation_references, + validate_clarification_answers, +) +from functions_orchestration_plan_revisions import PlanRevisionError, read_revision_run +from functions_orchestration_planner import plan_request +from functions_orchestration_registry import resolve_available_capabilities +from functions_orchestration_schema import ( + PlanValidationError, + apply_plan_edits, + normalize_plan, + plan_document_ids, +) + +TURN_CONTEXT_FIELDS = ( + 'conversation_id', 'turn_id', 'user_message', 'user_message_id', + 'user_message_fingerprint', 'seeds', 'original_seeds', 'answered_questions', + 'conversation_context', 'request_resolution', 'resolved_message', + 'planning_token_usage', 'prompt_selection', 'edit_user_urls', +) + + +def _turn_context(record): + return {key: deepcopy(record[key]) for key in TURN_CONTEXT_FIELDS if key in record} + + +def _chat_turn(role, content): + return { + 'role': role, 'content': content, + 'timestamp': datetime.now(timezone.utc).isoformat(), + } + + +def _add_usage(context, usage): + previous = context.get('planning_token_usage') or {} + context['planning_token_usage'] = { + key: (previous.get(key) or 0) + ((usage or {}).get(key) or 0) + for key in ('prompt_tokens', 'completion_tokens', 'total_tokens') + } + + +def _editor_urls(context, instruction='', chat=()): + prior_user_urls = [ + url + for turn in reversed(chat[-20:]) + if isinstance(turn, dict) and turn.get('role') == 'user' and isinstance(turn.get('content'), str) + for url in conversation_user_urls(turn.get('content')) + ] + return list(dict.fromkeys([ + *conversation_user_urls(instruction), + *prior_user_urls, + *(context.get('edit_user_urls') or []), + ]))[:8] + + +def revision_allowed_urls(context): + """Only actual user edits and accepted answers can authorize an additional URL.""" + snapshot = context.get('conversation_context') or {} + resolution = context.get('request_resolution') or {} + return list(dict.fromkeys([ + *(context.get('edit_user_urls') or []), + *conversation_user_urls( + context['user_message'], snapshot, resolution.get('message_ids'), + context.get('answered_questions'), + ), + ]))[:8] + + +def _available_sources(context, plan, user_id, settings, candidates=()): + seeds = context.get('seeds') or {} + offered = {item['document_id']: item for item in candidates} + ids = list(dict.fromkeys([ + *(seeds.get('document_ids') or []), + *plan_document_ids(plan, include_disabled=True), + *offered, + ])) + if not ids: + return [] + try: + manifest = resolve_authorized_source_manifest( + ids, user_id, conversation_id=context['conversation_id'], + doc_scope=seeds.get('doc_scope') or 'all', + active_group_ids=seeds.get('active_group_ids') or None, + active_public_workspace_ids=seeds.get('active_public_workspace_ids') or None, + ) + except ValueError as exc: + raise PlanRevisionError( + 'The selected sources could not be used. Start a new request with fewer sources.', + code='invalid_request', status_code=400, + ) from exc + available = { + item['document_id']: item for item in manifest + if item.get('authorization_status') == 'authorized' + } + if set(seeds.get('document_ids') or []) - set(available): + raise PlanRevisionError( + 'A selected source is no longer available. Start a new request with accessible sources.', + code='source_changed', + ) + return [ + { + **offered.get(document_id, {}), + 'document_id': document_id, + 'file_name': item.get('file_name') or item.get('display_name') or document_id, + 'title': item.get('display_name') or item.get('file_name') or document_id, + 'scope': item.get('scope'), + 'selected_by_user': document_id in (seeds.get('document_ids') or []), + } + for document_id, item in available.items() + ] + + +def _revision_catalogs(context, user_id, settings, identity): + seeds = context.get('seeds') or {} + # Resolve current access even for a pinned agent; the ordinary first-plan fast path + # assumes that the composer's selection is still fresh. + agents = resolve_agent_catalog( + user_id, seeds={**seeds, 'agent': None}, settings=settings, + ) + selected = seeds.get('agent') + if selected: + scope = selected.get('scope_type') or ( + 'group' if selected.get('is_group') else + 'global' if selected.get('is_global') else 'personal' + ) + agents = [ + agent for agent in agents + if agent.get('name') == selected.get('name') + and agent.get('scope_type') == scope + and (not selected.get('id') or agent.get('id') == selected['id']) + and ( + scope != 'group' + or agent.get('group_id') == (selected.get('group_id') or selected.get('scope_id')) + ) + ] + if len(agents) != 1: + raise PlanRevisionError( + 'The selected agent is no longer available. Start a new request to choose an agent.', + code='source_changed', + ) + actions = resolve_action_catalog(user_id, seeds=seeds, settings=settings) + caller = build_capability_request_context( + user_id, identity, context.get('resolved_message') or context['user_message'], + agents, actions, allowed_user_urls=revision_allowed_urls(context), + ) + return agents, actions, caller + + +def validate_edited_plan(plan, context, user_id, settings, identity): + """Recheck a restored/generated plan before committing it, including current access.""" + seeds = context.get('seeds') or {} + resolve_elicitation_references( + seeds.get('elicitation_references') or [], + user_id, context['conversation_id'], settings=settings, + ) + candidates = _available_sources(context, plan, user_id, settings) + authorized = {item['document_id'] for item in candidates} + if set(plan_document_ids(plan, include_disabled=True)) - authorized: + raise PlanRevisionError( + 'That version names sources that are no longer available. Your current plan is unchanged.', + code='source_changed', + ) + agents, actions, caller = _revision_catalogs(context, user_id, settings, identity) + capabilities = resolve_available_capabilities( + settings, allowed_ids=settings.get('chat_orchestration_enabled_capabilities'), + request_context=caller, + ) + try: + checked = normalize_plan( + deepcopy(plan), context['conversation_id'], user_id, settings=settings, + approval_mode='manual', authorized_document_ids=authorized, + available_capability_ids=[item['id'] for item in capabilities], + turn_id=context['turn_id'], seeds=seeds, + document_labels={item['document_id']: item['file_name'] for item in candidates}, + agent_names=[item['name'] for item in agents], actions=actions, + ) + except PlanValidationError as exc: + raise PlanRevisionError( + 'That version cannot be used with the current capabilities. Your current plan is unchanged.', + code='source_changed', + ) from exc + if checked['validation']['errors']: + raise PlanRevisionError( + 'That version contains work that is no longer available. Your current plan is unchanged.', + code='source_changed', + ) + checked['validation']['repairs'] = list(dict.fromkeys([ + *(plan.get('validation', {}).get('repairs') or []), + *checked['validation']['repairs'], + ])) + return checked + + +def build_plan_edit_outcome( + record, data, user_id, settings, *, identity, conversation_context, ledger=None, +): + """Return publication arguments; no model response is a committed revision yet.""" + context = _turn_context(record) + context['conversation_context'] = conversation_context + chat = deepcopy(record.get('edit_chat') or []) + action = data['action'] + current_plan = apply_plan_edits( + deepcopy(record['plan']), data.get('edits', record.get('edit_narrowing')), + ) + instruction = data.get('instruction', '') + user_content = instruction + allow_elicitation = True + if action == 'discard': + return { + 'kind': 'discard', + 'chat': [*chat, _chat_turn('assistant', 'The proposed change was cancelled. The current plan is unchanged.')], + } + if action == 'restore': + source = read_revision_run(data['source_run_id'], user_id, context['conversation_id']) + if ( + source.get('turn_id') != record['turn_id'] or source.get('started_at') + or source.get('user_message_id') != record.get('user_message_id') + or source.get('user_message_fingerprint') != record.get('user_message_fingerprint') + or (source.get('revision_root_run_id') or source['id']) + != (record.get('revision_root_run_id') or record['id']) + ): + raise PlanRevisionError('Choose a version of this unexecuted plan.', code='invalid_request', status_code=400) + restored_context = _turn_context(source) + restored_context['conversation_context'] = conversation_context + note = f"Restored version {int(source.get('revision') or 0) + 1}." + return { + 'kind': 'plan', 'document': deepcopy(source['plan']), + 'turn_context': restored_context, 'origin': 'restore', 'instruction': note, + 'chat': [*chat, _chat_turn('user', note), _chat_turn('assistant', note)], + } + if action == 'answer': + pending = record['edit_pending'] + question = pending['elicitation'] + context = deepcopy(pending['turn_context']) + context['conversation_context'] = conversation_context + current_plan = deepcopy(pending['base_plan']) + instruction = pending['instruction'] + validated, answer_context = normalize_elicitation_answer( + question, data['elicitation_response'], data.get('elicitation_context'), + user_id, context['conversation_id'], settings=settings, + ) + if validated['action'] == 'cancel': + return { + 'kind': 'discard', + 'chat': [*chat, _chat_turn('assistant', 'The proposed change was cancelled. The current plan is unchanged.')], + } + context['answered_questions'] = [ + *(context.get('answered_questions') or []), + { + 'elicitation_id': question['elicitation_id'], 'revision': question['revision'], + 'question': question['message'], 'action': validated['action'], + 'answer': validated['content'], 'context': answer_context, + }, + ] + if validated['action'] == 'accept': + context['seeds'] = merge_elicitation_context(context.get('seeds'), answer_context) + allow_elicitation = validated['action'] == 'accept' + answer_text = json.dumps(validated['content'], ensure_ascii=False) + user_content = ( + 'Declined to provide more information.' if validated['action'] == 'decline' + else f"Answer: {answer_text[:1900]}" + (' (excerpt)' if len(answer_text) > 1900 else '') + ) + + validate_clarification_answers(context.get('answered_questions') or []) + seeds = context.get('seeds') or {} + resolve_elicitation_references( + seeds.get('elicitation_references') or [], + user_id, context['conversation_id'], settings=settings, + ) + context['edit_user_urls'] = _editor_urls(context, instruction, chat) + current_request = context.get('resolved_message') or context['user_message'] + changed_request = ( + f'Current task:\n{current_request}\n\n' + f'User-requested change (takes precedence where it changes the task):\n{instruction}' + ) + candidates, _probed = resolve_candidate_documents( + build_elicitation_user_request(changed_request, context.get('answered_questions')), + user_id, seeds=seeds, conversation_id=context['conversation_id'], settings=settings, + ) + candidates = _available_sources(context, current_plan, user_id, settings, candidates) + agents, actions, caller = _revision_catalogs(context, user_id, settings, identity) + resolution = context.get('request_resolution') or {} + signals = build_conversation_signals( + conversation_context['messages'], context['user_message'], + truncated=conversation_context.get('truncated', False), + message_ids=resolution.get('message_ids'), + ) + signals['urls'] = revision_allowed_urls(context) + planner_context = build_planner_context( + changed_request, candidates=candidates, seeds=seeds, ledger=ledger, + signals=signals, agents=agents, actions=actions, + original_message=context['user_message'], request_resolution=resolution, + answered_questions=context.get('answered_questions'), + ) + edit_context = { + 'current_plan': { + key: deepcopy(current_plan.get(key)) for key in ('intent', 'assumptions', 'steps') + }, + 'current_request': current_request, 'instruction': instruction, + 'chat': [{key: turn[key] for key in ('role', 'content')} for turn in chat[-20:]], + } + kind, document = plan_request( + changed_request, planner_context, context['conversation_id'], user_id, + settings=settings, approval_mode='manual', + authorized_document_ids={item['document_id'] for item in candidates}, + turn_id=context['turn_id'], seeds=seeds, + document_labels={item['document_id']: item['file_name'] for item in candidates}, + request_context=caller, edit_context=edit_context, allow_elicitation=allow_elicitation, + revision=int(record.get('revision') or 0) + 1, + ) + _add_usage(context, document.get('token_usage')) + chat.append(_chat_turn('user', user_content)) + if kind == 'elicitation': + document.update({ + 'conversation_id': context['conversation_id'], 'turn_id': context['turn_id'], + }) + return { + 'kind': kind, 'document': document, 'turn_context': context, + 'chat': [*chat, _chat_turn('assistant', document['message'])], + 'pending': { + 'elicitation': document, 'turn_context': context, + 'instruction': instruction, 'base_plan': current_plan, + }, + } + if kind == 'message': + return { + 'kind': kind, 'turn_context': context, + 'chat': [*chat, _chat_turn('assistant', document['message'])], + } + context['resolved_message'] = document.pop('revised_request').strip() + count = sum(step.get('enabled', True) for step in document['steps']) + summary = f"Updated the plan to {count} {'step' if count == 1 else 'steps'}. Review it before running." + if document['validation']['repairs']: + summary += ' Adjustments: ' + ' '.join(document['validation']['repairs']) + return { + 'kind': 'plan', 'document': document, 'turn_context': context, + 'instruction': instruction, 'origin': 'ai', + 'chat': [*chat, _chat_turn('assistant', summary)], + } diff --git a/application/single_app/functions_orchestration_plan_revisions.py b/application/single_app/functions_orchestration_plan_revisions.py new file mode 100644 index 000000000..d02b2e720 --- /dev/null +++ b/application/single_app/functions_orchestration_plan_revisions.py @@ -0,0 +1,784 @@ +# functions_orchestration_plan_revisions.py +""" +Conditional pre-execution editing and execution claims for orchestration plans. + +Revision publication uses a transactional batch in the conversation partition. Neither +an editor lease nor a browser approval may bypass the run's ETag boundary. + +Version: 0.261.102 +""" + +import hashlib +import json +import logging +import uuid +from copy import deepcopy +from datetime import datetime, timedelta, timezone + +from azure.core import MatchConditions +from azure.core.exceptions import AzureError +from azure.cosmos import exceptions + +import functions_orchestration_runs as run_store +from functions_appinsights import log_event +from functions_orchestration_schema import apply_plan_edits, summarize_plan + + +EDIT_CLAIM_SECONDS = 900 +EDIT_RETAINED_SUBMISSIONS = 12 +EDIT_CHAT_LIMIT = 20 +EDIT_CHAT_CONTENT_LIMIT = 4000 +EDIT_HISTORY_PAGE_SIZE = 20 +EDIT_REQUEST_MAX_BYTES = 131072 +EDIT_INSTRUCTION_LIMIT = 2000 +EDIT_NOTE_LIMIT = 600 + +_EDITABLE_STATUSES = {'draft', 'awaiting_approval'} +_RUNNABLE_STATUSES = _EDITABLE_STATUSES | {'approved'} +_CONTEXT_FIELDS = ( + 'seeds', 'answered_questions', 'request_resolution', 'resolved_message', + 'planning_token_usage', 'prompt_selection', 'edit_user_urls', +) +_IMMUTABLE_FIELDS = ( + 'user_message', 'user_message_id', 'user_message_fingerprint', 'turn_id', + 'original_seeds', 'conversation_context', 'snapshot', 'request_fingerprint', +) +_PLAN_FIELDS = ( + 'plan_id', 'run_id', 'turn_id', 'revision', 'conversation_id', 'user_id', + 'planner_contract_version', 'intent', 'assumptions', 'approval', 'status', + 'steps', 'inputs', 'outputs', 'validation', 'edit_version', +) +_STEP_FIELDS = ( + 'step_id', 'capability_id', 'title', 'rationale', 'arguments', 'depends_on', + 'optional', 'enabled', 'estimated_cost', 'phase', 'status', +) +_QUESTION_FIELDS = ( + 'elicitation_id', 'contract_version', 'run_id', 'revision', 'message', + 'requested_schema', 'ui_hints', 'conversation_id', 'turn_id', +) +_REQUEST_FIELDS = { + 'conversation_id', 'expected_version', 'submission_id', 'action', 'edits', +} +_ACTION_FIELDS = { + 'ask': {'instruction'}, + 'restore': {'source_run_id'}, + 'answer': { + 'elicitation_id', 'elicitation_revision', 'elicitation_response', + 'elicitation_context', + }, + 'discard': {'elicitation_id', 'elicitation_revision'}, +} + + +class PlanRevisionError(ValueError): + """An actionable error whose message is safe to return to an API client.""" + + def __init__(self, message, code='plan_changed', status_code=409, current_run_id=None): + super().__init__(message) + self.message = message + self.code = code + self.status_code = status_code + self.current_run_id = current_run_id + + +def _now(): + return datetime.now(timezone.utc) + + +def _valid_id(value): + return ( + isinstance(value, str) and 0 < len(value) <= 200 and value == value.strip() + and not any( + ord(character) < 32 or 0xD800 <= ord(character) <= 0xDFFF + or character in '/\\?#' for character in value + ) + ) + + +def _invalid(message='Invalid plan edit request.'): + return PlanRevisionError(message, code='invalid_request', status_code=400) + + +def _not_found(): + return PlanRevisionError('The plan could not be found.', code='not_found', status_code=404) + + +def _turn_id(record): + return record.get('turn_id') or (record.get('plan') or {}).get('turn_id') + + +def _root_id(record): + return record.get('revision_root_run_id') or record['id'] + + +def _revision(record): + value = record.get('revision', 0) + return value if isinstance(value, int) and not isinstance(value, bool) and value >= 0 else 0 + + +def _owned_run(record, run_id, user_id, conversation_id): + if ( + not all(_valid_id(value) for value in (run_id, user_id, conversation_id)) + or not run_store._is_run_record(record) + or record.get('id') != run_id + or record.get('user_id') != user_id + or record.get('conversation_id') != conversation_id + or not isinstance(record.get('plan'), dict) + ): + return False + plan = record['plan'] + return ( + _valid_id(plan.get('plan_id')) and record.get('run_id', run_id) == run_id + and plan.get('run_id') == run_id + and plan.get('user_id', user_id) == user_id + and plan.get('conversation_id', conversation_id) == conversation_id + and ( + not record.get('turn_id') or not plan.get('turn_id') + or record['turn_id'] == plan['turn_id'] + ) + ) + + +def _same_lineage(left, right): + return ( + left['user_id'] == right.get('user_id') + and left['conversation_id'] == right.get('conversation_id') + and _turn_id(left) == _turn_id(right) + and _root_id(left) == _root_id(right) + ) + + +def read_revision_run(run_id, user_id, conversation_id, *, follow_current=False): + """Read an owned raw run, optionally following only its own revision chain.""" + if not all(_valid_id(value) for value in (run_id, user_id, conversation_id)): + raise _not_found() + original = None + previous = None + visited = set() + while True: + if run_id in visited: + raise _not_found() + visited.add(run_id) + try: + record = run_store.cosmos_orchestration_runs_container.read_item( + item=run_id, partition_key=conversation_id, + ) + except exceptions.CosmosResourceNotFoundError as exc: + raise _not_found() from exc + if not _owned_run(record, run_id, user_id, conversation_id): + raise _not_found() + if original is not None and ( + not _same_lineage(original, record) + or _revision(record) <= _revision(previous) + or record.get('parent_run_id', previous['id']) != previous['id'] + ): + raise _not_found() + if not follow_current or 'superseded_by_run_id' not in record: + return record + target = record['superseded_by_run_id'] + if not _valid_id(target): + raise _not_found() + original = original or record + previous = record + run_id = target + + +def _changed(record): + current_run_id = None + if 'superseded_by_run_id' in record: + current = read_revision_run( + record['id'], record['user_id'], record['conversation_id'], follow_current=True, + ) + current_run_id = current['id'] + return PlanRevisionError( + 'This plan changed. Reload the latest plan before continuing.', + current_run_id=current_run_id, + ) + + +def _busy(): + return PlanRevisionError( + 'A plan edit or planner question is in progress. Finish it before continuing.', + code='edit_in_progress', + ) + + +def _assert_editable(record): + if ( + record.get('status') not in _EDITABLE_STATUSES or record.get('started_at') + or 'superseded_by_run_id' in record + ): + raise _changed(record) + + +def _version(record): + return record.get('edit_version') or record['plan'].get('edit_version') or '' + + +def _check_version(record, expected_version, *, required=False): + current = _version(record) + if ( + (required and not current) + or (current and record.get('edit_version') != record['plan'].get('edit_version')) + or (required and expected_version != current) + or (expected_version is not None and expected_version != current) + ): + raise _changed(record) + + +def _check_plan_id(record, plan_id, *, required=False): + if (required or plan_id is not None) and ( + not _valid_id(plan_id) or plan_id != record['plan'].get('plan_id') + ): + raise _changed(record) + + +def _json_text(value): + try: + text = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False) + size = len(text.encode('utf-8')) + except (TypeError, ValueError, UnicodeError, RecursionError) as exc: + raise _invalid() from exc + if size > EDIT_REQUEST_MAX_BYTES: + raise PlanRevisionError( + 'The plan edit request is too large.', code='request_too_large', status_code=413, + ) + return text + + +def _normalize_edits(plan, edits): + if edits is None: + return {'disabled_step_ids': [], 'removed_document_ids': {}} + _json_text(edits) + if not isinstance(edits, dict) or set(edits) - {'disabled_step_ids', 'removed_document_ids'}: + raise _invalid('Only step disabling and document removal are allowed.') + steps = {step['step_id']: step for step in plan.get('steps') or []} + disabled = edits.get('disabled_step_ids', []) + removed = edits.get('removed_document_ids', {}) + if not isinstance(disabled, list) or not isinstance(removed, dict): + raise _invalid('Invalid step or document removals.') + if any(not _valid_id(step_id) or step_id not in steps for step_id in disabled): + raise _invalid('Choose steps from the current plan.') + if any(steps[step_id].get('capability_id') == 'respond' for step_id in disabled): + raise _invalid('The final answering step cannot be disabled.') + clean_removed = {} + for step_id, document_ids in removed.items(): + if ( + not _valid_id(step_id) or step_id not in steps + or not isinstance(document_ids, list) + ): + raise _invalid('Choose documents from the current plan.') + arguments = steps[step_id].get('arguments') or {} + available = { + document_id for field in ('document_ids', 'right_document_ids') + for document_id in arguments.get(field) or [] + } + if any(not _valid_id(document_id) or document_id not in available for document_id in document_ids): + raise _invalid('Choose documents from the current plan.') + if document_ids: + clean_removed[step_id] = list(dict.fromkeys(document_ids)) + return { + 'disabled_step_ids': list(dict.fromkeys(disabled)), + 'removed_document_ids': clean_removed, + } + + +def _bounded_chat(chat): + if not isinstance(chat, list): + return [] + return [ + { + 'role': entry['role'], + 'content': entry['content'][:EDIT_CHAT_CONTENT_LIMIT], + 'timestamp': entry['timestamp'][:64] if isinstance(entry.get('timestamp'), str) else _now().isoformat(), + } + for entry in chat + if isinstance(entry, dict) and entry.get('role') in ('user', 'assistant') + and isinstance(entry.get('content'), str) + ][-EDIT_CHAT_LIMIT:] + + +def _manual_plan(plan, version): + result = deepcopy(plan) + result.update(status='awaiting_approval', edit_version=version) + result['approval'] = { + **(result.get('approval') or {}), 'mode': 'manual', 'state': 'pending', + 'approved_at': None, 'approved_by': None, + } + return result + + +def _replace(record, updates): + if not record.get('_etag'): + raise _changed(record) + replacement = deepcopy(run_store._strip_cosmos_metadata(record)) + replacement.update(deepcopy(updates)) + replacement['updated_at'] = _now().isoformat() + try: + result = run_store.cosmos_orchestration_runs_container.replace_item( + item=record['id'], body=replacement, etag=record['_etag'], + match_condition=MatchConditions.IfNotModified, + ) + except exceptions.CosmosHttpResponseError as exc: + if exc.status_code in (404, 409, 412): + latest = read_revision_run(record['id'], record['user_id'], record['conversation_id']) + raise _changed(latest) from exc + raise + if isinstance(result, dict) and result.get('_etag'): + return result + return read_revision_run(record['id'], record['user_id'], record['conversation_id']) + + +def begin_plan_edit( + run_id, user_id, conversation_id, *, plan_id, edits=None, expected_version=None, +): + """Establish a manual hold without changing the original executable steps.""" + record = read_revision_run(run_id, user_id, conversation_id) + _assert_editable(record) + _check_plan_id(record, plan_id, required=True) + _check_version(record, expected_version) + if _version(record): + return record + overlay = _normalize_edits(record['plan'], edits) + version = str(uuid.uuid4()) + plan = _manual_plan(record['plan'], version) + return _replace(record, { + 'plan': plan, 'approval': deepcopy(plan['approval']), 'status': 'awaiting_approval', + 'plan_summary': summarize_plan(plan), 'edit_version': version, + 'revision_root_run_id': _root_id(record), 'edit_narrowing': overlay, + 'edit_chat': _bounded_chat(record.get('edit_chat')), + 'edit_pending': None, 'edit_claim': None, 'edit_submissions': [], 'edit_attempts': [], + 'revision_origin': record.get('revision_origin') or 'original', + 'revision_note': record.get('revision_note') or 'Original plan.', + }) + + +def _claim_is_active(claim): + if not isinstance(claim, dict) or not claim: + return False + try: + started = datetime.fromisoformat(claim['started_at']) + deadline = started + timedelta(seconds=EDIT_CLAIM_SECONDS) + if claim.get('expires_at'): + deadline = min(deadline, datetime.fromisoformat(claim['expires_at'])) + return _now() < deadline + except (KeyError, TypeError, ValueError, OverflowError): + return True + + +def _public_plan(plan): + result = {key: deepcopy(plan[key]) for key in _PLAN_FIELDS if key in plan} + result['steps'] = [ + {key: deepcopy(step[key]) for key in _STEP_FIELDS if key in step} + for step in plan.get('steps') or [] + ] + return result + + +def plan_editor_state(record, user_id, *, before_revision=None): + """Project only the canonical plan and bounded, owned editor conversation/history.""" + if not isinstance(record, dict) or not _owned_run( + record, record.get('id'), user_id, record.get('conversation_id'), + ): + raise _not_found() + if before_revision is not None and ( + not isinstance(before_revision, int) or isinstance(before_revision, bool) + or before_revision < 0 + ): + raise _invalid('Invalid history cursor.') + parameters = [ + {'name': '@conversation_id', 'value': record['conversation_id']}, + {'name': '@user_id', 'value': user_id}, + {'name': '@turn_id', 'value': _turn_id(record)}, + {'name': '@revision_root_run_id', 'value': _root_id(record)}, + ] + cursor_filter = '' + if before_revision is not None: + parameters.append({'name': '@before_revision', 'value': before_revision}) + cursor_filter = 'AND c.revision < @before_revision ' + rows = run_store.cosmos_orchestration_runs_container.query_items( + query=( + f'SELECT TOP {EDIT_HISTORY_PAGE_SIZE + 1} * FROM c ' + 'WHERE c.conversation_id = @conversation_id AND c.user_id = @user_id ' + 'AND (c.turn_id = @turn_id OR (NOT IS_DEFINED(c.turn_id) AND c.plan.turn_id = @turn_id)) ' + 'AND (NOT IS_DEFINED(c.record_type) OR c.record_type = "run" OR c.record_type = "orchestration_run") ' + 'AND (c.revision_root_run_id = @revision_root_run_id OR c.id = @revision_root_run_id) ' + f'{cursor_filter}ORDER BY c.revision DESC' + ), + parameters=parameters, partition_key=record['conversation_id'], + ) + history_records = [ + row for row in rows + if _owned_run(row, row.get('id'), user_id, record['conversation_id']) + and _same_lineage(record, row) + and (before_revision is None or _revision(row) < before_revision) + ] + history_records.sort(key=_revision, reverse=True) + page = history_records[:EDIT_HISTORY_PAGE_SIZE] + history = [ + { + 'run_id': row['id'], 'plan_id': row['plan'].get('plan_id'), + 'revision': _revision(row), + 'created_at': row['created_at'][:64] if isinstance(row.get('created_at'), str) else None, + 'origin': ( + row['revision_origin'] if row.get('revision_origin') in ('ai', 'restore') + else 'original' if row['id'] == _root_id(row) else 'ai' + ), + 'note': row['revision_note'][:EDIT_NOTE_LIMIT] if isinstance(row.get('revision_note'), str) else '', + } + for row in page + ] + pending = (record.get('edit_pending') or {}).get('elicitation') + return { + 'plan': _public_plan(record['plan']), + 'version': _version(record), + 'edits': deepcopy(record.get('edit_narrowing') or _normalize_edits(record['plan'], None)), + 'chat': _bounded_chat(record.get('edit_chat')), + 'history': history, + 'next_before_revision': _revision(page[-1]) if len(history_records) > len(page) else None, + 'pending': { + key: deepcopy(pending[key]) for key in _QUESTION_FIELDS if key in pending + } if isinstance(pending, dict) else None, + 'busy': _claim_is_active(record.get('edit_claim')), + } + + +def _normalize_request(data, conversation_id): + _json_text(data) + if not isinstance(data, dict): + raise _invalid() + action = data.get('action') + if not isinstance(action, str) or action not in _ACTION_FIELDS: + raise _invalid('Choose a supported plan edit action.') + if set(data) - (_REQUEST_FIELDS | _ACTION_FIELDS[action]): + raise _invalid('Invalid plan edit request fields.') + if data.get('conversation_id') != conversation_id: + raise _not_found() + if not _valid_id(data.get('submission_id')): + raise _invalid('A submission ID is required.') + version = data.get('expected_version') + if not isinstance(version, str) or len(version) > 200: + raise PlanRevisionError('Reload the plan before editing it.') + result = deepcopy(data) + if action == 'ask': + instruction = data.get('instruction') + if ( + not isinstance(instruction, str) or not instruction.strip() + or len(instruction) > EDIT_INSTRUCTION_LIMIT + ): + raise _invalid('Enter a plan change of at most 2,000 characters.') + result['instruction'] = instruction.strip() + if action == 'restore' and not _valid_id(data.get('source_run_id')): + raise _invalid('Choose a saved version to restore.') + if action == 'answer' or any( + key in data for key in ('elicitation_id', 'elicitation_revision') + ): + revision = data.get('elicitation_revision') + if ( + not _valid_id(data.get('elicitation_id')) or not isinstance(revision, int) + or isinstance(revision, bool) or revision < 0 + ): + raise _invalid('Use the current planner question.') + if action == 'answer': + response = data.get('elicitation_response') + context = data.get('elicitation_context') + if ( + not isinstance(response, dict) or set(response) - {'action', 'content'} + or response.get('action') not in ('accept', 'decline', 'cancel') + or ('content' in response and not isinstance(response['content'], dict)) + or (context is not None and not isinstance(context, dict)) + ): + raise _invalid('Invalid planner question response.') + return result + + +def _submission_receipt(record, request, fingerprint): + for entry in [ + *(record.get('edit_submissions') or []), *(record.get('edit_attempts') or []), + record.get('edit_claim') or {}, + ]: + if entry.get('submission_id') != request['submission_id']: + continue + if entry.get('fingerprint') != fingerprint: + raise PlanRevisionError( + 'That submission ID was already used for a different edit.', + code='submission_conflict', + ) + if entry.get('result_run_id'): + result = read_revision_run( + entry['result_run_id'], record['user_id'], record['conversation_id'], + ) + if not _same_lineage(record, result): + raise _not_found() + return { + 'record': record, 'request': request, 'claim_id': None, + 'replayed': True, 'outcome_run_id': result['id'], + } + return None + + +def claim_plan_revision(run_id, user_id, conversation_id, data): + """Claim one bounded edit lease, or replay an already committed identical request.""" + record = read_revision_run(run_id, user_id, conversation_id) + request = _normalize_request(data, conversation_id) + fingerprint = hashlib.sha256(_json_text(request).encode('utf-8')).hexdigest() + replay = _submission_receipt(record, request, fingerprint) + if replay: + return replay + _assert_editable(record) + _check_version(record, request['expected_version'], required=True) + active_claim = record.get('edit_claim') or {} + discarding = request['action'] == 'discard' + if _claim_is_active(active_claim) and ( + not discarding or active_claim.get('submission_id') == request['submission_id'] + ): + raise _busy() + pending = record.get('edit_pending') + if request['action'] in ('ask', 'restore') and pending: + raise _busy() + if request['action'] in ('answer', 'discard'): + question = (pending or {}).get('elicitation') + # A discard can revoke an abandoned worker even before it produces a question. + if not isinstance(question, dict) and not (discarding and active_claim): + raise _changed(record) + if request['action'] == 'answer' or 'elicitation_id' in request: + if ( + not isinstance(question, dict) + or request['elicitation_id'] != question.get('elicitation_id') + or request['elicitation_revision'] != question.get('revision') + ): + raise _changed(record) + if request['action'] == 'restore': + source = read_revision_run(request['source_run_id'], user_id, conversation_id) + if not _same_lineage(record, source) or source.get('started_at'): + raise _not_found() + overlay = _normalize_edits( + record['plan'], request.get('edits') if request.get('edits') is not None + else record.get('edit_narrowing'), + ) + request['edits'] = overlay + now = _now() + lease = { + 'claim_id': str(uuid.uuid4()), 'submission_id': request['submission_id'], + 'fingerprint': fingerprint, 'started_at': now.isoformat(), + 'expires_at': (now + timedelta(seconds=EDIT_CLAIM_SECONDS)).isoformat(), + } + attempts = [ + item for item in record.get('edit_attempts') or [] + if item.get('submission_id') != request['submission_id'] + ] + attempts.append({'submission_id': request['submission_id'], 'fingerprint': fingerprint}) + try: + held = _replace(record, { + 'edit_claim': lease, 'edit_narrowing': overlay, + 'edit_attempts': attempts[-EDIT_RETAINED_SUBMISSIONS:], + }) + except PlanRevisionError: + latest = read_revision_run(run_id, user_id, conversation_id) + replay = _submission_receipt(latest, request, fingerprint) + if replay: + return replay + if _claim_is_active(latest.get('edit_claim')): + raise _busy() + raise + return {'record': held, 'request': request, 'claim_id': lease['claim_id'], 'replayed': False} + + +def _claimed_record(claim): + if not isinstance(claim, dict) or claim.get('replayed') or not claim.get('claim_id'): + raise PlanRevisionError('This edit is no longer active. Reload the plan.') + original = claim['record'] + record = read_revision_run(original['id'], original['user_id'], original['conversation_id']) + saved = record.get('edit_claim') or {} + if ( + saved.get('claim_id') != claim['claim_id'] + or saved.get('submission_id') != claim['request']['submission_id'] + or saved.get('fingerprint') != (original.get('edit_claim') or {}).get('fingerprint') + or record.get('_etag') != original.get('_etag') + or _version(record) != _version(original) + or not _claim_is_active(saved) + ): + raise _changed(record) + _assert_editable(record) + return record + + +def _completion_updates(record, result_run_id, kind): + lease = record['edit_claim'] + receipt = { + 'submission_id': lease['submission_id'], 'fingerprint': lease['fingerprint'], + 'result_run_id': result_run_id, 'kind': kind, 'completed_at': _now().isoformat(), + } + receipts = [ + item for item in record.get('edit_submissions') or [] + if item.get('submission_id') != lease['submission_id'] + ] + return { + 'edit_claim': None, + 'edit_submissions': (receipts + [receipt])[-EDIT_RETAINED_SUBMISSIONS:], + 'edit_attempts': [ + item for item in record.get('edit_attempts') or [] + if item.get('submission_id') != lease['submission_id'] + ][-EDIT_RETAINED_SUBMISSIONS:], + } + + +def _new_revision(record, document, turn_context, chat, instruction, origin): + if not isinstance(document, dict) or not isinstance(document.get('steps'), list): + raise _invalid('A validated plan is required.') + if origin not in ('ai', 'restore'): + raise _invalid('Invalid revision origin.') + identity = json.dumps([ + record['conversation_id'], record['user_id'], record['id'], + record['edit_claim']['submission_id'], + ], separators=(',', ':')) + run_id = f'run_{uuid.uuid5(uuid.NAMESPACE_URL, identity + ":run").hex}' + plan_id = f'plan_{uuid.uuid5(uuid.NAMESPACE_URL, identity + ":plan").hex}' + version = str(uuid.uuid4()) + plan = _manual_plan(_public_plan(document), version) + plan.update({ + 'plan_id': plan_id, 'run_id': run_id, 'revision': _revision(record) + 1, + 'turn_id': _turn_id(record), 'conversation_id': record['conversation_id'], + 'user_id': record['user_id'], + }) + for step in plan['steps']: + step['status'] = 'pending' + summary = summarize_plan(plan) + now = _now().isoformat() + result = { + 'id': run_id, 'run_id': run_id, 'record_type': run_store.RUN_RECORD_TYPE, + 'conversation_id': record['conversation_id'], 'user_id': record['user_id'], + 'turn_index': record.get('turn_index', 0), 'plan': plan, 'plan_summary': summary, + 'revision': plan['revision'], 'status': 'awaiting_approval', + 'created_at': now, 'updated_at': now, 'started_at': None, 'completed_at': None, + 'error': None, 'approval': deepcopy(plan['approval']), + 'capabilities_used': list(summary['capabilities_used']), + 'documents_touched': [], 'artifacts': [], 'token_usage': {}, + 'unresolved': [], 'answered_questions': [], + 'edit_version': version, 'edit_narrowing': _normalize_edits(plan, None), + 'edit_chat': _bounded_chat(chat), 'edit_pending': None, 'edit_claim': None, + 'edit_submissions': [], 'edit_attempts': [], + 'revision_root_run_id': _root_id(record), 'parent_run_id': record['id'], + 'revision_origin': origin, + 'revision_note': instruction[:EDIT_NOTE_LIMIT] if isinstance(instruction, str) else '', + } + for key in (*_IMMUTABLE_FIELDS, *_CONTEXT_FIELDS): + if key in record: + result[key] = deepcopy(record[key]) + result['turn_id'] = _turn_id(record) + for key in _CONTEXT_FIELDS: + if isinstance(turn_context, dict) and key in turn_context: + result[key] = deepcopy(turn_context[key]) + return result + + +def complete_plan_revision( + claim, *, kind, document=None, turn_context=None, chat=None, instruction='', + origin='ai', pending=None, +): + """Commit a new plan and its supersession together, or save a same-plan outcome.""" + record = _claimed_record(claim) + desired_chat = record.get('edit_chat') if chat is None else chat + if kind == 'plan': + result = _new_revision(record, document, turn_context, desired_chat, instruction, origin) + predecessor = deepcopy(run_store._strip_cosmos_metadata(record)) + predecessor.update(_completion_updates(record, result['id'], kind)) + predecessor.update({ + 'status': 'superseded', 'superseded_by_run_id': result['id'], + 'superseded_at': _now().isoformat(), 'updated_at': _now().isoformat(), + 'edit_pending': None, + }) + # The SDK's per-operation option is if_match_etag, not replace_item's etag. + operations = [ + ('replace', (record['id'], predecessor), {'if_match_etag': record['_etag']}), + ('create', (result,)), + ] + try: + responses = run_store.cosmos_orchestration_runs_container.execute_item_batch( + batch_operations=operations, partition_key=record['conversation_id'], + ) + except (exceptions.CosmosBatchOperationError, exceptions.CosmosHttpResponseError) as exc: + if exc.status_code in (404, 409, 412): + latest = read_revision_run(record['id'], record['user_id'], record['conversation_id']) + raise _changed(latest) from exc + raise + claim['completed'] = True + saved = responses[1].get('resourceBody') if responses and len(responses) > 1 else None + if isinstance(saved, dict) and saved.get('_etag'): + return saved + return read_revision_run(result['id'], result['user_id'], result['conversation_id']) + if kind not in ('elicitation', 'message', 'discard'): + raise _invalid('Invalid plan edit outcome.') + if kind == 'elicitation' and ( + not isinstance(pending, dict) or not isinstance(pending.get('elicitation'), dict) + ): + raise _invalid('A saved planner question is required.') + version = str(uuid.uuid4()) + plan = _manual_plan(record['plan'], version) + updates = { + **_completion_updates(record, record['id'], kind), + 'plan': plan, 'approval': deepcopy(plan['approval']), 'status': 'awaiting_approval', + 'edit_version': version, 'edit_chat': _bounded_chat(desired_chat), + 'edit_pending': deepcopy(pending) if kind == 'elicitation' else None, + } + if isinstance(turn_context, dict) and 'planning_token_usage' in turn_context: + updates['planning_token_usage'] = deepcopy(turn_context['planning_token_usage']) + result = _replace(record, updates) + claim['completed'] = True + return result + + +def release_plan_revision(claim): + """Best-effort cleanup of our own still-live lease, never another worker's outcome.""" + if not claim or claim.get('replayed') or claim.get('completed'): + return + try: + record = _claimed_record(claim) + _replace(record, {'edit_claim': None}) + except PlanRevisionError: + log_event( + '[ORCHESTRATION_RUNS] Kept a newer or completed plan edit during lease cleanup.', + level=logging.INFO, debug_only=True, + ) + except AzureError as exc: + log_event( + '[ORCHESTRATION_RUNS] Plan edit lease cleanup could not be saved.', + extra={'exception_type': type(exc).__name__}, level=logging.ERROR, + ) + + +def claim_plan_run( + run_id, user_id, conversation_id, *, plan_id=None, expected_version=None, + edits=None, conversation_context=None, +): + """Atomically turn a current pre-execution plan into the one executable run.""" + record = read_revision_run(run_id, user_id, conversation_id) + if record.get('started_at') or record.get('status') in ('running', 'completed'): + raise PlanRevisionError('This plan has already started.', code='already_run') + if record.get('status') not in _RUNNABLE_STATUSES or 'superseded_by_run_id' in record: + raise _changed(record) + _check_plan_id(record, plan_id) + _check_version(record, expected_version, required=bool(_version(record))) + if _claim_is_active(record.get('edit_claim')) or record.get('edit_pending'): + raise _busy() + overlay = _normalize_edits( + record['plan'], edits if edits is not None else record.get('edit_narrowing'), + ) + plan = apply_plan_edits(deepcopy(record['plan']), overlay) + now = _now().isoformat() + plan['status'] = 'running' + plan['approval'] = { + **(plan.get('approval') or {}), 'state': 'approved', + 'approved_at': now, 'approved_by': user_id, + } + summary = summarize_plan(plan) + updates = { + 'plan': plan, 'plan_summary': summary, 'approval': deepcopy(plan['approval']), + 'status': 'running', 'started_at': now, 'edit_claim': None, + 'capabilities_used': list(summary['capabilities_used']), + } + if conversation_context is not None: + if not isinstance(conversation_context, dict): + raise _invalid('Invalid conversation context.') + updates['conversation_context'] = deepcopy(conversation_context) + return _replace(record, updates) diff --git a/application/single_app/functions_orchestration_planner.py b/application/single_app/functions_orchestration_planner.py index 1ec29a205..0e13cac12 100644 --- a/application/single_app/functions_orchestration_planner.py +++ b/application/single_app/functions_orchestration_planner.py @@ -27,7 +27,7 @@ answering step rather than to an error -- a user who asked a question should get an answer even when the planning layer had a bad day. -Version: 0.261.099 +Version: 0.261.102 """ import json @@ -52,6 +52,7 @@ PlanValidationError, normalize_elicitation, normalize_plan, + plan_document_ids, ) PLANNER_MAX_TOKENS = 2000 @@ -368,8 +369,33 @@ def build_trivial_plan(user_message, planner_context=None): Only ask when you truly cannot proceed; a reasonable assumption stated in "assumptions" is better than a question.""" +PLAN_EDIT_INSTRUCTIONS = """ +You are now in the plan editor, not executing a request. No step of this plan has run. +The plan_edit object contains the current effective plan, the current task, the user's +latest change, and this plan's own editing conversation. Revise THAT plan rather than +starting a new conversation or answering the original task. + +Preserve the user's previous changes, disabled steps, source selections, and constraints +unless the latest instruction explicitly changes them. An earlier version or chat turn +does not undo the current plan. Keep existing step IDs for work that remains the same. +You may add, remove, or change work only using the offered capabilities and authorized +sources. Adding a capability to a plan cannot enable a disabled product feature. All +capability gates, limits, argument schemas, and the final respond step still apply. + +For a change, return kind "plan" with the normal plan fields AND "revised_request": a +self-contained description of the complete updated task, at most 6000 characters. This +request will guide retrieval and the final answer, so include the latest changes and +retain the relevant earlier constraints. Do not claim to have performed any planned work. + +If the user asks about the plan, or requests unavailable work, you may instead return +{"kind": "message", "message": ""}. +That keeps the current plan unchanged. Do not silently substitute a different capability +for one the user specifically requested. If necessary information is missing, return the +existing elicitation shape; the editor will ask without discarding the current plan. +""" + -def build_planner_messages(planner_context, replan_hint=None): +def build_planner_messages(planner_context, replan_hint=None, edit_context=None): """The two messages the planner sees. The context is passed as JSON rather than prose because it is data the model has to @@ -377,6 +403,8 @@ def build_planner_messages(planner_context, replan_hint=None): paraphrased document id is a plan step that fails validation. """ payload = dict(planner_context or {}) + if edit_context is not None: + payload['plan_edit'] = edit_context user_content = json.dumps(payload, ensure_ascii=False, separators=(',', ':'), default=str) @@ -389,7 +417,12 @@ def build_planner_messages(planner_context, replan_hint=None): ) return [ - {'role': 'system', 'content': PLANNER_SYSTEM_PROMPT}, + { + 'role': 'system', + 'content': PLANNER_SYSTEM_PROMPT + ( + '\n\n' + PLAN_EDIT_INSTRUCTIONS if edit_context is not None else '' + ), + }, {'role': 'user', 'content': user_content}, ] @@ -620,10 +653,12 @@ def plan_request( seeds=None, document_labels=None, request_context=None, + edit_context=None, ): """Produce a validated plan, or a question set, for one request. - Returns ``(kind, document)`` where ``kind`` is ``'plan'`` or ``'elicitation'``. + Returns ``(kind, document)`` where ``kind`` is ``'plan'`` or ``'elicitation'``, + or ``'message'`` for an editor-only explanation. ``request_context`` describes *this caller*, as opposed to the deployment: their app roles, whether their message contains a URL, whether they have an agent to invoke. It @@ -636,7 +671,8 @@ def plan_request( A planner that fails -- unreachable, unparseable, or producing something that cannot be validated -- degrades to a single answering step rather than raising. The user asked a question; an orchestration layer having a bad day is not a reason to refuse to - answer it. + answer it. An editor request is different: a failed change must preserve the prior + plan, so ``edit_context`` disables that fallback and permits an explanatory message. """ settings = settings if isinstance(settings, dict) else {} @@ -655,6 +691,14 @@ def plan_request( actions = context.get('actions') or [] def _fallback(reason): + if edit_context is not None: + log_event( + '[ORCHESTRATION_PLANNER] Could not revise the plan.', + level=logging.WARNING, extra={'reason': reason}, + ) + raise PlannerError( + 'The requested change could not be planned. Your previous plan is unchanged.' + ) log_event( f"[ORCHESTRATION_PLANNER] Falling back to a direct answer: {reason}", level=logging.WARNING, @@ -684,7 +728,9 @@ def _fallback(reason): try: reply, usage = _call_planner( - client, deployment, build_planner_messages(context, replan_hint=replan_hint) + client, deployment, build_planner_messages( + context, replan_hint=replan_hint, edit_context=edit_context, + ) ) except Exception as exc: return _fallback(f'the planner call failed: {exc}') @@ -695,6 +741,29 @@ def _fallback(reason): kind = str(parsed.get('kind') or '').strip().lower() + if edit_context is not None: + if kind == 'message': + message = parsed.get('message') + if not isinstance(message, str) or not message.strip() or len(message) > 2000: + return _fallback('the editor explanation was invalid') + return 'message', { + 'message': message.strip(), + 'token_usage': { + field: getattr(usage, field) + for field in ('prompt_tokens', 'completion_tokens', 'total_tokens') + if isinstance(getattr(usage, field, None), int) + }, + } + if kind not in ('plan', 'elicitation'): + return _fallback('the editor response did not identify a plan or question') + if kind == 'plan': + revised_request = parsed.get('revised_request') + if ( + not isinstance(revised_request, str) or not revised_request.strip() + or len(revised_request) > RESOLVED_REQUEST_MAX_LENGTH + ): + return _fallback('the revised task was missing or too large') + if kind == 'elicitation' and allow_elicitation: try: fields = (parsed.get('ui_hints') or {}).get('fields') or {} @@ -740,11 +809,16 @@ def _fallback(reason): seeds=seeds, document_labels=document_labels, request_context=request_context, + edit_context=edit_context, ) if kind == 'elicitation': return _fallback('the planner asked a question when it had already asked one') + if edit_context is not None and authorized_document_ids is not None: + if set(plan_document_ids(parsed, include_disabled=True)) - set(authorized_document_ids): + return _fallback('the revised plan named unavailable documents') + try: plan = normalize_plan( parsed, @@ -763,6 +837,9 @@ def _fallback(reason): except PlanValidationError as exc: return _fallback(f'no runnable step survived validation: {exc}') + if edit_context is not None and plan.get('validation', {}).get('errors'): + return _fallback('the revised plan contained unavailable or invalid work') + plan['revision'] = revision plan['planner_model'] = deployment if usage is not None: diff --git a/application/single_app/functions_orchestration_runs.py b/application/single_app/functions_orchestration_runs.py index a8a134d3b..3bbb702ef 100644 --- a/application/single_app/functions_orchestration_runs.py +++ b/application/single_app/functions_orchestration_runs.py @@ -21,7 +21,7 @@ Shaped and styled after ``functions_personal_workflows.py`` so the run/step CRUD reads the same as the workflow-run CRUD it sits beside. -Version: 0.261.099 +Version: 0.261.102 """ import hashlib @@ -56,7 +56,9 @@ PENDING_TURN_MAX_BYTES = 65536 RUN_RECORD_FILTER = ( '(NOT IS_DEFINED(c.record_type) OR c.record_type = "run" ' - 'OR c.record_type = "orchestration_run")' + 'OR c.record_type = "orchestration_run") ' + 'AND NOT IS_DEFINED(c.superseded_by_run_id) ' + 'AND (NOT IS_DEFINED(c.status) OR c.status != "superseded")' ) @@ -83,6 +85,13 @@ def _is_run_record(document): ) +def _is_current_run_record(document): + return ( + _is_run_record(document) and 'superseded_by_run_id' not in document + and document.get('status') != 'superseded' + ) + + def _pending_turn_id(conversation_id, user_id, turn_id): if not conversation_id or not user_id or not turn_id: raise ValueError('Conversation, user, and turn IDs are required.') @@ -207,6 +216,7 @@ def create_orchestration_run( initial_updates=None, idempotent=False, turn_context=None, + expected_previous_run=None, ): """Persist a new run record for a validated plan. @@ -214,6 +224,9 @@ def create_orchestration_run( resolved from the conversation when the caller does not supply one -- the ordering that the ledger and the map view rely on has to be assigned somewhere, and assigning it at creation keeps it monotonic without the route having to track a counter. + + ``expected_previous_run`` is the owned raw record observed before ordinary replanning. + It makes publication conditional on that unstarted plan remaining unchanged. """ plan = plan if isinstance(plan, dict) else {} conversation_id = conversation_id or plan.get('conversation_id') @@ -225,6 +238,17 @@ def create_orchestration_run( run_id = plan.get('run_id') or new_run_id() if str(run_id).startswith((PENDING_TURN_PREFIX, 'elicitation_')): raise ValueError('The plan uses a reserved run ID.') + if expected_previous_run is not None: + if ( + not _is_run_record(expected_previous_run) + or expected_previous_run.get('user_id') != user_id + or expected_previous_run.get('conversation_id') != conversation_id + or not expected_previous_run.get('_etag') + or not expected_previous_run.get('id') + ): + raise ConversationContextError('The previous plan could not be matched to this turn.') + if turn_index is None: + turn_index = expected_previous_run.get('turn_index', 0) if turn_index is None: turn_index = next_turn_index(conversation_id, user_id) @@ -279,7 +303,11 @@ def create_orchestration_run( }) try: - if idempotent: + if expected_previous_run is not None: + record = _publish_replanned_run( + record, expected_previous_run, idempotent=idempotent, + ) + elif idempotent: try: cosmos_orchestration_runs_container.create_item(body=record) except exceptions.CosmosResourceExistsError: @@ -304,6 +332,105 @@ def create_orchestration_run( return _strip_cosmos_metadata(record) +def _replanned_outcome(record, previous): + """Replay only a matching owned run, never a guessed or unrelated result ID.""" + if ( + not _is_run_record(previous) + or previous.get('user_id') != record['user_id'] + or previous.get('conversation_id') != record['conversation_id'] + or previous.get('turn_id') != record.get('turn_id') + ): + raise ConversationContextError('The saved plan could not be matched to this turn.') + if previous['id'] == record['id']: + existing = previous + elif previous.get('superseded_by_run_id') == record['id']: + existing = cosmos_orchestration_runs_container.read_item( + item=record['id'], partition_key=record['conversation_id'], + ) + if ( + existing.get('parent_run_id') != previous['id'] + or existing.get('revision_root_run_id') + != (previous.get('revision_root_run_id') or previous['id']) + ): + raise ConversationContextError('The saved plan could not be matched to this turn.') + else: + return None + if ( + not _is_run_record(existing) + or existing.get('id') != record['id'] + or existing.get('user_id') != record['user_id'] + or existing.get('conversation_id') != record['conversation_id'] + or existing.get('turn_id') != record.get('turn_id') + or (existing.get('plan') or {}).get('plan_id') != record['plan'].get('plan_id') + or existing.get('revision') != record.get('revision') + ): + raise ConversationContextError('The saved plan could not be matched to this turn.') + return existing + + +def _publish_replanned_run(record, expected_previous_run, *, idempotent=False): + """An ordinary replan must also lose to an editor hold or execution claim.""" + previous = cosmos_orchestration_runs_container.read_item( + item=expected_previous_run['id'], partition_key=record['conversation_id'], + ) + if idempotent: + replay = _replanned_outcome(record, previous) + if replay is not None: + return replay + if ( + not _is_current_run_record(previous) + or previous.get('user_id') != record['user_id'] + or previous.get('conversation_id') != record['conversation_id'] + or previous.get('turn_id') != record.get('turn_id') + or previous.get('_etag') != expected_previous_run['_etag'] + or previous.get('status') not in ('draft', 'awaiting_approval', 'approved') + or previous.get('started_at') or previous.get('edit_version') + or (previous.get('plan') or {}).get('edit_version') + or previous['id'] == record['id'] + or _coerce_int(record.get('revision')) <= _coerce_int(previous.get('revision')) + ): + raise ConversationContextError('The plan changed while planning. Reload the latest plan.') + for key in ( + 'user_message', 'user_message_id', 'user_message_fingerprint', 'turn_id', + 'original_seeds', 'conversation_context', 'snapshot', 'request_fingerprint', + ): + if key in previous: + record[key] = deepcopy(previous[key]) + record['turn_index'] = previous.get('turn_index', 0) + record['revision_root_run_id'] = previous.get('revision_root_run_id') or previous['id'] + record['parent_run_id'] = previous['id'] + record['revision_origin'] = 'ai' + record['revision_note'] = 'Plan regenerated.' + predecessor = deepcopy(_strip_cosmos_metadata(previous)) + predecessor.update({ + 'status': 'superseded', 'superseded_by_run_id': record['id'], + 'revision_root_run_id': record['revision_root_run_id'], + 'superseded_at': _utc_now_iso(), 'updated_at': _utc_now_iso(), + }) + try: + cosmos_orchestration_runs_container.execute_item_batch( + batch_operations=[ + ('replace', (previous['id'], predecessor), {'if_match_etag': previous['_etag']}), + ('create', (record,)), + ], + partition_key=record['conversation_id'], + ) + except (exceptions.CosmosBatchOperationError, exceptions.CosmosHttpResponseError) as exc: + if exc.status_code in (404, 409, 412): + if idempotent: + latest = cosmos_orchestration_runs_container.read_item( + item=previous['id'], partition_key=record['conversation_id'], + ) + replay = _replanned_outcome(record, latest) + if replay is not None: + return replay + raise ConversationContextError( + 'The plan changed while planning. Reload the latest plan.' + ) from exc + raise + return record + + def get_orchestration_run(run_id, user_id, conversation_id=None): """Fetch one run, returning ``None`` unless it exists and belongs to ``user_id``. @@ -372,7 +499,12 @@ def get_latest_turn_run(conversation_id, user_id, turn_id): ], partition_key=conversation_id, )) - return _strip_cosmos_metadata(rows[0]) if rows else None + current = [ + row for row in rows if _is_current_run_record(row) + and row.get('conversation_id') == conversation_id + and row.get('user_id') == user_id and row.get('turn_id') == turn_id + ] + return _strip_cosmos_metadata(current[0]) if current else None def update_orchestration_run(run_id, user_id, updates, conversation_id=None): @@ -451,7 +583,10 @@ def list_conversation_runs(conversation_id, user_id, limit=10): ) return [] - trimmed = [item for item in items if _is_run_record(item)][:limit] + trimmed = [ + item for item in items if _is_current_run_record(item) + and item.get('user_id') == user_id and item.get('conversation_id') == conversation_id + ][:limit] trimmed.reverse() return [_strip_cosmos_metadata(item) for item in trimmed] diff --git a/application/single_app/functions_orchestration_schema.py b/application/single_app/functions_orchestration_schema.py index 133a36b75..36a80aeb8 100644 --- a/application/single_app/functions_orchestration_schema.py +++ b/application/single_app/functions_orchestration_schema.py @@ -775,6 +775,22 @@ def validate_plan( return plan +def plan_document_ids(plan, *, include_disabled=False): + """Read named sources before authorization, without treating their presence as access.""" + document_ids = [] + for step in (plan or {}).get('steps') or (): + if not isinstance(step, dict) or ( + not include_disabled and not step.get('enabled', True) + ): + continue + arguments = step.get('arguments') + if not isinstance(arguments, dict): + continue + for field in ('document_ids', 'right_document_ids', 'left_document_id'): + document_ids.extend(_string_list(arguments.get(field))) + return list(dict.fromkeys(document_ids)) + + def build_plan_inputs(plan, seeds=None, document_labels=None, actions=None): """Describe what the plan will actually act on, for the approval card. @@ -789,7 +805,7 @@ def build_plan_inputs(plan, seeds=None, document_labels=None, actions=None): seeds = seeds if isinstance(seeds, dict) else {} labels = document_labels if isinstance(document_labels, dict) else {} - document_ids = [] + document_ids = plan_document_ids(plan) action_refs = [] uses_web = False for step in (plan or {}).get('steps') or (): @@ -802,14 +818,6 @@ def build_plan_inputs(plan, seeds=None, document_labels=None, actions=None): action_ref = arguments.get('action_ref') if action_ref and action_ref not in action_refs: action_refs.append(action_ref) - for field in ('document_ids', 'right_document_ids'): - for value in arguments.get(field) or (): - if value not in document_ids: - document_ids.append(value) - single = arguments.get('left_document_id') - if single and single not in document_ids: - document_ids.append(single) - selected = set(seeds.get('document_ids') or ()) # Named, not quoted. The plan document is stored and shown, and the prompt's full wording diff --git a/application/single_app/route_backend_orchestration.py b/application/single_app/route_backend_orchestration.py index 2b31a0dc2..0a4bb4524 100644 --- a/application/single_app/route_backend_orchestration.py +++ b/application/single_app/route_backend_orchestration.py @@ -18,7 +18,7 @@ raises rather than returning the value it would have had -- a failure that only appears once streaming is actually exercised. -Version: 0.261.099 +Version: 0.261.102 """ import hashlib @@ -52,6 +52,7 @@ HISTORY_MAX_MESSAGES, HISTORY_SCAN_LIMIT, ConversationContextError, + build_capability_request_context as _capability_request_context, build_conversation_snapshot, build_conversation_signals, build_elicitation_user_request, @@ -84,10 +85,27 @@ build_step_thought, build_synthesis_thought, build_triage_thought, + serialize_sse, ) from functions_orchestration_executor import RunContext, execute_plan +from functions_orchestration_plan_editing import ( + build_plan_edit_outcome, + revision_allowed_urls, + validate_edited_plan, +) +from functions_orchestration_plan_revisions import ( + PlanRevisionError, + begin_plan_edit, + claim_plan_revision, + claim_plan_run, + complete_plan_revision, + plan_editor_state, + read_revision_run, + release_plan_revision, +) from functions_orchestration_planner import ( ConversationResolutionError, + PlannerError, build_trivial_plan, plan_request, resolve_conversation_request, @@ -114,7 +132,6 @@ update_orchestration_run, ) from functions_orchestration_schema import ( - APPROVAL_STATE_APPROVED, COMPLEXITY_TRIVIAL, ELICITATION_ACTION_ACCEPT, ELICITATION_ACTION_CANCEL, @@ -122,7 +139,6 @@ PLAN_STATUS_COMPLETED, PLAN_STATUS_FAILED, PLAN_STATUS_RUNNING, - apply_plan_edits, normalize_plan, normalize_elicitation, summarize_plan, @@ -605,35 +621,6 @@ def _request_identity(user_id=None, seeded_agent=None): } -def _capability_request_context( - user_id, identity, user_message, agent_catalog, action_catalog=None, *, allowed_user_urls=None -): - """Describe this caller, so the capability request gates can answer for them. - - Distinct from the deployment-level gates: those answer "does this installation have - URL access at all", these answer "may this person read a URL, and is there one to - read". Without it the planner would be offered capabilities the caller cannot use -- - reading links in a message containing none, or an agent they have none of -- and - produce plans whose steps could only fail. - - Built in the route because that is where the request is. The registry stays free of - Flask, and the gates stay next to the capabilities they belong to. - """ - identity = identity or {} - urls = list(allowed_user_urls) if allowed_user_urls is not None else conversation_user_urls(user_message) - - return { - 'user_id': user_id, - 'user_message': user_message or '', - 'message_urls': urls, - 'user_roles': identity.get('user_roles') or [], - 'user_email': identity.get('user_email'), - 'user_enable_agents': identity.get('user_enable_agents', True), - 'agent_catalog': list(agent_catalog or ()), - 'action_catalog': list(action_catalog or ()), - } - - def _partition_citations(citations): """Split citations into the document, web, and tool fields chat already renders. @@ -732,7 +719,14 @@ def _elicitation_outcome_events(outcome): yield build_plan_event(outcome['document']) -def _persist_planned_turn(plan, turn_context, user_id, conversation_id, submission=None): +def _persist_planned_turn( + plan, turn_context, user_id, conversation_id, submission=None, expected_previous_run=None, +): + latest = get_latest_turn_run(conversation_id, user_id, turn_context['turn_id']) + if latest and latest['run_id'] != plan['run_id']: + expected_previous_run = expected_previous_run or read_revision_run( + latest['run_id'], user_id, conversation_id, + ) message_id, fingerprint = _save_turn_message( conversation_id, user_id, turn_context['turn_id'], turn_context['user_message'], previous=turn_context, prompt_selection=turn_context.get('prompt_selection'), @@ -741,7 +735,7 @@ def _persist_planned_turn(plan, turn_context, user_id, conversation_id, submissi turn_context['user_message_fingerprint'] = fingerprint create_orchestration_run( plan, user_id, conversation_id=conversation_id, idempotent=True, - turn_context=turn_context, + turn_context=turn_context, expected_previous_run=expected_previous_run, ) if submission: prepare_elicitation_outcome(submission, 'plan', plan, turn_context) @@ -905,6 +899,71 @@ def _run_detail_row(record): return row +def _plan_edit_identity(data, settings): + if not isinstance(data, dict): + raise PlanRevisionError('An editor request must be an object.', code='invalid_request', status_code=400) + user_id = get_current_user_id() + if not user_id: + raise PlanRevisionError('User not authenticated.', code='unauthenticated', status_code=401) + if not _orchestration_enabled(settings): + raise PlanRevisionError('Chat orchestration is not enabled.', code='disabled', status_code=403) + conversation_id = data.get('conversation_id') + if not isinstance(conversation_id, str) or not conversation_id.strip(): + raise PlanRevisionError('A conversation is required.', code='invalid_request', status_code=400) + conversation_id = conversation_id.strip() + try: + _authorize_context_conversation(conversation_id, user_id) + except ConversationContextError as exc: + raise PlanRevisionError('Plan not found.', code='not_found', status_code=404) from exc + return user_id, conversation_id + + +def _plan_edit_error(exc): + if isinstance(exc, PlanRevisionError): + payload = {'error': exc.message, 'code': exc.code} + if exc.current_run_id: + payload['current_run_id'] = exc.current_run_id + status = exc.status_code + elif isinstance(exc, ElicitationContextError): + payload = {'error': 'The answers were not valid.', 'code': 'invalid_request', 'details': [exc.message]} + if exc.field: + payload['field_errors'] = {exc.field: exc.message} + status = 400 + elif isinstance(exc, ConversationContextError): + payload = { + 'error': 'Conversation context changed or is unavailable. Create a new plan.', + 'code': 'plan_changed', + } + status = 409 + elif isinstance(exc, PlannerError): + payload = { + 'error': 'The requested change could not be planned. Your previous plan is unchanged. Please retry.', + 'code': 'unavailable', + } + status = 503 + else: + payload = { + 'error': 'The plan change could not be confirmed. Reload the plan or retry to recover its saved state.', + 'code': 'unavailable', + } + status = 503 + log_event( + '[ORCHESTRATION] Plan editor request could not be completed.', + level=logging.WARNING, extra={'error_type': type(exc).__name__, 'code': payload['code']}, + ) + return payload, status + + +def _plan_editor_event(record, user_id): + editor = plan_editor_state(record, user_id) + question = editor['pending'] + return serialize_sse({ + 'type': 'orchestration_elicitation' if question else 'orchestration_plan', + **({'elicitation': question} if question else {'plan': editor['plan']}), + 'editor': editor, 'done': True, + }) + + def register_route_backend_orchestration(bp): @bp.route("/api/v2/orchestration/plan", methods=["POST"]) @@ -1133,6 +1192,7 @@ def generate(): observed_pending = None current_revision = revision + planning_base = None if submission: snapshot = _conversation_context_for_run({ **turn_context, 'conversation_id': resolved_conversation_id, @@ -1163,6 +1223,17 @@ def generate(): if not previous: raise ConversationContextError('The last plan could not be opened. Submit a new request.') if previous: + planning_base = read_revision_run( + previous['run_id'], user_id, resolved_conversation_id, + ) + if ( + planning_base.get('edit_version') or planning_base.get('started_at') + or planning_base.get('status') not in ('draft', 'awaiting_approval', 'approved') + ): + raise PlanRevisionError( + 'Use the plan editor to revise this plan, or send a new request.', + current_run_id=planning_base.get('superseded_by_run_id') or planning_base['run_id'], + ) if previous.get('user_message') != message: raise ConversationContextError('This turn changed. Submit a new request.') snapshot = _conversation_context_for_run(previous, user_id, settings) @@ -1333,11 +1404,17 @@ def generate(): outcome = {'kind': kind, 'document': plan} if submission: outcome = prepare_elicitation_outcome(submission, kind, plan, turn_context) - _persist_planned_turn(plan, turn_context, user_id, resolved_conversation_id, submission) + _persist_planned_turn( + plan, turn_context, user_id, resolved_conversation_id, submission, + expected_previous_run=planning_base, + ) if submission: complete_elicitation_submission(submission) yield from _elicitation_outcome_events(outcome) + except PlanRevisionError as exc: + payload, _status = _plan_edit_error(exc) + yield serialize_sse(payload) except (ConversationContextError, ConversationResolutionError) as exc: log_event( '[ORCHESTRATION] Conversation context could not be used for planning.', @@ -1361,6 +1438,100 @@ def generate(): streamed.call_on_close(lambda: release_elicitation_submission(submission)) return streamed + @bp.route("/api/v2/orchestration/runs//editor", methods=["GET"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + def orchestration_plan_editor(run_id): + """Read the current revision and paged history without approving or pausing it.""" + try: + settings = get_settings() + user_id, conversation_id = _plan_edit_identity(request.args.to_dict(), settings) + before = request.args.get('before_revision') + if before is not None: + if len(before) > 10 or not before.isdecimal(): + raise PlanRevisionError('Invalid history cursor.', code='invalid_request', status_code=400) + before = int(before) + record = read_revision_run(run_id, user_id, conversation_id, follow_current=True) + return jsonify({'editor': plan_editor_state(record, user_id, before_revision=before)}) + except (PlanRevisionError, ConversationContextError, AzureError) as exc: + payload, status = _plan_edit_error(exc) + return jsonify(payload), status + + @bp.route("/api/v2/orchestration/runs//edit", methods=["POST"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + def orchestration_begin_plan_edit(run_id): + """Acquire a durable manual approval hold before editing.""" + try: + settings = get_settings() + data = request.get_json(silent=True) + user_id, conversation_id = _plan_edit_identity(data, settings) + if set(data) - {'conversation_id', 'plan_id', 'edits', 'expected_version'}: + raise PlanRevisionError('Invalid edit request fields.', code='invalid_request', status_code=400) + record = read_revision_run(run_id, user_id, conversation_id) + _conversation_context_for_run(record, user_id, settings) + record = begin_plan_edit( + run_id, user_id, conversation_id, plan_id=data.get('plan_id'), + edits=data.get('edits'), expected_version=data.get('expected_version'), + ) + return jsonify({'editor': plan_editor_state(record, user_id)}) + except (PlanRevisionError, ConversationContextError, AzureError) as exc: + payload, status = _plan_edit_error(exc) + return jsonify(payload), status + + @bp.route("/api/v2/orchestration/runs//revisions", methods=["POST"]) + @swagger_route(security=get_auth_security()) + @login_required + @user_required + def orchestration_revise_plan(run_id): + """Ask, answer a planner question, restore a version, or discard a pending change.""" + claim = None + try: + settings = get_settings() + data = request.get_json(silent=True) + user_id, conversation_id = _plan_edit_identity(data, settings) + record = read_revision_run(run_id, user_id, conversation_id) + snapshot = _conversation_context_for_run(record, user_id, settings) + claim = claim_plan_revision(run_id, user_id, conversation_id, data) + identity = _request_identity(user_id, seeded_agent=(record.get('seeds') or {}).get('agent')) + except (PlanRevisionError, ConversationContextError, ElicitationContextError, AzureError) as exc: + release_plan_revision(claim) + payload, status = _plan_edit_error(exc) + return jsonify(payload), status + + def generate_revision(): + try: + if claim.get('replayed'): + saved = read_revision_run( + claim['outcome_run_id'], user_id, conversation_id, follow_current=True, + ) + else: + yield build_planning_thought('Updating the plan without running it.') + outcome = build_plan_edit_outcome( + claim['record'], claim['request'], user_id, settings, + identity=identity, conversation_context=snapshot, + ledger=_load_ledger(conversation_id, user_id, settings), + ) + _conversation_context_for_run(record, user_id, settings) + if outcome['kind'] == 'plan': + outcome['document'] = validate_edited_plan( + outcome['document'], outcome['turn_context'], user_id, + get_settings(), identity, + ) + saved = complete_plan_revision(claim, **outcome) + yield _plan_editor_event(saved, user_id) + except (PlanRevisionError, ConversationContextError, ElicitationContextError, PlannerError, AzureError) as exc: + payload, _status = _plan_edit_error(exc) + yield serialize_sse(payload) + finally: + release_plan_revision(claim) + + streamed = _sse(generate_revision()) + streamed.call_on_close(lambda: release_plan_revision(claim)) + return streamed + @bp.route("/api/v2/orchestration/run", methods=["POST"]) @swagger_route(security=get_auth_security()) @login_required @@ -1376,6 +1547,8 @@ def orchestration_run(): return jsonify({'error': 'User not authenticated'}), 401 data = request.get_json(silent=True) or {} + if not isinstance(data, dict) or 'plan' in data: + return jsonify({'error': 'Run the saved plan by its ID, not a submitted plan.'}), 400 run_id = _text(data.get('run_id')) conversation_id = _text(data.get('conversation_id')) if not run_id: @@ -1389,20 +1562,15 @@ def orchestration_run(): plan = record.get('plan') or {} if record.get('status') in (PLAN_STATUS_RUNNING, PLAN_STATUS_COMPLETED): - return jsonify({'error': 'This plan has already been run.'}), 409 - - try: - plan = apply_plan_edits(plan, data.get('edits')) - except Exception as exc: - log_event(f"[ORCHESTRATION] Rejected plan edits: {exc}", level=logging.WARNING) - return jsonify({'error': 'The plan edits were not valid.'}), 400 + return jsonify({'error': 'This plan has already been run.', 'code': 'already_run'}), 409 conversation_id = conversation_id or _text(record.get('conversation_id')) try: snapshot = _conversation_context_for_run(record, user_id, settings) except ConversationContextError: return jsonify({ - 'error': 'Conversation context changed or is unavailable. Create a new plan.' + 'error': 'Conversation context changed or is unavailable. Create a new plan.', + 'code': 'plan_changed', }), 409 except AzureError as exc: log_event( @@ -1423,6 +1591,7 @@ def orchestration_run(): except ElicitationContextError as exc: return jsonify({ 'error': 'Some accepted answer context is no longer available.', + 'code': 'plan_changed', 'details': [exc.message], }), 409 @@ -1439,9 +1608,9 @@ def orchestration_run(): ) resolution = record.get('request_resolution') or {} context_message_ids = resolution.get('message_ids') - allowed_user_urls = conversation_user_urls( - user_message, snapshot, context_message_ids, record.get('answered_questions') - ) + allowed_user_urls = revision_allowed_urls({ + **record, 'user_message': user_message, 'conversation_context': snapshot, + }) # Captured out here, on the request thread, and closed over by the generator. A # streamed response's generator body runs after the view has returned, so reading @@ -1471,24 +1640,18 @@ def orchestration_run(): 'active_group_ids': seeds.get('active_group_ids') or [], } - def generate(): - approved_at = _now_iso() - try: - update_orchestration_run(run_id, user_id, { - 'status': PLAN_STATUS_RUNNING, - 'started_at': approved_at, - 'plan': plan, - 'plan_summary': summarize_plan(plan), - 'conversation_context': snapshot, - 'approval': {**(plan.get('approval') or {}), - 'state': APPROVAL_STATE_APPROVED, - 'approved_at': approved_at, - 'approved_by': user_id}, - }, conversation_id=conversation_id) - except Exception as exc: - log_event(f"[ORCHESTRATION] Could not mark a run started: {exc}", - level=logging.ERROR) + try: + record = claim_plan_run( + run_id, user_id, conversation_id, plan_id=data.get('plan_id'), + expected_version=data.get('expected_version'), edits=data.get('edits'), + conversation_context=snapshot, + ) + except (PlanRevisionError, AzureError) as exc: + payload, status = _plan_edit_error(exc) + return jsonify(payload), status + plan = record['plan'] + def generate(): # Progress is streamed from a worker thread rather than collected and flushed at # the end. The executor is synchronous and calls `emit` from inside its own loop, # and a generator cannot yield from a callback -- so buffering was the obvious diff --git a/application/v2_ui/src/components/chat/ElicitationCard.tsx b/application/v2_ui/src/components/chat/ElicitationCard.tsx index d04021cdb..0b1df413c 100644 --- a/application/v2_ui/src/components/chat/ElicitationCard.tsx +++ b/application/v2_ui/src/components/chat/ElicitationCard.tsx @@ -23,7 +23,7 @@ import { } from '../../lib/elicitationAnswers'; import { messageToPlainText } from '../../lib/messageText'; import type { ComposerDraft } from '../../lib/composerDraft'; -import type { Elicitation, ElicitationAction } from '../../lib/orchestration'; +import type { Elicitation, ElicitationAction, ElicitationContext, ElicitationResponse } from '../../lib/orchestration'; import type { PromptResolutionContext } from '../../lib/promptVariables'; export function ElicitationCard({ @@ -37,6 +37,46 @@ export function ElicitationCard({ selectElicitation(state, conversationId, turnId)); const draft = useOrchestrationStore((state) => selectElicitationDraft(state, conversationId, turnId)); + if (!elicitation || !draft) { + return null; + } + return ( + useOrchestrationStore.getState().updateElicitationDraft( + conversationId, turnId, elicitation.elicitation_id, elicitation.revision ?? 0, update, + )} + onAnswer={(response, context) => void answerElicitation({ + conversationId, turnId, response, context, + elicitationId: elicitation.elicitation_id, + elicitationRevision: elicitation.revision ?? 0, + })} + /> + ); +} + +/** Shared question inputs; the caller owns either the main-turn or editor-scoped continuation. */ +export function ElicitationForm({ + conversationId, + elicitation, + draft, + onDraftChange, + onAnswer, + onCancel, + cancelLabel = 'Cancel and abandon this request', + ariaLabel = 'Follow-up questions', +}: { + conversationId: string; + elicitation: Elicitation; + draft: ElicitationDraft; + onDraftChange: (update: (current: ElicitationDraft) => ElicitationDraft) => void; + onAnswer: (response: ElicitationResponse, context?: ElicitationContext) => void; + onCancel?: () => void; + cancelLabel?: string; + ariaLabel?: string; +}) { const bootstrap = useBootstrapStore((state) => state.data); const messages = useChatStore((state) => state.messages); const conversations = useChatStore((state) => state.conversations); @@ -72,24 +112,21 @@ export function ElicitationCard({ } const isLastPage = pageIndex === pages.length - 1; const canFinish = Object.keys(answer.errors).length === 0 && !answer.pendingUploads; - const updateDraft = (update: (current: ElicitationDraft) => ElicitationDraft) => - useOrchestrationStore.getState().updateElicitationDraft( - conversationId, turnId, elicitation.elicitation_id, elicitation.revision ?? 0, update, - ); + const updateDraft = onDraftChange; const changePage = (index: number) => updateDraft((current) => ({ ...current, pageIndex: index })); const send = (action: ElicitationAction) => { if (draft.submitting || (action === 'accept' && !canFinish)) { return; } - void answerElicitation({ - conversationId, - turnId, - elicitationId: elicitation.elicitation_id, - elicitationRevision: elicitation.revision ?? 0, - response: action === 'accept' ? answer.response : { action, content: {} }, - context: action === 'accept' ? answer.context : undefined, - }); + if (action === 'cancel' && onCancel) { + onCancel(); + } else { + onAnswer( + action === 'accept' ? answer.response : { action, content: {} }, + action === 'accept' ? answer.context : undefined, + ); + } }; const advance = () => { if (draft.submitting) { @@ -104,7 +141,7 @@ export function ElicitationCard({ return (
@@ -162,7 +199,7 @@ export function ElicitationCard({ variant="ghost" onClick={() => send('cancel')} disabled={draft.submitting} - aria-label="Cancel and abandon this request" + aria-label={cancelLabel} >
) : null} -
+ {held ? ( +

+ {editor?.loading ? 'Saving the manual-approval hold…' + : editor?.state || plan.edit_version ? 'Manual approval required — this plan will not run automatically.' + : 'Approval paused in this tab. Open Edit to retry saving the hold.'} +

+ ) : null} + {editor?.error ? ( +

+ {editor.error} +

+ ) : null} +
{runTimed ? ( Runs in {remainingSeconds}s ) : null} -
+
Review + {canEdit ? ( + void openOrchestrationPlanEditor({ conversationId, turnId })} + aria-label="Edit the plan" + > + + Edit + + ) : null} void approveAndRunPlan({ conversationId, turnId })} - aria-label="Approve and run the plan" + aria-label={held ? 'Run the saved plan' : 'Approve and run the plan'} > - Approve + {held ? 'Run' : 'Approve'}
diff --git a/application/v2_ui/src/components/chat/OrchestrationPlanEditor.tsx b/application/v2_ui/src/components/chat/OrchestrationPlanEditor.tsx new file mode 100644 index 000000000..394514c68 --- /dev/null +++ b/application/v2_ui/src/components/chat/OrchestrationPlanEditor.tsx @@ -0,0 +1,413 @@ +// OrchestrationPlanEditor.tsx + +import { useEffect, useId, useMemo, useRef } from 'react'; +import { createPortal } from 'react-dom'; +import { clsx } from 'clsx'; +import { Check, History, Loader2, RotateCcw, Send, Sparkles, X } from 'lucide-react'; +import { GlassButton, GlassPanel } from '../ui/primitives'; +import { ElicitationForm } from './ElicitationCard'; +import { OrchestrationRunView } from './OrchestrationRunView'; +import { useChatStore } from '../../stores/chatStore'; +import { + selectCanEditPlan, + selectEdits, + selectPlan, + selectPlanEditor, + selectPlanRunBlocked, + useOrchestrationStore, + type PlanEditorTarget, +} from '../../stores/orchestrationStore'; +import { applyPlanEdits, isPlanRunnable } from '../../lib/orchestrationPlan'; +import { MAX_PLAN_INSTRUCTION_LENGTH } from '../../lib/orchestration'; +import { + approveAndRunPlan, + loadPlanEditorHistory, + openOrchestrationPlanEditor, + previewPlanEditorRevision, + refreshOrchestrationPlanEditor, + submitPlanRevision, +} from '../../lib/orchestrationController'; + +const originLabels = { original: 'Original plan', ai: 'Planner revision', restore: 'Restored plan' }; + +function timestampLabel(value: string): string { + const date = new Date(value); + return Number.isNaN(date.getTime()) ? '' : date.toLocaleString(); +} + +/** Mounted by ChatPage, never by a message/card that streaming can replace. */ +export function OrchestrationPlanEditorHost() { + const target = useOrchestrationStore((state) => state.editorTarget); + const visibleConversationId = useOrchestrationStore((state) => state.visibleConversationId); + const activeConversationId = useChatStore((state) => state.activeConversationId); + if (!target || target.conversationId !== visibleConversationId + || target.conversationId !== activeConversationId) { + return null; + } + return ; +} + +function OrchestrationPlanEditor({ conversationId, turnId }: PlanEditorTarget) { + const session = useOrchestrationStore((state) => selectPlanEditor(state, conversationId, turnId)); + const plan = useOrchestrationStore((state) => selectPlan(state, conversationId, turnId)); + const edits = useOrchestrationStore((state) => selectEdits(state, conversationId, turnId)); + const canEdit = useOrchestrationStore((state) => selectCanEditPlan(state, conversationId, turnId)); + const runBlocked = useOrchestrationStore((state) => selectPlanRunBlocked(state, conversationId, turnId)); + const dialogRef = useRef(null); + const closeRef = useRef(null); + const id = useId(); + const target = useMemo(() => ({ conversationId, turnId }), [conversationId, turnId]); + const currentPreview = useMemo(() => plan ? applyPlanEdits(plan, edits) : null, [plan, edits]); + const close = () => useOrchestrationStore.getState().setEditorTarget(null); + + useEffect(() => { + const previous = document.activeElement instanceof HTMLElement ? document.activeElement : null; + const oldOverflow = document.body.style.overflow; + document.body.style.overflow = 'hidden'; + closeRef.current?.focus(); + const keyDown = (event: KeyboardEvent) => { + if (event.defaultPrevented) { + return; + } + if (event.key === 'Escape') { + event.preventDefault(); + event.stopPropagation(); + useOrchestrationStore.getState().setEditorTarget(null); + } else if (event.key === 'Tab') { + const focusable = Array.from(dialogRef.current?.querySelectorAll( + 'button:not([disabled]), a[href], input:not([disabled]), textarea:not([disabled]), ' + + 'select:not([disabled]), [tabindex]:not([tabindex="-1"]), [contenteditable="true"]', + ) ?? []).filter((element) => element.tabIndex >= 0 && element.getClientRects().length > 0); + const first = focusable[0]; + const last = focusable[focusable.length - 1]; + const inside = dialogRef.current?.contains(document.activeElement); + if (!inside || (event.shiftKey && document.activeElement === first) + || (!event.shiftKey && document.activeElement === last)) { + event.preventDefault(); + (event.shiftKey ? last : first)?.focus(); + } + } + }; + document.addEventListener('keydown', keyDown); + return () => { + document.removeEventListener('keydown', keyDown); + document.body.style.overflow = oldOverflow; + if (previous?.isConnected && useChatStore.getState().activeConversationId === conversationId) { + previous.focus(); + } + }; + }, [conversationId, turnId]); + + if (!session || !plan || !currentPreview) { + return null; + } + const editor = session.state; + const pending = editor?.pending; + const busy = session.loading || session.submitting || Boolean(editor?.busy); + const canRequest = canEdit && Boolean(editor) && !busy && !session.blocked; + const preview = session.previewRunId ? session.previewPlan : currentPreview; + const canRun = canEdit && !runBlocked && isPlanRunnable(currentPreview); + const setTab = (tab: 'ask' | 'history') => + useOrchestrationStore.getState().updatePlanEditor(conversationId, turnId, + (current) => ({ + ...current, tab, + ...(tab === 'ask' ? { previewRunId: null, previewPlan: null, previewLoading: false } : {}), + })); + const ask = () => { + if (canRequest && !pending && session.instruction.trim()) { + void submitPlanRevision(target, { action: 'ask', instruction: session.instruction }); + } + }; + const cancelChange = () => void submitPlanRevision(target, { + action: 'discard', + ...(pending ? { + elicitation_id: pending.elicitation_id, + elicitation_revision: pending.revision ?? 0, + } : {}), + }); + const retryLoad = () => { + if (editor) { + void refreshOrchestrationPlanEditor(target); + } else { + void openOrchestrationPlanEditor(target); + } + }; + + return createPortal( +
{ if (event.target === event.currentTarget) close(); }} + > + +
+
+

Edit orchestration plan

+

Saved revision {plan.revision}

+
+ void approveAndRunPlan(target)} + aria-label="Run saved revision" + > + + +
+
+

+ {session.loading ? 'Saving the manual-approval hold…' + : !editor ? 'Approval is paused in this tab. The server hold is not yet confirmed.' + : 'Manual approval required. Closing this editor never runs the plan.'} +

+ {busy ?
+ {session.error ? ( +

+ {session.error} +

+ ) : null} + {!canEdit ? ( +

+ This plan is no longer pending and cannot be edited or run here. +

+ ) : null} + +
+
+
+

+ {session.previewRunId + ? `History preview${preview ? ` — revision ${preview.revision}` : ''}` + : 'Current saved plan'} +

+ {session.previewRunId ? ( + void previewPlanEditorRevision(target, plan.run_id)}> + Back to current plan + + ) : null} +
+
+ {session.previewLoading ?

Loading revision…

+ : preview ? ( + <> + + {preview.validation.repairs.length ? ( +
+

Planner adjustments

+
    + {preview.validation.repairs.map((repair, index) =>
  • {repair}
    • )} +
+
+ ) : null} + + ) :

Preview unavailable. Your current plan is unchanged.

} +
+
+