Enable workflow chat capability parity and scoped results delivery

application/single_app/agent_execution_context.py

@@ -125,6 +125,28 @@
 
 
 _execution_frame = ContextVar("agent_execution_frame", default=None)
+_workflow_identity = ContextVar("workflow_execution_identity", default=None)
+
+
+def current_workflow_identity(user_id, conversation_id):
+    identity = _workflow_identity.get()
+    if identity is not None and (
+        identity.user_id == user_id and identity.conversation_id == conversation_id
+    ):
+        return identity
+    return None
+
+
+@contextmanager
+def workflow_identity_scope(identity):
+    """Bind a server-authorized workflow principal without persisting session claims."""
+    if type(identity) is not ExecutionIdentity or not identity.user_id or not identity.conversation_id:
+        raise PermissionError("The workflow execution identity is unavailable.")
+    token = _workflow_identity.set(identity)
+    try:
+        yield identity
+    finally:
+        _workflow_identity.reset(token)
 
 
 def current_agent_execution():
@@ -177,6 +199,10 @@
     dependency. The closure creates a distinct app/request context for every
     invocation; concurrent children never share ``g`` or a mutable session.
     """
+    workflow_identity = current_workflow_identity(user_id, conversation_id)
+    if workflow_identity is not None:
+        return workflow_identity
+
     from flask import current_app, g, has_request_context, request, session
     from public_chat_scope_state import current_public_chat_scope, public_chat_scope_context
 

application/single_app/collaboration_models.py

@@ -703,6 +703,7 @@ def build_collaboration_message_doc_from_legacy(
         'm365_citations',
         'agent_display_name',
         'agent_name',
+        'generated_artifacts',
         'extracted_text',
         'vision_analysis',
         'filename',

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.343"
+VERSION = "0.261.344"
 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

@@ -4,6 +4,7 @@
 
 from copy import deepcopy
 import uuid
+from azure.cosmos.exceptions import CosmosResourceExistsError
 
 from config import *
 from collaboration_models import (
@@ -882,6 +883,8 @@
         'title': conversation_doc.get('title', ''),
         'conversation_kind': conversation_doc.get('conversation_kind'),
         'chat_type': conversation_doc.get('chat_type'),
+        'workflow_id': conversation_doc.get('workflow_id'),
+        'workflow_provenance': deepcopy(conversation_doc.get('workflow_provenance')),
         'status': conversation_doc.get('status', 'active'),
         'created_at': conversation_doc.get('created_at'),
         'updated_at': conversation_doc.get('updated_at'),
@@ -1377,13 +1380,46 @@
     return collaboration_conversation_doc, invited_state_docs, True, source_conversation_doc
 
 
-def create_personal_collaboration_conversation_record(title, creator_user, invited_participants=None):
+def _persist_created_collaboration_conversation(conversation_doc, *, retry_id=None):
+    """Retry a server-bound creation without replacing participants or manual edits."""
+    if not retry_id:
+        cosmos_collaboration_conversations_container.upsert_item(conversation_doc)
+        return conversation_doc
+    try:
+        cosmos_collaboration_conversations_container.create_item(conversation_doc)
+        return conversation_doc
+    except CosmosResourceExistsError:
+        existing = cosmos_collaboration_conversations_container.read_item(
+            item=retry_id, partition_key=retry_id,
+        )
+        if (
+            existing.get('created_by_user_id') != conversation_doc.get('created_by_user_id')
+            or existing.get('chat_type') != conversation_doc.get('chat_type')
+            or (existing.get('scope') or {}).get('group_id') != (conversation_doc.get('scope') or {}).get('group_id')
+            or existing.get('workflow_provenance') != conversation_doc.get('workflow_provenance')
+        ):
+            raise PermissionError('The result conversation does not belong to this workflow run.')
+        return existing
+
+
+def create_personal_collaboration_conversation_record(
+    title, creator_user, invited_participants=None, *, conversation_id=None, workflow_provenance=None,
+):
     conversation_doc = build_personal_collaboration_conversation(
         title=title,
         creator_user=creator_user,
         invited_participants=invited_participants,
+        conversation_id=conversation_id,
+    )
+    if workflow_provenance:
+        conversation_doc.update(
+            workflow_id=workflow_provenance['workflow_id'],
+            workflow_provenance=dict(workflow_provenance),
+            tags=['workflow'],
+        )
+    conversation_doc = _persist_created_collaboration_conversation(
+        conversation_doc, retry_id=conversation_id,
     )
-    cosmos_collaboration_conversations_container.upsert_item(conversation_doc)
 
     user_states = []
     for participant in conversation_doc.get('participants', []):
@@ -1392,15 +1428,17 @@
         invited_by_user_id = ''
         if participant.get('user_id') != conversation_doc.get('created_by_user_id'):
             invited_by_user_id = conversation_doc.get('created_by_user_id')
-        state_doc = build_collaboration_user_state(
+        state_builder = ensure_collaboration_user_state_for_participant if conversation_id else build_collaboration_user_state
+        state_doc = state_builder(
             conversation_doc=conversation_doc,
-            user_summary=participant,
+            **({'participant_summary': participant} if conversation_id else {'user_summary': participant}),
             role=role,
             membership_status=membership_status,
             invited_by_user_id=invited_by_user_id,
             created_at=participant.get('invited_at') or conversation_doc.get('created_at'),
         )
-        cosmos_collaboration_user_state_container.upsert_item(state_doc)
+        if not conversation_id:
+            cosmos_collaboration_user_state_container.upsert_item(state_doc)
         user_states.append(state_doc)
 
     invalidate_conversation_cache_for_item(conversation_doc, reason="collaboration_created")
@@ -1417,7 +1455,9 @@
     return conversation_doc, user_states
 
 
-def create_group_collaboration_conversation_record(title, creator_user, group_doc, invited_participants=None):
+def create_group_collaboration_conversation_record(
+    title, creator_user, group_doc, invited_participants=None, *, conversation_id=None, workflow_provenance=None,
+):
     conversation_doc = build_group_collaboration_conversation(
         title=title,
         creator_user=creator_user,
@@ -1427,8 +1467,17 @@
             group_doc,
             invited_participants,
         ) if invited_participants else None,
+        conversation_id=conversation_id,
+    )
+    if workflow_provenance:
+        conversation_doc.update(
+            workflow_id=workflow_provenance['workflow_id'],
+            workflow_provenance=dict(workflow_provenance),
+            tags=['workflow'],
+        )
+    conversation_doc = _persist_created_collaboration_conversation(
+        conversation_doc, retry_id=conversation_id,
     )
-    cosmos_collaboration_conversations_container.upsert_item(conversation_doc)
 
     user_states = []
     for participant in conversation_doc.get('participants', []):
@@ -2118,12 +2167,15 @@
     return conversation_doc, updated
 
 
-def ensure_collaboration_source_conversation(conversation_doc, current_user):
+def ensure_collaboration_source_conversation(conversation_doc, current_user, *, source_conversation_id=None):
     normalized_current_user = normalize_collaboration_user(current_user)
     if not normalized_current_user:
         raise PermissionError('User not authenticated')
 
-    source_conversation_id = str((conversation_doc or {}).get('source_conversation_id') or '').strip()
+    retry_source_id = source_conversation_id
+    if retry_source_id and conversation_doc.get('created_by_user_id') != normalized_current_user['user_id']:
+        raise PermissionError('Only the conversation owner can create its result backing.')
+    source_conversation_id = str((conversation_doc or {}).get('source_conversation_id') or retry_source_id or '').strip()
     source_chat_type = 'group' if is_group_collaboration_conversation(conversation_doc) else 'personal_single_user'
     source_conversation_doc = None
     source_updated = False
@@ -2134,13 +2186,18 @@
                 item=source_conversation_id,
                 partition_key=source_conversation_id,
             )
+            if retry_source_id and (
+                source_conversation_doc.get('collaboration_conversation_id') != conversation_doc.get('id')
+                or source_conversation_doc.get('user_id') != normalized_current_user['user_id']
+            ):
+                raise PermissionError('The result backing does not belong to this conversation.')
         except CosmosResourceNotFoundError:
             source_conversation_doc = None
             source_conversation_id = ''
 
     timestamp = utc_now_iso()
     if source_conversation_doc is None:
-        source_conversation_id = str(uuid.uuid4())
+        source_conversation_id = retry_source_id or str(uuid.uuid4())
         source_conversation_doc = {
             'id': source_conversation_id,
             'user_id': str((conversation_doc or {}).get('created_by_user_id') or normalized_current_user.get('user_id') or '').strip(),
@@ -2190,7 +2247,21 @@
 
     if source_updated:
         source_conversation_doc['last_updated'] = timestamp
-        cosmos_conversations_container.upsert_item(source_conversation_doc)
+        if retry_source_id and not source_conversation_doc.get('_etag'):
+            try:
+                cosmos_conversations_container.create_item(source_conversation_doc)
+            except CosmosResourceExistsError:
+                existing = cosmos_conversations_container.read_item(
+                    item=source_conversation_id, partition_key=source_conversation_id,
+                )
+                if (
+                    existing.get('collaboration_conversation_id') != conversation_doc.get('id')
+                    or existing.get('user_id') != normalized_current_user['user_id']
+                ):
+                    raise PermissionError('The result backing does not belong to this conversation.')
+                source_conversation_doc = existing
+        else:
+            cosmos_conversations_container.upsert_item(source_conversation_doc)
         invalidate_conversation_cache_for_item(source_conversation_doc, reason="collaboration_source_synced")
 
     if str((conversation_doc or {}).get('source_conversation_id') or '').strip() != source_conversation_id:
@@ -2206,17 +2277,33 @@
     default_sender_user,
     reply_to_message_id=None,
     extra_metadata=None,
+    *,
+    message_id=None,
 ):
     source_message_id = str((source_message_doc or {}).get('id') or '').strip()
     if not source_message_id:
         raise ValueError('source_message_doc.id is required')
 
-    existing_message = get_collaboration_message_by_source_message(
-        (conversation_doc or {}).get('id'),
-        source_message_id,
-    )
-    if existing_message:
-        return existing_message, conversation_doc, False
+    if message_id:
+        try:
+            existing_message = cosmos_collaboration_messages_container.read_item(
+                item=message_id, partition_key=conversation_doc['id'],
+            )
+        except CosmosResourceNotFoundError:
+            existing_message = None
+        if existing_message is not None:
+            if (existing_message.get('metadata') or {}).get('source_message_id') != source_message_id:
+                raise PermissionError('The result message does not belong to this delivery.')
+            conversation_doc = _refresh_collaboration_conversation_message_summary(conversation_doc)
+            return existing_message, conversation_doc, False
+
+    if not message_id:
+        existing_message = get_collaboration_message_by_source_message(
+            (conversation_doc or {}).get('id'),
+            source_message_id,
+        )
+        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)
@@ -2227,6 +2314,8 @@
     )
     if not collaboration_message:
         return None, conversation_doc, False
+    if message_id:
+        collaboration_message['id'] = message_id
 
     source_role = str((source_message_doc or {}).get('role') or '').strip().lower()
     source_metadata = (source_message_doc or {}).get('metadata', {}) if isinstance((source_message_doc or {}).get('metadata'), dict) else {}

application/single_app/functions_group_workflows.py

@@ -551,6 +551,7 @@ def build_group_workflow_document(group_id, workflow_data, actor_user_id, user_i
     reference_owner_id = (existing_workflow or {}).get('user_id') or actor_user_id
     definition_fields = normalize_workflow_definition(
         workflow_data, existing_workflow, tasks, user_id=reference_owner_id, group_id=group_id,
+        settings=settings, new_defaults=origin is None,
     )
     tasks = definition_fields['tasks']
     reference_options = {'resolve_document': resolve_document} if resolve_document is not None else {}
@@ -725,6 +726,14 @@ def build_group_workflow_document(group_id, workflow_data, actor_user_id, user_i
         workflow['next_run_at'] = None
 
     workflow.update(definition_fields)
+    if 'deep_research_authorized' in workflow_data or 'deep_research_authorized' in (existing_workflow or {}):
+        workflow['deep_research_authorized'] = _normalize_bool(
+            workflow_data.get('deep_research_authorized', (existing_workflow or {}).get('deep_research_authorized', False)),
+        )
+        for field in ('deep_research_authorized_by', 'deep_research_authorized_at'):
+            workflow[field] = _normalize_text(
+                workflow_data.get(field, (existing_workflow or {}).get(field)), 'Deep research authorization',
+            ) if workflow['deep_research_authorized'] else ''
     normalize_workflow_run_as(workflow, workflow_data, existing_workflow)
     if workflow.get('definition_version') == 3:
         from functions_workflow_loop_runners import validate_workflow_loop_runners

application/single_app/functions_m365_workflow_binding.py

@@ -122,6 +122,7 @@ def workflow_execution_fingerprint(
     execution.update({
         key: workflow[key] for key in (
             "definition_version", "reference_inputs", "durable_execution", "flow", "limits",
+            "chat_execution", "results_delivery",
         ) if key in workflow
     })
     execution["m365_run_as_user_id"] = str(

application/single_app/functions_notifications.py

@@ -404,7 +404,7 @@ def _join_rule_names(names):
     return f"{', '.join(names[:-1])} and {names[-1]}"
 
 
-def _member_workflow_alert_links(notification, metadata):
+def _member_workflow_alert_links(notification, metadata, reader_user_id=None):
     """Keep only links a member can follow: group conversations the run created in this group.
 
     The workflow's own conversation belongs to its owner, and a personal conversation the agent
