Phase 6c: in-plan wait for a quick saved workflow

application/single_app/admin_settings_fields.py

@@ -211,6 +211,9 @@
     CHAT_ORCHESTRATION_MIN_WORKFLOW_INTERVAL_DEFAULT,
     CHAT_ORCHESTRATION_MIN_WORKFLOW_INTERVAL_MAX,
     CHAT_ORCHESTRATION_MIN_WORKFLOW_INTERVAL_MIN,
+    CHAT_ORCHESTRATION_WORKFLOW_RUN_WAIT_SECONDS_DEFAULT,
+    CHAT_ORCHESTRATION_WORKFLOW_RUN_WAIT_SECONDS_MAX,
+    CHAT_ORCHESTRATION_WORKFLOW_RUN_WAIT_SECONDS_MIN,
     WORKFLOW_LOOP_ITEMS_DEFAULT,
     WORKFLOW_LOOP_ITEMS_MAX,
     WORKFLOW_LOOP_ITEMS_MIN,
@@ -224,6 +227,7 @@
     validate_chat_orchestration_max_workflow_handoffs_per_day,
     validate_chat_orchestration_max_workflows_per_user,
     validate_chat_orchestration_min_workflow_interval_seconds,
+    validate_chat_orchestration_workflow_run_wait_max_seconds,
     validate_workflow_max_loop_items,
     validate_workflow_max_repeat_iterations,
     validate_workflow_min_schedule_interval_seconds,
@@ -4359,6 +4363,36 @@
                 {"key": "allow_user_workflows", "equals": True},
             ],
         },
+        {
+            "key": "enable_chat_orchestration_workflow_run_wait",
+            "type": "switch",
+            "label": "Wait For Quick Workflows In Chat",
+            "help": (
+                "When a user asks to run a saved workflow and then use what it found, such as "
+                "\"run my sales digest, then compare its totals with the Q3 report\", the plan "
+                "waits for the run and uses its result in the same answer. Only quick workflows "
+                "qualify: a personal workflow with durable execution on, at most 5 tasks, not a "
+                "structured workflow (so no For each or Repeat until loops), not a one-time "
+                "hand-off, no Microsoft 365 run-as user, no file sync, and nothing that waits for "
+                "a person, a review or indexing. A "
+                "waiting plan finishes in the background without the user's sign-in, so no plan "
+                "waits while Require WorkflowUser App Role is on, or when the plan also searches "
+                "the web, reads URLs, runs Deep Research, invokes an agent or action, or starts, "
+                "hands off or proposes another workflow. The wait is capped by Wait For A "
+                "Workflow From Chat under Limits and by the plan's own time limit; if the run is "
+                "still going when the wait ends, the answer says so and the result is posted to "
+                "the chat when it finishes. Off by default, so a plan never holds a chat turn "
+                "open for a workflow unless an administrator chooses to. Requires Chat "
+                "Orchestration, Enable Personal Workflows, Run Workflows From Chat and Use "
+                "Workflow Results In Chat."
+            ),
+            "default": False,
+            "depends_on": [
+                {"key": "enable_chat_orchestration", "equals": True},
+                {"key": "allow_user_workflows", "equals": True},
+                {"key": "enable_chat_orchestration_workflow_runs", "equals": True},
+            ],
+        },
         {
             "key": "enable_chat_orchestration_workflow_handoff",
             "type": "switch",
@@ -4585,6 +4619,30 @@
             ],
             "group": {"id": "limits", "label": "Limits", "variant": "limits"},
         },
