Phase 5: orchestration runs existing workflows (workflow_run, start-and-link)

application/single_app/admin_settings_fields.py

@@ -3779,6 +3779,26 @@
                 {"key": "allow_user_workflows", "equals": True},
             ],
         },
+        {
+            "key": "enable_chat_orchestration_workflow_runs",
+            "type": "switch",
+            "label": "Run Workflows From Chat",
+            "help": (
+                "When a user asks for it, such as \"run my weekly digest now\", a plan can start "
+                "one of the user's saved personal workflows from the user's own private "
+                "conversation. A plan that starts a workflow always waits for the user to run "
+                "it, whatever the approval mode, and approving it starts each named workflow "
+                "once. The answer links to the run; results appear where the workflow already "
+                "puts them. Only workflows with durable execution on can be started this way. "
+                "Off by default, because a run does whatever its workflow is set up to do, with "
+                "the user's access. Requires Chat Orchestration and Enable Personal Workflows."
+            ),
+            "default": False,
+            "depends_on": [
+                {"key": "enable_chat_orchestration", "equals": True},
+                {"key": "allow_user_workflows", "equals": True},
+            ],
+        },
         {
             "key": "chat_orchestration_enabled_capabilities",
             "type": "checkbox_set",
@@ -3790,6 +3810,7 @@
                 "the content of a file, and Create a file to deliver a downloadable file. "
                 "Generate images also requires Image Generation. Use an action also "
                 "requires Enable Action Access. Propose workflows also requires Propose "
+                "Workflows From Chat and personal workflows. Run workflows also requires Run "
                 "Workflows From Chat and personal workflows."
             ),
             "default": [],
@@ -3807,6 +3828,7 @@
                 {"value": "render_file", "label": "Create a file"},
                 {"value": "generate_image", "label": "Generate images"},
                 {"value": "workflow_propose", "label": "Propose workflows"},
+                {"value": "workflow_run", "label": "Run workflows"},
             ],
             "depends_on": {"key": "enable_chat_orchestration", "equals": True},
         },

application/single_app/config.py

@@ -100,7 +100,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.210"
+VERSION = "0.261.212"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_orchestration_adapters.py

@@ -91,6 +91,7 @@
     CAPABILITY_URL_FETCH,
     CAPABILITY_WEB_SEARCH,
     CAPABILITY_WORKFLOW_PROPOSE,
+    CAPABILITY_WORKFLOW_RUN,
     DEPENDENCY_PLAN_CONTRACT_VERSION,
     DOCUMENT_ACTION_TYPE_COMPARISON,
     VISUAL_CHART,
@@ -1855,6 +1856,19 @@ def run_workflow_propose(step, context, *, settings, user_id, emit=None, cancel_
     )
 
 
+def run_workflow_run(step, context, *, settings, user_id, emit=None, cancel_requested=None):
+    """Start one of the user's saved workflows and link to its run.
+
+    The workflow runtime loads only when a plan starts a workflow, so resolving this adapter
+    imports nothing beyond this module.
+    """
+    from functions_orchestration_workflow_runs import adapter_workflow_run
+
+    return adapter_workflow_run(
+        step, context, settings=settings, user_id=user_id, emit=emit, cancel_requested=cancel_requested,
+    )
+
+
 # Keyed by capability id, which is also each descriptor's declared ``adapter`` name, so the
 # executor can look an adapter up straight from the step's capability without a second map.
 ADAPTER_REGISTRY = {
@@ -1868,6 +1882,7 @@ def run_workflow_propose(step, context, *, settings, user_id, emit=None, cancel_
     CAPABILITY_AGENT_INVOKE: run_agent_invoke,
     CAPABILITY_ACTION_INVOKE: run_action_invoke,
     CAPABILITY_WORKFLOW_PROPOSE: run_workflow_propose,
+    CAPABILITY_WORKFLOW_RUN: run_workflow_run,
 }
 
 

application/single_app/functions_orchestration_deliverables.py

@@ -42,12 +42,14 @@
     CAPABILITY_GENERATE_IMAGE,
     CAPABILITY_RENDER_FILE,
     CAPABILITY_WORKFLOW_PROPOSE,
+    CAPABILITY_WORKFLOW_RUN,
     MAX_GENERATED_IMAGES_PER_PLAN,
     VISUAL_CHART,
     VISUAL_DIAGRAM,
     VISUAL_IMAGE_PROPOSAL,
     VISUAL_KINDS,
     WORKFLOW_PROPOSALS_SETTING,
+    WORKFLOW_RUNS_SETTING,
     resolve_admitted_export_catalog,
 )
 from functions_orchestration_result_contracts import (
@@ -109,6 +111,18 @@
     ),
     'workflow_one_per_request': 'One request can propose one workflow. Ask for the next one in a new message.',
 }