@@ -416,6 +416,9 @@ def _member_workflow_alert_links(notification, metadata):
     for target in metadata.get('link_targets') or []:
         if not isinstance(target, dict):
             continue
+        authorized_user_ids = target.get('authorized_user_ids')
+        if authorized_user_ids is not None and reader_user_id not in authorized_user_ids:
+            continue
         link_context = target.get('link_context') if isinstance(target.get('link_context'), dict) else {}
         chat_type = str(link_context.get('chat_type') or '').strip().lower()
         conversation_id = str(target.get('conversation_id') or link_context.get('conversation_id') or '').strip()
@@ -425,11 +428,13 @@ def _member_workflow_alert_links(notification, metadata):
             and conversation_id
             and conversation_id != workflow_conversation_id
         ):
-            links.append(copy.deepcopy(target))
+            member_target = copy.deepcopy(target)
+            member_target.pop('authorized_user_ids', None)
+            links.append(member_target)
     return links
 
 
-def _project_workflow_alert_for_member(notification):
+def _project_workflow_alert_for_member(notification, reader_user_id=None):
     """Return the reduced team alert a group member other than the owner reads."""
     metadata = _get_workflow_alert_metadata(notification)
     member_metadata = {
@@ -442,7 +447,7 @@ def _project_workflow_alert_for_member(notification):
         for match in metadata.get('matched_rules') or []
         if isinstance(match, dict)
     ]