+        {
+            "key": "chat_orchestration_workflow_run_wait_max_seconds",
+            "type": "number",
+            "label": "Wait For A Workflow From Chat (seconds)",
+            "help": (
+                "The longest a chat plan may wait for a quick saved workflow before the answer "
+                "moves on. The plan's own Run timeout also caps every wait, so a plan always "
+                "leaves time to write its answer. When a wait runs out, the answer says the run "
+                "is still going and its result is posted to the chat when it finishes. Applies "
+                "only when Wait For Quick Workflows In Chat is on. Default is 300 (five "
+                "minutes); supported range is 60 to 1,800."
+            ),
+            "default": CHAT_ORCHESTRATION_WORKFLOW_RUN_WAIT_SECONDS_DEFAULT,
+            "min": CHAT_ORCHESTRATION_WORKFLOW_RUN_WAIT_SECONDS_MIN,
+            "max": CHAT_ORCHESTRATION_WORKFLOW_RUN_WAIT_SECONDS_MAX,
+            "step": 1,
+            "depends_on": [
+                {"key": "enable_chat_orchestration", "equals": True},
+                {"key": "allow_user_workflows", "equals": True},
+                {"key": "enable_chat_orchestration_workflow_runs", "equals": True},
+                {"key": "enable_chat_orchestration_workflow_run_wait", "equals": True},
+            ],
+            "group": {"id": "limits", "label": "Limits", "variant": "limits"},
+        },
     ],
     "chat-orchestration-planner-model-section": [
         # One dropdown writes the four planner keys below as a set. They stay declared, as
@@ -9812,6 +9870,12 @@
         except WorkflowLoopLimitError as error:
             return None, error.public_message, None
 
+    if key == "chat_orchestration_workflow_run_wait_max_seconds":
+        try:
+            return validate_chat_orchestration_workflow_run_wait_max_seconds(value), None, None
+        except WorkflowLoopLimitError as error:
+            return None, error.public_message, None
+
     if key == "chat_orchestration_min_workflow_interval_seconds":
         try:
             return validate_chat_orchestration_min_workflow_interval_seconds(value), None, None

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.308"
+VERSION = "0.261.309"
 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_composition.py

@@ -38,8 +38,8 @@
 )
 from functions_orchestration_memory import OrchestrationMemoryError
 from functions_orchestration_registry import (
-    CAPABILITY_GENERATE_IMAGE, CAPABILITY_WORKFLOW_RESULTS, KNOWLEDGE_BASIS_GENERAL, KNOWLEDGE_BASIS_MIXED,
-    KNOWLEDGE_BASIS_SOURCES, VISUAL_CHART, VISUAL_DIAGRAM, VISUAL_IMAGE_PROPOSAL,
+    CAPABILITY_GENERATE_IMAGE, CAPABILITY_WORKFLOW_RESULTS, CAPABILITY_WORKFLOW_RUN, KNOWLEDGE_BASIS_GENERAL,
+    KNOWLEDGE_BASIS_MIXED, KNOWLEDGE_BASIS_SOURCES, VISUAL_CHART, VISUAL_DIAGRAM, VISUAL_IMAGE_PROPOSAL,
 )
 from functions_generated_export_registry import PREPARED_SLIDE_DECK_VERSION
 from functions_orchestration_result_contracts import (
@@ -377,10 +377,13 @@
             for name, reader in readers.items() if name not in image_names
         }
         # A saved workflow's result reaches the model only as fenced, untrusted notes, read again
-        # for this call; the retained value it replaces never does.
+        # for this call; the retained value it replaces never does. That includes a run the plan
+        # waited for, which only a setting-gated plan can have compose read.
         workflow_results_policy = None
         if any(
-            getattr(getattr(reader.reference, 'producer', None), 'capability_id', None) == CAPABILITY_WORKFLOW_RESULTS
+            getattr(getattr(reader.reference, 'producer', None), 'capability_id', None) in (
+                CAPABILITY_WORKFLOW_RESULTS, CAPABILITY_WORKFLOW_RUN,
+            )
             for reader in readers.values()
         ):
             # The workflow result reader loads only when a plan read a saved result.

application/single_app/functions_orchestration_continuation.py

@@ -4,7 +4,8 @@
 Version: 0.261.139
 Initialized callers import this module after application bootstrap. It neither
 admits plans nor creates retry attempts, and never starts or cancels native jobs.
-Valid native waits use the existing recovery claim and native restore engine.
+Valid native waits use the existing recovery claim and native restore engine, and so,
+since 0.261.309, does a step waiting on a quick saved-workflow run.
 Other scheduler states share its stable producer token and fresh claim_id fences.
 Saved native and generic result waits retain their original core dispatch data.
 A run from the removed legacy plan contract is refused with the legacy-plan failure;
@@ -16,7 +17,7 @@
 import uuid
 from copy import deepcopy
 from datetime import datetime, timezone
-from types import SimpleNamespace
+from types import MappingProxyType, SimpleNamespace
 
 from azure.core import MatchConditions
 from azure.core.exceptions import AzureError