+# Why a saved workflow cannot be started from this request. Starting a workflow is not a
+# deliverable, so these reach the planner as a fact, and only while an administrator has turned
+# starting workflows from chat on.
+WORKFLOW_RUN_UNAVAILABLE_REASONS = {
+    'capability_not_enabled_for_orchestration': 'Starting saved workflows is not enabled for chat orchestration.',
+    'workflow_runs_disabled': 'Starting saved workflows from chat is turned off for this deployment.',
+    'workflow_role_required': 'Your account does not have access to personal workflows.',
+    'workflow_shared_conversation': (
+        'Saved workflows can be started only from your own conversations, not from shared ones.'
+    ),
+    'workflow_context_unavailable': 'Your saved workflows could not be loaded for this request. Try again later.',
+}
 _KIND_LABELS = {
     KIND_ANSWER: 'answer', KIND_FILE: 'file', KIND_IMAGE: 'image',
     KIND_CHART: 'chart', KIND_DIAGRAM: 'diagram', KIND_WORKFLOW: 'workflow',
@@ -186,6 +200,17 @@
     return _unavailable('workflow_proposals_disabled')
 
 
+def _workflow_run_reason(unavailable):
+    """The closed reason this caller's plan cannot start a saved workflow."""
+    reason = unavailable.get(CAPABILITY_WORKFLOW_RUN)
+    if reason == 'not_enabled_for_orchestration':
+        return 'capability_not_enabled_for_orchestration'
+    if isinstance(reason, str) and reason in WORKFLOW_RUN_UNAVAILABLE_REASONS:
+        return reason
+    # A settings gate, or a condition this module has no text for, reads as turned off.
+    return 'workflow_runs_disabled'
+
+
 def _image_options(capabilities):
     """The exact size, quality, and background values the configured image model accepts."""
     for capability in capabilities or ():
@@ -333,6 +358,27 @@
                 'depends_on and no input bindings) and that delivers the workflow deliverable. When '
                 'the request also wants a result now, answer it once with the usual steps as well.'
             )})
+    # Starting a saved workflow is described only while an administrator has turned it on;
+    # otherwise the planner is told exactly what it was before.
+    if settings.get(WORKFLOW_RUNS_SETTING) is True:
+        if CAPABILITY_WORKFLOW_RUN in available:
+            facts.append(
+                'workflow_run starts one of the user\'s saved workflows after the user approves the plan; '
+                'a plan that starts one always waits for approval. The server adds to the reply whether '
+                'each workflow started, so never say that a workflow started, ran or finished, and never '
+                'promise its results in this conversation.'
+            )
+            recipes.append({'for': 'A saved workflow the user asks to run now', 'steps': (
+                'one workflow_run step per workflow, naming its durable catalog handle, with no depends_on, '
+                'no inputs and no deliverable. When the request only starts workflows, declare no '
+                'deliverables and no final_response; otherwise answer the rest as usual.'
+            )})
+        else:
+            facts.append(
+                'Starting saved workflows is unavailable for this request. '
+                f'{WORKFLOW_RUN_UNAVAILABLE_REASONS[_workflow_run_reason(unavailable)]} '
+                'When the user asks to run a workflow, say why in the answer.'
+            )
     truth.update({'unavailable_reasons': unavailable_reasons, 'facts': facts, 'recipes': recipes})
     return truth
 

