Read saved workflow results from chat orchestration (workflow_results)

application/single_app/admin_settings_fields.py

@@ -3811,7 +3811,8 @@
                 "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."
+                "Workflows From Chat and personal workflows. Read workflow results also "
+                "requires Use Workflow Results In Chat and personal workflows."
             ),
             "default": [],
             "options": [
@@ -3829,6 +3830,7 @@
                 {"value": "generate_image", "label": "Generate images"},
                 {"value": "workflow_propose", "label": "Propose workflows"},
                 {"value": "workflow_run", "label": "Run workflows"},
+                {"value": "workflow_results", "label": "Read workflow results"},
             ],
             "depends_on": {"key": "enable_chat_orchestration", "equals": True},
         },
@@ -4066,6 +4068,8 @@
             "help": (
                 "Users can ask chat about the stored result of one of their own "
                 "finished workflow runs, from the run history or a workflow alert. "
+                "With Chat Orchestration on, a plan can also read such a result when "
+                "the user asks about it, such as \"what did my Monday digest say?\". "
                 "The workflow is not re-run, and the answer is only available in "
                 "the user's private chats. Answers stop showing once the run is "
                 "deleted or the user loses access to its sources."

application/single_app/config.py

@@ -100,7 +100,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.216"
+VERSION = "0.261.217"
 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_RESULTS,
     CAPABILITY_WORKFLOW_RUN,
     DEPENDENCY_PLAN_CONTRACT_VERSION,
     DOCUMENT_ACTION_TYPE_COMPARISON,
@@ -1869,6 +1870,19 @@ def run_workflow_run(step, context, *, settings, user_id, emit=None, cancel_requ
     )
 
 
+def run_workflow_results(step, context, *, settings, user_id, emit=None, cancel_requested=None):
+    """Read the stored result of one of the user's finished saved workflow runs, as fenced notes.
+
+    The workflow result reader loads only when a plan reads a result, so resolving this adapter
+    imports nothing beyond this module.
+    """
+    from functions_orchestration_workflow_results import run_workflow_results as read_workflow_results
+
+    return read_workflow_results(
+        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 = {
@@ -1883,6 +1897,7 @@ def run_workflow_run(step, context, *, settings, user_id, emit=None, cancel_requ
     CAPABILITY_ACTION_INVOKE: run_action_invoke,
     CAPABILITY_WORKFLOW_PROPOSE: run_workflow_propose,
     CAPABILITY_WORKFLOW_RUN: run_workflow_run,
+    CAPABILITY_WORKFLOW_RESULTS: run_workflow_results,
 }
 
 

application/single_app/functions_orchestration_composition.py

@@ -30,8 +30,8 @@
 )
 from functions_orchestration_memory import OrchestrationMemoryError
 from functions_orchestration_registry import (
-    CAPABILITY_GENERATE_IMAGE, KNOWLEDGE_BASIS_GENERAL, KNOWLEDGE_BASIS_MIXED, KNOWLEDGE_BASIS_SOURCES,
-    VISUAL_CHART, VISUAL_DIAGRAM, VISUAL_IMAGE_PROPOSAL,
+    CAPABILITY_GENERATE_IMAGE, CAPABILITY_WORKFLOW_RESULTS, 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 (
@@ -50,7 +50,7 @@
     build_answer_visual_guidance, build_existing_charts_note, collect_run_charts,
     image_proposals_available, place_chart_blocks,
 )
-from functions_orchestration_workflow_context import workflow_answer_time_line
+from functions_orchestration_workflow_context import resolve_turn_time_zone, workflow_answer_time_line
 
 
 COMPOSE_POLICY = (
@@ -302,6 +302,7 @@
         for reader in readers.values():
             reader.recheck()
 
+    workflow_results_error = ()
     try:
         recheck()
         if sum(reader.reference.size_bytes for reader in readers.values()) > MAX_VALUE_BYTES:
@@ -320,6 +321,26 @@
             }
             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.
+        workflow_results_policy = None
+        if any(
+            getattr(getattr(reader.reference, 'producer', None), 'capability_id', None) == CAPABILITY_WORKFLOW_RESULTS
+            for reader in readers.values()
+        ):
+            # The workflow result reader loads only when a plan read a saved result.
+            from functions_orchestration_workflow_results import (
+                WorkflowResultsComposeError, workflow_results_compose_inputs,
+            )
+
+            workflow_results_error = WorkflowResultsComposeError
+            planning = getattr(context, 'workflow_planning', None)
+            inputs, workflow_results_policy = workflow_results_compose_inputs(
+                readers, inputs, user_id=user_id, time_zone=resolve_turn_time_zone(
+                    planning.get('time_zone') if isinstance(planning, dict) else None,
+                    getattr(context, 'time_zone', None),
+                ),
+            )
         outputs = step['outputs']
         plain_text = len(outputs) == 1 and outputs[0]['kind'] in ('text-v1', 'markdown-v1')
         unavailable = _missing_optional_inputs(step, context)
@@ -369,6 +390,8 @@
         )
         if time_line:
             messages.append({'role': 'system', 'content': time_line})
+        if workflow_results_policy:
+            messages.append({'role': 'system', 'content': workflow_results_policy})
         messages.append({
             'role': 'user',
             'content': json.dumps({
@@ -522,6 +545,8 @@
             authority_reason = code
         if code == 'result_requires_streaming' or isinstance(exc, WorkflowContextBudgetError):
             failure = build_failure('result_input_too_large')
+        elif isinstance(exc, workflow_results_error):
+            failure = build_failure(exc.code)
         elif isinstance(exc, OrchestrationMemoryError):
             failure = build_failure('context_unavailable')
         elif isinstance(exc, (ResultUnavailableError, PermissionError, ScreeningError)):

application/single_app/functions_orchestration_deliverables.py

@@ -42,13 +42,15 @@
     CAPABILITY_GENERATE_IMAGE,
     CAPABILITY_RENDER_FILE,
     CAPABILITY_WORKFLOW_PROPOSE,
+    CAPABILITY_WORKFLOW_RESULTS,
     CAPABILITY_WORKFLOW_RUN,
     MAX_GENERATED_IMAGES_PER_PLAN,
     VISUAL_CHART,
     VISUAL_DIAGRAM,
     VISUAL_IMAGE_PROPOSAL,
     VISUAL_KINDS,
     WORKFLOW_PROPOSALS_SETTING,
+    WORKFLOW_RESULTS_SETTING,
     WORKFLOW_RUNS_SETTING,
     resolve_admitted_export_catalog,
 )
@@ -123,6 +125,21 @@
     ),
     'workflow_context_unavailable': 'Your saved workflows could not be loaded for this request. Try again later.',
 }
+# Why a saved workflow's stored results cannot be read from this request. Reading a result is not
+# a deliverable either, so these reach the planner as a fact, and only while an administrator has
+# turned using workflow results in chat on.
+WORKFLOW_RESULTS_UNAVAILABLE_REASONS = {
+    'capability_not_enabled_for_orchestration': (
+        'Reading saved workflow results is not enabled for chat orchestration.'
+    ),
+    'workflow_results_disabled': 'Reading saved workflow results in chat is turned off for this deployment.',
+    'workflow_role_required': 'Your account does not have access to saved workflow results in chat.',
+    'workflow_shared_conversation': (
+        'Saved workflow results can be read only in your own conversations, not in shared ones.'
+    ),
+    'workflow_context_unavailable': 'Your saved workflows could not be loaded for this request. Try again later.',
+    'workflow_results_no_workflows': 'You have no saved workflows whose results could be read.',
+}
 _KIND_LABELS = {
     KIND_ANSWER: 'answer', KIND_FILE: 'file', KIND_IMAGE: 'image',
     KIND_CHART: 'chart', KIND_DIAGRAM: 'diagram', KIND_WORKFLOW: 'workflow',
@@ -211,6 +228,17 @@
     return 'workflow_runs_disabled'
 
 
+def _workflow_results_reason(unavailable):
+    """The closed reason this caller's plan cannot read a saved workflow's stored results."""
+    reason = unavailable.get(CAPABILITY_WORKFLOW_RESULTS)
+    if reason == 'not_enabled_for_orchestration':
+        return 'capability_not_enabled_for_orchestration'
+    if isinstance(reason, str) and reason in WORKFLOW_RESULTS_UNAVAILABLE_REASONS:
+        return reason
+    # A settings gate, or a condition this module has no text for, reads as turned off.
+    return 'workflow_results_disabled'
+
+
 def _image_options(capabilities):
     """The exact size, quality, and background values the configured image model accepts."""
     for capability in capabilities or ():
@@ -379,6 +407,27 @@
                 f'{WORKFLOW_RUN_UNAVAILABLE_REASONS[_workflow_run_reason(unavailable)]} '
                 'When the user asks to run a workflow, say why in the answer.'
             )
+    # Reading a saved workflow's stored results is described only while an administrator has turned
+    # it on; otherwise the planner is told exactly what it was before.
+    if settings.get(WORKFLOW_RESULTS_SETTING) is True:
+        if CAPABILITY_WORKFLOW_RESULTS in available:
+            facts.append(
+                'workflow_results reads what one of the user\'s saved workflows produced in a run that '
+                'already finished. The server checks access again when the step runs and adds to the reply '
+                'which run was read. What it returns is the user\'s own earlier output, not evidence: use it '
+                'as notes, never cite it as a source, and never say that a workflow ran again.'
+            )
+            recipes.append({'for': 'What one of the user\'s saved workflows found in a finished run', 'steps': (
+                'one workflow_results step per run to read, naming its catalog handle and a selector, with '
+                'no depends_on and no inputs, bound as an input of the compose step that answers and is '
+                'the final_response. To compare runs, read each one and compose once.'
+            )})
+        else:
+            facts.append(
+                'Reading saved workflow results is unavailable for this request. '
+                f'{WORKFLOW_RESULTS_UNAVAILABLE_REASONS[_workflow_results_reason(unavailable)]} '
+                'When the user asks what a saved workflow found, say why in the answer.'
+            )
     truth.update({'unavailable_reasons': unavailable_reasons, 'facts': facts, 'recipes': recipes})
     return truth
 

application/single_app/functions_orchestration_execution.py

@@ -153,7 +153,8 @@
     resolve_orchestration_model,
 )
 from functions_orchestration_registry import (
-    CapabilityResolutionError, resolve_admitted_export_catalog, resolve_available_capability_ids,
+    CAPABILITY_WORKFLOW_RESULTS, CapabilityResolutionError, resolve_admitted_export_catalog,
+    resolve_available_capability_ids,
 )
 from functions_orchestration_result_contracts import InputBinding, ResultContractError, ResultRef, TaskResult
 from functions_orchestration_result_runtime import read_complete_input, read_result_document_citations
@@ -169,10 +170,11 @@
     summarize_plan,
 )
 from functions_orchestration_workflow_context import (
+    resolve_turn_time_zone,
     workflow_planning_configured,
     workflow_planning_option,
-    workflow_proposals_configured,
     workflow_run_options,
+    workflow_time_zone_configured,
 )
 from functions_workflow_context import (
     WorkflowContextBudgetError,
@@ -699,7 +701,7 @@ def _initialize(self, settings, identity_context, execution_identity):
             self._raise_delivery_infrastructure_failure(exc, storage_required=True)
             raise
         self._preparation_stage = "capabilities"
-        workflow_configured = workflow_proposals_configured(self.settings)
+        workflow_configured = workflow_time_zone_configured(self.settings)
         workflow_planning = (
             self.record.get("workflow_planning") if workflow_planning_configured(self.settings) else None
         )
@@ -1202,6 +1204,40 @@ def _touch_conversation(self, documents):
         except Exception as exc:
             _log_failure("The saved conversation index could not be refreshed.", self.record, exc)
 
+    def _reads_workflow_results(self):
+        return any(
+            step.get("enabled", True) and step.get("capability_id") == CAPABILITY_WORKFLOW_RESULTS
+            for step in self.record["plan"].get("steps") or []
+        )
+
+    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():
+            return []
+        # Imported here, like every caller of the results step's module.
+        from functions_orchestration_workflow_results import (
+            WorkflowResultsComposeError, workflow_results_lineage,
+        )
+
+        try:
+            return workflow_results_lineage(
+                self.record["user_id"], self.record["plan"], record_state.get("execution_steps") or [],
+            )
+        except WorkflowResultsComposeError as exc:
+            # Without a cause this stays a plain failure, so the failure reply replaces the answer
+            # saved on the run instead of being reported as a storage failure that keeps it.
+            raise HarnessExecutionError(exc.code) from None
+
+    def _workflow_results_time_zone(self):
+        planning = getattr(self.context, "workflow_planning", None) or (
+            self.record.get("workflow_planning") if workflow_planning_configured(self.settings) else None
+        )
+        return resolve_turn_time_zone(
+            planning.get("time_zone") if isinstance(planning, dict) else None,
+            getattr(self.context, "time_zone", None)
+            or (self.record.get("time_zone") if workflow_time_zone_configured(self.settings) else None),
+        )
+
     def _finalize(self, result, error, *, refreshes=0):
         # Recovery is an application-owned store dependency, not an import-time dependency.
         from functions_orchestration_recovery import public_execution_fields
@@ -1217,7 +1253,7 @@ def _finalize(self, result, error, *, refreshes=0):
         if status not in {"completed", "waiting", "failed", "cancelled"}:
             error = error or HarnessExecutionError("result_invalid")
         prepared, citations, reader = "", [], None
-        assets = {}
+        assets, workflow_result_contexts = {}, []
         if error is None:
             try:
                 self._revalidate_context()
@@ -1235,12 +1271,15 @@ def _finalize(self, result, error, *, refreshes=0):
                 if not self._delivery_only:
                     citations = result.get("citations") or []
                 self._validate_citations(citations)
+                if prepared:
+                    # The answer is published only while every result it read is still the user's.
+                    workflow_result_contexts = self._workflow_result_lineage(current)
                 if self.context is not None and not current.get("cancellation_requested_at"):
                     assets = generated_image_assets(self.context.task_results, self._read_image_asset)
             except Exception as exc:
                 self._raise_delivery_infrastructure_failure(exc)
                 _log_failure("Execution context could not be reauthorized.", self.record, exc)
-                error, prepared, citations, assets = exc, "", [], {}
+                error, prepared, citations, assets, workflow_result_contexts = exc, "", [], {}, []
         if error is not None:
             failure = _failure(error)
             status = "cancelled" if failure["code"] == "user_cancelled" else "failed"
@@ -1300,6 +1339,21 @@ def _finalize(self, result, error, *, refreshes=0):
         )
         if workflow_runs:
             content.append(workflow_runs)
+        # 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)
+        ):
+            from functions_orchestration_workflow_results import workflow_results_note
+
+            workflow_results = workflow_results_note(
+                self.record["plan"], current.get("execution_steps") or [],
+                time_zone=self._workflow_results_time_zone(), reads=workflow_result_reads,
+            )
+            if workflow_results:
+                content.append(workflow_results)
         if status == "waiting":
             content.append(build_failure("result_not_ready")["message"])
         elif status in {"failed", "cancelled"}:
@@ -1387,6 +1441,11 @@ def _finalize(self, result, error, *, refreshes=0):
                 **({"generated_images": generated_images} if generated_images else {}),
             },
             "token_usage": combined_usage, **reasoning,
+            # The lineage the answer's stored workflow results need, so every later read masks it.
+            **(
+                {"workflow_result_contexts": deepcopy(workflow_result_contexts)}
+                if workflow_result_reads and workflow_result_contexts else {}
+            ),
         }
         documents, web, tools = _partition_citations(citations)
         document = {
@@ -1414,6 +1473,8 @@ def _finalize(self, result, error, *, refreshes=0):
                 if reader is not None:
                     reader.recheck()
                 self._validate_citations(citations)
+                if workflow_result_reads and self._workflow_result_lineage(latest) != workflow_result_contexts:
+                    raise HarnessExecutionError("workflow_result_changed")
             latest_outputs, latest_artifacts = self._file_state()
         except Exception as exc:
             self._raise_delivery_infrastructure_failure(exc)