@@ -55,13 +56,21 @@
 from functions_orchestration_schema import (
     LEGACY_PLAN_CODE, PLAN_HARD_MAX_STEPS, build_failure, build_step_result, is_legacy_plan, safe_failure,
 )
+from functions_orchestration_workflow_run_wait import SAVED_WORKFLOW_RUN_WAIT_KIND
 
 
 CONTINUATION_VERSION = "orchestration-continuation-v1"
 _MODES = frozenset({"execute", "outputs", "delivery"})
 _TERMINAL = frozenset({"completed", "failed", "cancelled"})
 _COMPLETE = frozenset({"completed", "partial"})
 _TRANSIENT = (AzureError, TimeoutError, ConnectionError)
+# The waits the scheduler claims a continuation for, by wait kind, with the capability whose step
+# may declare it. 6c adds the quick saved-workflow run; a native compute waits as before.
+_SCHEDULED_WAIT_KINDS = MappingProxyType({
+    "native_tabular_compute": "tabular_analyze",
+    SAVED_WORKFLOW_RUN_WAIT_KIND: "workflow_run",
+})
+_SCHEDULED_WAIT_CAPABILITIES = frozenset(_SCHEDULED_WAIT_KINDS.values())
 
 
 def _now():
@@ -125,7 +134,8 @@ def _lease_live(record):
     return _timestamp(lease.get("expires_at")) > _now()
 
 