-    links = _member_workflow_alert_links(notification, metadata)
+    links = _member_workflow_alert_links(notification, metadata, reader_user_id)
     member_metadata['link_targets'] = links
 
     priority = _get_workflow_alert_priority(notification)
@@ -474,7 +479,7 @@ def _decorate_workflow_alert(notification, reader_user_id):
     alert reports only the reader's own read and dismissed state, never other members' ids.
     """
     if _reads_team_workflow_alert_as_member(notification, reader_user_id):
-        notification = _project_workflow_alert_for_member(notification)
+        notification = _project_workflow_alert_for_member(notification, reader_user_id)
         content_scope = 'member'
     else:
         notification = dict(notification)
@@ -1003,6 +1008,7 @@ def create_workflow_priority_notification(
     link_context=None,
     metadata=None,
     group_id=None,
+    idempotency_key=None,
 ):
     """Create a workflow alert notification with a priority-aware display.
 
@@ -1031,6 +1037,7 @@ def create_workflow_priority_notification(
             link_url=link_url,
             link_context=link_context or {},
             metadata=alert_metadata,
+            **({'idempotency_key': idempotency_key} if idempotency_key is not None else {}),
         )
 
     return create_notification(
@@ -1041,6 +1048,7 @@ def create_workflow_priority_notification(
         link_url=link_url,
         link_context=link_context or {},
         metadata=alert_metadata,
+        **({'idempotency_key': idempotency_key} if idempotency_key is not None else {}),
     )