application/single_app/functions_orchestration_execution.py

@@ -169,6 +169,7 @@
     summarize_plan,
 )
 from functions_orchestration_workflow_context import (
+    workflow_planning_configured,
     workflow_planning_option,
     workflow_proposals_configured,
     workflow_run_options,
@@ -699,7 +700,9 @@
             raise
         self._preparation_stage = "capabilities"
         workflow_configured = workflow_proposals_configured(self.settings)
-        workflow_planning = self.record.get("workflow_planning") if workflow_configured else None
+        workflow_planning = (
+            self.record.get("workflow_planning") if workflow_planning_configured(self.settings) else None
+        )
         request_context = build_capability_request_context(
             user_id, identity, user_message, agents, actions, allowed_user_urls=allowed_urls,
             **workflow_planning_option(workflow_planning),
@@ -788,6 +791,8 @@
                 "user_id": user_id, "active_group_ids": seeds.get("active_group_ids") or [],
             },
             agent_execution_identity=principal, plan_contract_version=2,
+            attempt_root_run_id=self.record.get("attempt_root_run_id") or self.record["id"],
+            signed_in_session=getattr(principal, "bridge", None) is not None,
             **workflow_run_options(
                 workflow_planning, self.record.get("time_zone") if workflow_configured else None,
             ),
@@ -1008,6 +1013,8 @@
             },
             user_roles=identity.get("user_roles") or [], user_email=principal.email,
             user_enable_agents=False, agent_execution_identity=principal,
+            attempt_root_run_id=self.record.get("attempt_root_run_id") or self.record["id"],
+            signed_in_session=getattr(principal, "bridge", None) is not None,
             plan_contract_version=2, result_service=self.services.results,
             task_results={name: TaskResult.from_dict(value) for name, value in tasks.items()},
             execution_deadline_at=self.record["execution_deadline_at"],
@@ -1284,6 +1291,15 @@
             )
             if notes:
                 content.append(notes)
+        # Imported here, like every caller of the run step's module. Deterministic, model-free:
+        # the saved workflows the plan started, even when it stopped, failed or waits afterwards.
+        from functions_orchestration_workflow_runs import workflow_run_note
+
+        workflow_runs = workflow_run_note(
+            self.record["plan"], current.get("execution_steps") or [], stopped=status == "cancelled",
+        )
+        if workflow_runs:
+            content.append(workflow_runs)
         if status == "waiting":
             content.append(build_failure("result_not_ready")["message"])
         elif status in {"failed", "cancelled"}:

application/single_app/functions_orchestration_executor.py

@@ -63,8 +63,8 @@
 )
 from functions_orchestration_deliverables import explicit_image_shortfalls
 from functions_orchestration_registry import (
-    CAPABILITY_TABULAR_ANALYZE, CAPABILITY_WORKFLOW_PROPOSE, DEPENDENCY_PLAN_CONTRACT_VERSION,
-    admitted_export_pairs, get_capability,
+    CAPABILITY_TABULAR_ANALYZE, CAPABILITY_WORKFLOW_PROPOSE, CAPABILITY_WORKFLOW_RUN,
+    DEPENDENCY_PLAN_CONTRACT_VERSION, admitted_export_pairs, external_effect_capability_ids, get_capability,
     resolve_available_capability_ids,
 )
 from functions_orchestration_result_contracts import (
@@ -220,6 +220,9 @@
 _DEFAULT_MAX_REPLANS = 2
 # Stops that bound time rather than express intent; a step that already finished keeps its result.
 _BUDGET_STOP_REASONS = frozenset({'step_timeout', 'run_timeout'})
+# Capabilities whose steps may act outside the plan. One that is interrupted, fails or is stopped
+# may already have acted, so its step is effects_uncertain. The registry owns the list.
+_EFFECT_CAPABILITIES = external_effect_capability_ids()
 
 
 def _now_iso():
@@ -360,13 +363,23 @@
         export_catalog=None,
         workflow_planning=None,
         time_zone=None,
+        attempt_root_run_id=None,
+        signed_in_session=False,
     ):
         self.run_id = run_id
         self.plan_id = plan_id
         self.conversation_id = conversation_id
         self.user_id = user_id
         self.turn_index = turn_index
         self.attempt_index = attempt_index