-def _native_wait_claimable(record, authorize):
+def _scheduled_wait_claimable(record, authorize):
+    """Whether a required step waits on work the scheduler checks back on: a native compute or, 6c, a quick run."""
     if (
         record.get("cancellation_requested_at") or not record.get("execution_binding")
         or _timestamp(record["execution_deadline_at"]) <= _now()
@@ -137,13 +147,14 @@ def _native_wait_claimable(record, authorize):
     for row in record.get("execution_steps") or []:
         step = planned.get(row.get("step_id")) or {}
         if (
-            row.get("status") != "waiting" or step.get("capability_id") != "tabular_analyze"
+            row.get("status") != "waiting" or step.get("capability_id") not in _SCHEDULED_WAIT_CAPABILITIES
             or step.get("optional") or not store.has_manifest(row["step_id"], waiting=True)
             or store.has_manifest(row["step_id"])
         ):
             continue
         payload = store.load(row["step_id"], waiting=True)
-        if (payload.get("result", {}).get("wait") or {}).get("kind") == "native_tabular_compute":
+        kind = (payload.get("result", {}).get("wait") or {}).get("kind")
+        if _SCHEDULED_WAIT_KINDS.get(kind) == step["capability_id"]:
             return True
     return False
 
@@ -195,7 +206,7 @@ def claim_run_continuation(
         return None
     if mode == "delivery" and status in _TERMINAL and record.get("finalization_status") != "pending":
         return None
-    if mode == "execute" and _native_wait_claimable(record, authorize):
+    if mode == "execute" and _scheduled_wait_claimable(record, authorize):
         request = {
             "conversation_id": conversation_id,
             "submission_id": f"scheduler_{fingerprint([run_id, record['recovery_version']])[:40]}",

application/single_app/functions_orchestration_execution.py

@@ -1261,6 +1261,10 @@ def _reads_workflow_results(self):
             for step in self.record["plan"].get("steps") or []
         )
 
+    def _waits_on_workflow_runs(self):
+        # Only a plan made with the 6c wait setting on carries this marker.
+        return bool(self.record["plan"].get("workflow_run_waits"))
+
     def _mentions_workflow_handoff(self):
         plan = self.record["plan"]
         return bool(plan.get("workflow_handoff_notes")) or any(
@@ -1269,7 +1273,7 @@ def _mentions_workflow_handoff(self):
 
     def _workflow_result_lineage(self, record_state):
         """The workflow result contexts the answer read, re-authorized now; [] when it read none."""
-        if not self._reads_workflow_results():
+        if not self._reads_workflow_results() and not self._waits_on_workflow_runs():
             return []
         # Imported here, like every caller of the results step's module.
         from functions_orchestration_workflow_results import (
@@ -1403,12 +1407,18 @@ def _finalize(self, result, error, *, refreshes=0):
             )
             if notes:
                 content.append(notes)
+        # Only a reply that carries the composed answer names the results it read; any other reply
+        # keeps just the fixed lines about results that were not read. The same flag tells a waited
+        # run step's note whether its result reached this answer, so "used in this answer" is never
+        # said when the reply itself was not written from the plan's steps.
+        workflow_result_reads = error is None and bool(prepared)
         # 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",
+            composed=workflow_result_reads,
         )
         if workflow_runs:
             content.append(workflow_runs)
@@ -1421,9 +1431,6 @@ def _finalize(self, result, error, *, refreshes=0):
             workflow_handoff = workflow_handoff_note(self.record["plan"], current.get("execution_steps") or [])
             if workflow_handoff:
                 content.append(workflow_handoff)
-        # Only a reply that carries the composed answer names the results it read; any other reply
-        # keeps just the fixed lines about results that were not read.
-        workflow_result_reads = error is None and bool(prepared)
         if (
             (self._reads_workflow_results() or self.record["plan"].get("workflow_results_notes"))
             and (status not in {"waiting", "cancelled"} or workflow_result_reads)

application/single_app/functions_orchestration_executor.py

@@ -109,6 +109,7 @@
 from functions_orchestration_timing import (
     initial_execution_deadline, positive_setting_int as _setting_int,
 )
+from functions_orchestration_workflow_run_wait import SAVED_WORKFLOW_RUN_WAIT_KIND, plan_runs_headless
 
 _LOG_PREFIX = '[ORCHESTRATION_EXECUTOR]'
 
@@ -1236,6 +1237,13 @@ def resume_waiting_dependency_step(
             step, context, settings=settings, user_id=user_id, cancel_requested=cancel_requested,
             saved_result=saved_result,
         )
+    if wait.get('kind') == SAVED_WORKFLOW_RUN_WAIT_KIND:
+        from functions_orchestration_workflow_runs import resume_workflow_run_wait
+
+        return resume_workflow_run_wait(
+            step, context, settings=settings, user_id=user_id, input_fingerprint=input_fingerprint,
+            cancel_requested=cancel_requested, saved_result=saved_result,
+        )
     if (
         context.plan_contract_version != 2 or user_id != context.user_id
         or step.get('role') not in ('gather', 'reason') or step.get('enabled') is not True
@@ -1299,6 +1307,12 @@ def _execute_dependency_plan(
         seeds=getattr(context, 'seeds', None),
     )
     steps = plan['steps']
+    # The server-computed run-step waits, and what a waited run step re-checks when it starts.
+    context.workflow_run_waits = deepcopy(plan.get('workflow_run_waits') or {})
+    context.plan_steps = deepcopy(steps) if context.workflow_run_waits else []
+    context.workflow_run_headless = bool(context.workflow_run_waits) and plan_runs_headless(
+        steps, settings, _dependency_request_context(context), export_catalog=context.export_catalog,
+    )
     resolver = get_adapter or _dependency_adapter
     cancel_probe = _make_cancel_probe(cancel_requested)
     planned_ids = _collect_plan_document_ids([step for step in steps if step['enabled']])

application/single_app/functions_orchestration_plan_editing.py

@@ -56,6 +56,7 @@
     validate_plan_requirements,
 )
 from functions_orchestration_workflow_context import workflow_planning_configured, workflow_planning_option
+from functions_orchestration_workflow_run_wait import with_headless_capability_ids
 
 TURN_CONTEXT_FIELDS = (
     'conversation_id', 'turn_id', 'user_message', 'user_message_id',
@@ -391,17 +392,22 @@ def validate_edited_plan(
         settings, allowed_ids=settings.get('chat_orchestration_enabled_capabilities'),
         request_context=caller, contract_version=contract_version, export_catalog=admitted_catalog,
     )
+    available_ids = [item['id'] for item in capabilities]
+    # A revised plan waits for a workflow run only when the server would let a new plan wait.
+    workflow_planning = with_headless_capability_ids(
+        caller.get('workflow_planning'), settings, caller, available_ids, export_catalog=admitted_catalog,
+    )
     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],
+            available_capability_ids=available_ids,
             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,
             contract_version=contract_version, existing_results=existing_results,
             composition_profiles=composition_profiles, export_catalog=admitted_catalog,
-            **workflow_planning_option(caller.get('workflow_planning')),
+            **workflow_planning_option(workflow_planning),
         )
         validate_plan_document_source_kinds(checked, {
             item['document_id']: item['source_kind']