Phase 6a: workflow results reader and Follow up in chat

application/single_app/admin_settings_fields.py

@@ -4059,6 +4059,20 @@
             "default": True,
             "depends_on": {"key": "allow_user_workflows", "equals": True},
         },
+        {
+            "key": "enable_chat_workflow_results",
+            "type": "switch",
+            "label": "Use Workflow Results In Chat",
+            "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. "
+                "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."
+            ),
+            "default": False,
+            "depends_on": {"key": "allow_user_workflows", "equals": True},
+        },
         {
             "key": "allow_group_workflows",
             "type": "switch",

application/single_app/config.py

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

application/single_app/functions_collaboration.py

@@ -69,6 +69,7 @@
     delete_thoughts_for_conversation,
     get_thoughts_for_message,
 )
+from functions_workflow_result_masking import message_uses_workflow_result, withhold_workflow_result_message
 
 
 PERSONAL_COLLABORATION_MANAGER_ROLES = {
@@ -700,6 +701,9 @@
 
 def build_collaboration_message_metadata_payload(message_doc, conversation_doc):
     source_message_doc = _get_collaboration_source_message(message_doc)
+    # The source is the owner's raw chat message; participants only ever see its withheld form.
+    if message_uses_workflow_result(source_message_doc):
+        source_message_doc = withhold_workflow_result_message(source_message_doc)
     message_metadata = deepcopy(message_doc.get('metadata', {}) if isinstance(message_doc.get('metadata'), dict) else {})
     source_metadata = deepcopy(source_message_doc.get('metadata', {}) if isinstance((source_message_doc or {}).get('metadata'), dict) else {})
     merged_metadata = _publicize_image_revisions({
@@ -967,6 +971,9 @@
     copied_messages = []
     source_to_collaboration_message_ids = {}
     for raw_message in raw_messages:
+        # Participants read these copies directly, so a workflow result's answer is stored withheld.
+        if message_uses_workflow_result(raw_message):
+            raw_message = withhold_workflow_result_message(raw_message)
         collaboration_message = build_collaboration_message_doc_from_legacy(
             collaboration_conversation_id,
             raw_message,
@@ -996,6 +1003,20 @@
     return copied_messages
 
 
+def _carry_summary_into_collaboration(collaboration_conversation_doc, source_conversation_doc, raw_messages):
+    """Copy the chat summary into the collaboration unless the copies withhold a workflow result.
+
+    A summary may repeat withheld workflow-result answers, so both chats drop it and syncing
+    can't restore it. Call after prepare_m365_history_publication has fingerprinted the original;
+    the caller's final source upsert persists the cleared summary.
+    """
+    if any(message_uses_workflow_result(message) for message in raw_messages):
+        collaboration_conversation_doc['summary'] = None
+        source_conversation_doc['summary'] = None
+        return
+    collaboration_conversation_doc['summary'] = source_conversation_doc.get('summary')
+
+
 def ensure_personal_collaboration_for_legacy_conversation(source_conversation_id, owner_user, invited_participants=None):
     source_conversation_doc = cosmos_conversations_container.read_item(
         item=source_conversation_id,
@@ -1065,7 +1086,9 @@
         source_conversation_doc,
     )
     collaboration_conversation_doc['strict'] = bool(source_conversation_doc.get('strict', False))
-    collaboration_conversation_doc['summary'] = source_conversation_doc.get('summary')
+    _carry_summary_into_collaboration(
+        collaboration_conversation_doc, source_conversation_doc, publication.messages,
+    )
 
     source_context = list(source_conversation_doc.get('context', []) or [])
     if source_context:
@@ -1277,7 +1300,9 @@
         source_conversation_doc,
     )
     collaboration_conversation_doc['strict'] = bool(source_conversation_doc.get('strict', False))
-    collaboration_conversation_doc['summary'] = source_conversation_doc.get('summary')
+    _carry_summary_into_collaboration(
+        collaboration_conversation_doc, source_conversation_doc, publication.messages,
+    )
     collaboration_conversation_doc[source_link_field] = source_conversation_id
     if publication.request_id:
         collaboration_conversation_doc['m365_publication'] = {
@@ -2172,6 +2197,8 @@
     if existing_message:
         return existing_message, conversation_doc, False
 
+    if message_uses_workflow_result(source_message_doc):
+        source_message_doc = withhold_workflow_result_message(source_message_doc)
     collaboration_message = build_collaboration_message_doc_from_legacy(
         (conversation_doc or {}).get('id'),
         source_message_doc,

application/single_app/functions_orchestration_context.py

@@ -1784,6 +1784,10 @@ def normalize_history_message(message):
     if (
         metadata.get('masked')
         or metadata.get('is_generated_chat_artifact')
+        # Answers built from a stored workflow result, or inheriting one, stay out of
+        # orchestration history until orchestration can re-check that result itself.
+        or 'workflow_result' in metadata
+        or 'workflow_result_contexts' in metadata
         or (metadata.get('chat_content_checks') or {}).get('decision') == 'block'
         or (metadata.get('content_moderation') or {}).get('removed') is True
         or thread.get('active_thread') is False

application/single_app/functions_orchestration_memory.py

@@ -1,7 +1,7 @@
 # functions_orchestration_memory.py
 """Read-only, audience-bound saved memory for orchestration.
 
-Version: 0.261.132
+Version: 0.261.214
 """
 
 from azure.core.exceptions import AzureError
@@ -48,6 +48,21 @@
     return audience
 
 
+# Kept beside the audience check so saved analysis and Follow up can ask it without importing
+# the workflow planning context, which would close an import cycle.
+def conversation_is_private(conversation, user_id):
+    """Whether only the requester can read this conversation.
+
+    A shared, collaborative or multi-user conversation, or one converted to a collaboration, is
+    not private: a proposal there would put the requester's workflows in front of other people.
+    """
+    try:
+        audience = validate_memory_audience(conversation, user_id)
+    except OrchestrationMemoryError:
+        return False
+    return audience.get('kind') == 'personal' and not conversation.get('converted_to_collaboration_at')
+
+
 def validate_memory_context(conversation, user_id, expected_audience=None, scope=None):
     """Reauthorize the scope actually recalled, without rereading or changing its facts."""
     audience = validate_memory_audience(conversation, user_id, expected_audience)

application/single_app/functions_orchestration_workflow_context.py

@@ -1,7 +1,7 @@
 # functions_orchestration_workflow_context.py
 """Planning context for workflow proposals and workflow runs from chat orchestration.
 
-Version: 0.261.212
+Version: 0.261.214
 
 Chat orchestration can propose a personal workflow (the ``workflow_propose`` capability) and start
 one the user already has (the ``workflow_run`` capability). The planner writes a workflow
@@ -34,7 +34,7 @@
     get_m365_enabled_function_names,
 )
 from functions_msgraph_operations import get_msgraph_enabled_function_names, resolve_msgraph_action_capabilities
-from functions_orchestration_memory import OrchestrationMemoryError, validate_memory_audience
+from functions_orchestration_memory import conversation_is_private
 # One definition of each: the step schema, the deliverables and this module read the same values.
 from functions_orchestration_registry import (
     CAPABILITY_WORKFLOW_PROPOSE as WORKFLOW_PROPOSE_CAPABILITY_ID,
@@ -273,19 +273,6 @@ def workflow_run_gate(settings, user_roles):
     return None
 
 
-def conversation_is_private(conversation, user_id):
-    """Whether only the requester can read this conversation.
-
-    A shared, collaborative or multi-user conversation, or one converted to a collaboration, is
-    not private: a proposal there would put the requester's workflows in front of other people.
-    """
-    try:
-        audience = validate_memory_audience(conversation, user_id)
-    except OrchestrationMemoryError:
-        return False
-    return audience.get('kind') == 'personal' and not conversation.get('converted_to_collaboration_at')
-
-
 # ---------------------------------------------------------------------------
 # Default readers (lazy: each one needs initialized application storage)
 # ---------------------------------------------------------------------------

application/single_app/functions_saved_analysis.py

@@ -18,6 +18,7 @@
 from functions_generated_file_exports import build_saved_analysis_export
 from functions_workflow_context import WorkflowContextBudgetError, calculate_workflow_context_budget
 from functions_workflow_identity import normalize_workflow_iteration_path
+from functions_workflow_result_masking import message_uses_workflow_result, withhold_workflow_result_message
 from functions_workflow_result_store import WorkflowResultStorageUnavailableError, _quota_bytes
 from functions_workflow_runtime_store import WorkflowRuntimeConflict
 from functions_workflow_results import (
@@ -437,11 +438,103 @@ def rebase_saved_analysis_message(message):
 def is_saved_analysis_unavailable(message):
     metadata = message.get("metadata") or {}
     descriptor = metadata.get("saved_analysis")
-    return isinstance(descriptor, Mapping) and descriptor.get("available") is False
+    workflow_result = metadata.get("workflow_result")
+    return (
+        (isinstance(descriptor, Mapping) and descriptor.get("available") is False)
+        or (isinstance(workflow_result, Mapping) and workflow_result.get("available") is False)
+    )
+
+
+_WORKFLOW_RESULT_WITHHELD_LOG = "[WorkflowResults] A chat answer's workflow result was withheld on read."
+
+
+def _read_personal_conversation(conversation_id):
+    from config import cosmos_conversations_container
+
+    return cosmos_conversations_container.read_item(item=conversation_id, partition_key=conversation_id)
+
+
+class _WorkflowResultLineage:
+    """Checks, once per read call, the workflow results that stored messages relied on.
+
+    A message stays readable only in the reader's own private chat and only while every
+    workflow result it relied on is still readable by that reader exactly as selected.
+    Each distinct result is authorized once and each distinct chat is read once per
+    call; any failure withholds the message and is never raised.
+    """
+
+    def __init__(self, user_id, *, result_reader=None, conversation_reader=None):
+        from functions_workflow_result_reader import authorize_workflow_result_context
+
+        self._user_id = user_id
+        self._authorize = result_reader or authorize_workflow_result_context
+        self._read_conversation = conversation_reader or _read_personal_conversation
+        self._private_chats = {}
+        self._readable_results = {}
+
+    @staticmethod
+    def _log(check, *, code=None, error=None):
+        extra = {"check": check}
+        if code:
+            extra["code"] = code
+        if error is not None:
+            extra["error_type"] = type(error).__name__
+        log_event(_WORKFLOW_RESULT_WITHHELD_LOG, extra=extra)
+
+    def _private(self, conversation_id):
+        # Answers from private results never follow a chat into a shared or converted one.
+        from functions_orchestration_memory import conversation_is_private
+
+        if not isinstance(conversation_id, str) or not conversation_id:
+            return False
+        if conversation_id not in self._private_chats:
+            try:
+                private = conversation_is_private(self._read_conversation(conversation_id), self._user_id) is True
+            except Exception as exc:
+                self._log("conversation", error=exc)
+                private = False
+            else:
+                if not private:
+                    self._log("conversation", code="workflow_result_private_only")
+            self._private_chats[conversation_id] = private
+        return self._private_chats[conversation_id]
+
+    def _readable(self, context):
+        from functions_workflow_result_reader import WorkflowResultUnavailable, workflow_result_context_key
+
+        key = workflow_result_context_key(context)
+        if key not in self._readable_results:
+            try:
+                self._authorize(self._user_id, context)
+            except Exception as exc:
+                self._log(
+                    "result", error=exc, code=exc.code if isinstance(exc, WorkflowResultUnavailable) else None,
+                )
+                self._readable_results[key] = False
+            else:
+                self._readable_results[key] = True
+        return self._readable_results[key]
+
+    def allows(self, message):
+        from functions_workflow_result_reader import workflow_result_message_contexts
+
+        contexts, malformed = workflow_result_message_contexts(message)
+        if malformed:
+            return False
+        if not contexts:
+            return True
+        if not self._private(message.get("conversation_id")):
+            return False
+        return all(self._readable(context) for context in contexts)
+
+    def sanitize(self, message):
+        # A question keeps the reader's own text; only its link to the result goes.
+        return message if self.allows(message) else withhold_workflow_result_message(message)
 
 
 def authorize_saved_analysis_message_read(
     user_id, conversation_id, message_id, *, allow_pending=False, message_loader=None, result_reader=None,
+    workflow_result_reader=None, conversation_reader=None,
 ):
     """Keep persisted thoughts and other message-derived views under result access."""
     try:
@@ -451,6 +544,12 @@ def authorize_saved_analysis_message_read(
             raise
         _authorize_conversation(user_id, conversation_id)
         return False
+    if message_uses_workflow_result(message):
+        lineage = _WorkflowResultLineage(
+            user_id, result_reader=workflow_result_reader, conversation_reader=conversation_reader,
+        )
+        if not lineage.allows(message):
+            raise PermissionError("The workflow result behind this message is unavailable.")
     read = result_reader or load_saved_analysis
     for context in analysis_result_contexts(message):
         read(user_id, context)
@@ -1684,11 +1783,21 @@ def read_saved_analysis_page(
     return response
 
 
-def sanitize_saved_analysis_messages(messages, user_id, *, result_reader=None):
+def sanitize_saved_analysis_messages(
+    messages, user_id, *, result_reader=None, workflow_result_reader=None, conversation_reader=None,
+):
     """Withhold result-derived content on new reads after source access is lost."""
     read = result_reader or load_saved_analysis
+    lineage = None
     sanitized = []
     for message in messages:
+        if message_uses_workflow_result(message):
+            # Checked first: a withheld answer keeps no analysis lineage to check afterwards.
+            if lineage is None:
+                lineage = _WorkflowResultLineage(
+                    user_id, result_reader=workflow_result_reader, conversation_reader=conversation_reader,
+                )
+            message = lineage.sanitize(message)
         metadata = message.get("metadata") or {}
         descriptor = metadata.get("saved_analysis")
         if not any(metadata.get(field) for field in ("saved_analysis", "saved_analyses", "analysis_result_contexts")):

application/single_app/functions_settings.py

@@ -1169,6 +1169,14 @@ def is_workflow_assistant_enabled_for_user(settings, user_roles=None):
     return source_settings.get('enable_workflow_ai_assistant', False) is True
 
 
+def is_chat_workflow_results_enabled_for_user(settings, user_roles=None):
+    """Return True when a user may ask about their personal workflow results in chat."""
+    source_settings = settings or {}
+    if source_settings.get('enable_chat_workflow_results', False) is not True:
+        return False
+    return is_user_workflows_enabled_for_user(source_settings, user_roles=user_roles)
+
+
 def _authorize_user_settings_access(user_id, operation, allow_cross_user=False):
     """Authorize user-settings access for the current request context."""
     normalized_user_id = str(user_id or '').strip()
@@ -1529,6 +1537,8 @@ def get_settings(use_cosmos=False, include_source=False):
         'require_member_of_workflow_user': False,
         # The AI workflow assistant in the V2 editor. It only applies where personal workflows do.
         'enable_workflow_ai_assistant': True,
+        # Lets users ask chat about the stored results of their personal workflow runs.
+        'enable_chat_workflow_results': False,
         'workflow_max_tasks': 50,
         'workflow_max_loop_items': WORKFLOW_LOOP_ITEMS_DEFAULT,
         'workflow_max_repeat_iterations': WORKFLOW_REPEAT_ITERATIONS_DEFAULT,
@@ -2271,6 +2281,12 @@ def update_settings(new_settings, *, expected_etag=None):
                 new_settings['workflow_max_repeat_iterations']
             ),
         }
+    if isinstance(new_settings, dict) and 'enable_chat_workflow_results' in new_settings:
+        # Only a real boolean true lets chat read workflow results; a string or number saves as off.
+        new_settings = {
+            **new_settings,
+            'enable_chat_workflow_results': new_settings['enable_chat_workflow_results'] is True,
+        }
     if isinstance(new_settings, dict) and 'workflow_min_schedule_interval_seconds' in new_settings:
         new_settings = {
             **new_settings,
@@ -3978,6 +3994,21 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
+def workflow_results_required(f):
+    """Allow the route only when workflow results in chat are available to the signed-in user."""
+    @wraps(f)
+    def wrapper(*args, **kwargs):
+        settings = get_settings()
+        user_roles = (session.get('user') or {}).get('roles', [])
+        if is_chat_workflow_results_enabled_for_user(settings, user_roles=user_roles):
+            return f(*args, **kwargs)
+        return jsonify({
+            'error': 'Workflow results in chat are not available.',
+            'code': 'workflow_results_disabled',
+        }), 403
+    return wrapper
+
+
 def enabled_required(setting_key):
     def decorator(f):
         @wraps(f)