+        # The first attempt's run id, shared by every retry of the plan. A step that starts a saved
+        # workflow derives its request id from it, so a retry finds the run instead of starting one.
+        self.attempt_root_run_id = (
+            attempt_root_run_id if isinstance(attempt_root_run_id, str) and attempt_root_run_id else run_id
+        )
+        # Whether the user's signed-in request is running this attempt. A saved workflow is started
+        # only then, never while a run continues in the background.
+        self.signed_in_session = signed_in_session is True
         if type(plan_contract_version) is not int or plan_contract_version != DEPENDENCY_PLAN_CONTRACT_VERSION:
             raise ResultContractError('result_version_unsupported')
         self.plan_contract_version = plan_contract_version
@@ -574,7 +587,7 @@
         'checkpoint_available': False,
         'reused': False,
         'reused_from_run_id': None,
-        'effects_uncertain': status == STEP_STATUS_RUNNING and step.get('capability_id') in ('agent_invoke', 'action_invoke'),
+        'effects_uncertain': status == STEP_STATUS_RUNNING and step.get('capability_id') in _EFFECT_CAPABILITIES,
     }
     if result.get('saved_analyses'):
         record['saved_analyses'] = deepcopy(result['saved_analyses'])
@@ -593,6 +606,9 @@
     if step.get('capability_id') == CAPABILITY_WORKFLOW_PROPOSE and isinstance(result.get('workflow_proposal'), dict):
         # Server-only: the proposal card and its accept route read it; public_step_record drops it.
         record['workflow_proposal'] = deepcopy(result['workflow_proposal'])
+    if step.get('capability_id') == CAPABILITY_WORKFLOW_RUN and isinstance(result.get('workflow_run'), dict):
+        # Server-only: the workflow and run ids the run link reads; public_step_record drops it.
+        record['workflow_run'] = deepcopy(result['workflow_run'])
     return record
 
 
@@ -624,6 +640,32 @@
     return {**result, 'workflow_proposal': sidecar} if isinstance(sidecar, dict) else result
 
 
+def _restore_workflow_run(step, context, status, result, *, user_id):
+    """Describe again the run a completed workflow_run step started, when its result came back without it.
+
+    A retained, recovered or reused result is rebuilt from the task result alone, which carries
+    the workflow's name and start status but never its ids.
+    """
+    if (
+        step.get('capability_id') != CAPABILITY_WORKFLOW_RUN or status != STEP_STATUS_COMPLETED
+        or not isinstance(result, dict) or result.get('task_result') is None
+        or isinstance(result.get('workflow_run'), dict)
+    ):
+        return result
+    try:
+        from functions_orchestration_workflow_runs import rebuild_workflow_run
+
+        sidecar = rebuild_workflow_run(step, context, user_id=user_id, task=result['task_result'])
+    except Exception as exc:
+        log_event(
+            f'{_LOG_PREFIX} A recovered workflow run could not be described.',
+            level=logging.WARNING,
+            extra={**_step_log_context(context, step), 'error_type': type(exc).__name__},
+        )
+        return result
+    return {**result, 'workflow_run': sidecar} if isinstance(sidecar, dict) else result
+
+
 def _persist(persist, record_type, record):
     if not callable(persist):
         return
@@ -1470,6 +1512,7 @@
         if step['role'] == 'render' and status == STEP_STATUS_COMPLETED:
             context.artifacts.extend(deepcopy(result['artifacts']))
         result = _restore_workflow_proposal(step, context, status, result, settings=settings, user_id=user_id)
+        result = _restore_workflow_run(step, context, status, result, user_id=user_id)
         record = _step_record(context, step, index, status, result, started_at, completed_at, elapsed)
         record.update({
             'checkpoint_available': checkpoints is not None
@@ -1478,7 +1521,7 @@
             'token_usage': {} if reused else deepcopy(context.step_token_usage),
             'reused': bool(reused),
             'reused_from_run_id': (reused.get('provenance') or {}).get('run_id') if reused else None,
-            'effects_uncertain': step['capability_id'] in ('agent_invoke', 'action_invoke')
+            'effects_uncertain': step['capability_id'] in _EFFECT_CAPABILITIES
             and status in (STEP_STATUS_FAILED, STEP_STATUS_CANCELLED),
         })
         save(record)

application/single_app/functions_orchestration_plan_editing.py

@@ -55,7 +55,7 @@
     validate_plan_document_source_kinds,
     validate_plan_requirements,
 )
-from functions_orchestration_workflow_context import workflow_planning_option, workflow_proposals_configured
+from functions_orchestration_workflow_context import workflow_planning_configured, workflow_planning_option
 
 TURN_CONTEXT_FIELDS = (
     'conversation_id', 'turn_id', 'user_message', 'user_message_id',
@@ -330,7 +330,7 @@ def _revision_catalogs(
         user_id, identity, context.get('resolved_message') or context['user_message'],
         agents, actions, allowed_user_urls=revision_allowed_urls(context),
         **workflow_planning_option(
-            context.get('workflow_planning') if workflow_proposals_configured(settings) else None,
+            context.get('workflow_planning') if workflow_planning_configured(settings) else None,
         ),
         **runtime_options,
     )

application/single_app/functions_orchestration_plan_revisions.py

@@ -41,7 +41,7 @@
 from functions_orchestration_result_contracts import InputBinding, ResultContractError, ResultRef
 from functions_orchestration_schema import (
     LEGACY_PLAN_CODE, LEGACY_PLAN_MESSAGE, PlanValidationError, apply_plan_edits, is_legacy_plan,
-    plan_contract_version, summarize_plan,
+    plan_approval_floor, plan_contract_version, summarize_plan,
 )
 from functions_orchestration_timing import initial_execution_deadline
 
@@ -963,6 +963,27 @@ def release_plan_revision(claim):
         )
 
 
+def _require_approval_floor(record, plan):
+    """Refuse to start a plan whose enabled steps set an approval floor it no longer carries.
+
+    ``normalize_plan`` saves such a plan as manual, and the run record copies that mode. A saved
+    plan or run record that reads otherwise was changed after planning, so it is refused rather
+    than run on its word.
+    """
+    floor = plan_approval_floor(plan)
+    if floor is None:
+        return
+    if any(
+        not isinstance(approval, dict) or approval.get('mode') != floor['mode']
+        for approval in (record.get('approval'), plan.get('approval'))
+    ):
+        raise PlanRevisionError(
+            'A plan that starts a saved workflow runs only when you run it yourself. This saved plan '
+            "was changed, so it can't run. Ask again for a new plan.",
+            code='approval_floor_required',
+        )
+
+
 def claim_plan_run(
     run_id, user_id, conversation_id, *, plan_id=None, expected_version=None,
     edits=None, conversation_context=None, result_alias_resolver=None,
@@ -1000,6 +1021,7 @@ def claim_plan_run(
         contract_version=contract_version, export_catalog=admitted_catalog,
         composition_profiles=composition_profiles,
     )
+    _require_approval_floor(record, plan)
     started_at = _now()
     now = started_at.isoformat()
     plan['status'] = 'running'