Phase 6b-1: post workflow results back into chat (server and status route)

application/single_app/background_tasks.py

@@ -921,6 +921,45 @@
         time.sleep(30)
 
 
+def run_workflow_chat_delivery_loop(app=None):
+    """Post chat-started workflow outcomes back into the chats that asked for them.
+
+    A hint from the runtime projection wakes the loop at once. The cross-user sweep runs at most
+    once per sweep interval, and the worker runs it only while it holds the distributed lock.
+    """
+    # Deferred like the orchestration loop: the worker binds application storage on first use.
+    import functions_workflow_chat_delivery as delivery_contract
+    import functions_workflow_chat_delivery_worker as delivery_worker
+
+    interval = delivery_contract.SWEEP_INTERVAL_SECONDS
+    services = None
+    next_sweep_at = 0.0
+    while True:
+        started = time.monotonic()
+        sweep_due = started >= next_sweep_at
+        if sweep_due:
+            next_sweep_at = started + interval
+        try:
+            if services is None:
+                services = delivery_worker.default_workflow_chat_delivery_services(
+                    acquire_lock=acquire_distributed_task_lock,
+                    release_lock=release_distributed_task_lock,
+                )
+            with app.app_context() if app is not None else nullcontext():
+                delivery_worker.process_workflow_chat_delivery_hints(services)
+                if sweep_due:
+                    delivery_worker.run_workflow_chat_delivery_sweep(services)
+        except Exception as exc:
+            log_event(
+                '[WORKFLOW_CHAT_DELIVERY] Delivery slice could not complete.',
+                level=logging.ERROR, extra={'error_type': type(exc).__name__},
+            )
+            # A failure can leave hints queued, so wait without them to avoid a busy loop.
+            time.sleep(interval)
+            continue
+        delivery_contract.wait_for_workflow_chat_delivery_hint(max(1.0, next_sweep_at - time.monotonic()))
+
+
 def run_data_management_scheduler_loop(app=None):
     """Queue due backup and recoverable migration jobs across scaled-out workers."""
     while True:
@@ -1045,6 +1084,7 @@
         ('File Sync scheduler background task started.', run_file_sync_scheduler_loop),
         ('Tabular generated-output scheduler background task started.', run_tabular_generated_output_scheduler_loop),
         ('Orchestration scheduler background task started.', lambda: run_orchestration_scheduler_loop(app=app)),
+        ('Workflow chat delivery background task started.', lambda: run_workflow_chat_delivery_loop(app=app)),
         ('Data Management scheduler background task started.', lambda: run_data_management_scheduler_loop(app=app)),
         ('Content screening scheduler background task started.', lambda: run_content_screening_scheduler_loop(app=app)),
         ('App maintenance background task started.', run_app_maintenance_loop),

application/single_app/config.py

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

application/single_app/functions_conversation_unread.py

@@ -2,7 +2,19 @@
 
 """Helpers for conversation unread assistant-response state."""
 
-from datetime import datetime
+from datetime import datetime, timezone
+
+from azure.core import MatchConditions
+from azure.cosmos import exceptions
+
+
+# Closed outcomes of mark_conversation_unread_guarded.
+UNREAD_MARKED = 'marked'
+UNREAD_ALREADY_MARKED = 'already_marked'
+UNREAD_READ_SINCE = 'read_since'
+UNREAD_NOT_FOUND = 'not_found'
+UNREAD_UNAVAILABLE = 'unavailable'
+UNREAD_CONFLICT = 'conflict'
 
 
 def normalize_conversation_unread_state(conversation_item):
@@ -42,3 +54,101 @@
     normalized_item['last_unread_assistant_message_id'] = None
     normalized_item['last_unread_assistant_at'] = None
     return normalized_item
+
+
+def _parse_utc_timestamp(value):
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.strip().replace('Z', '+00:00'))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _reached(value, threshold):
+    """Whether timestamp ``value`` is at or after ``threshold``; False when either is unreadable."""
+    parsed_value = _parse_utc_timestamp(value)
+    parsed_threshold = _parse_utc_timestamp(threshold)
+    return parsed_value is not None and parsed_threshold is not None and parsed_value >= parsed_threshold
+
+
+def _conversation_unavailable(conversation_item, user_id, require_private):
+    if not isinstance(conversation_item, dict) or conversation_item.get('user_id') != user_id:
+        return True
+    if conversation_item.get('orchestration_deleted') or conversation_item.get('deleted'):
+        return True
+    if not require_private:
+        return False
+    # Imported here so the plain unread helpers above keep their light import graph.
+    from functions_orchestration_memory import conversation_is_private
+
+    return not conversation_is_private(conversation_item, user_id)
+
+
+def mark_conversation_unread_guarded(
+    container,
+    conversation_id,
+    user_id,
+    message_id,
+    planned_at,
+    *,
+    require_private=True,
+    skip_if_read_since=None,
+    max_retries=5,
+):
+    """Mark one assistant message unread on a conversation, under an ETag guard.
+
+    Every attempt rereads the conversation and rechecks that it exists, belongs to ``user_id``,
+    isn't being deleted and, by default, is still private, so a concurrent delete or share wins
+    over the mark. ``last_updated`` is raised to at least ``planned_at`` so the conversation sorts
+    with its newest message.
+
+    ``skip_if_read_since`` is the planned time of an earlier attempt. Marking a chat read keeps its
+    ``last_updated``, so a conversation that already reached that time was either marked and then
+    read, or moved on without the mark; either way it is left alone.
+
+    Returns ``(outcome, conversation)``. The outcome is one of the ``UNREAD_*`` constants and the
+    conversation is the stored item when one was read. The caller invalidates any conversation
+    cache after ``UNREAD_MARKED``. Storage errors other than a missing item or an ETag conflict
+    propagate.
+    """
+    # An empty user id must never match a conversation that has no owner, so every id is required.
+    required = (conversation_id, user_id, message_id, planned_at)
+    if not all(isinstance(value, str) and value for value in required):
+        raise ValueError('An unread mark needs a conversation id, a user id, a message id and a planned time.')
+    attempts = max(1, int(max_retries) + 1)
+    conversation_item = None
+    for _attempt in range(attempts):
+        try:
+            conversation_item = container.read_item(item=conversation_id, partition_key=conversation_id)
+        except exceptions.CosmosResourceNotFoundError:
+            return UNREAD_NOT_FOUND, None
+        if _conversation_unavailable(conversation_item, user_id, require_private):
+            return UNREAD_UNAVAILABLE, conversation_item
+        if conversation_item.get('last_unread_assistant_message_id') == message_id:
+            return UNREAD_ALREADY_MARKED, conversation_item
+        if skip_if_read_since is not None and _reached(conversation_item.get('last_updated'), skip_if_read_since):
+            return UNREAD_READ_SINCE, conversation_item
+
+        candidate = dict(conversation_item)
+        if not _reached(candidate.get('last_updated'), planned_at):
+            candidate['last_updated'] = planned_at
+        candidate = mark_conversation_unread(candidate, message_id, planned_at)
+        try:
+            stored = container.replace_item(
+                item=conversation_id,
+                body=candidate,
+                etag=conversation_item.get('_etag'),
+                match_condition=MatchConditions.IfNotModified,
+            )
+        except exceptions.CosmosResourceNotFoundError:
+            return UNREAD_NOT_FOUND, None
+        except exceptions.CosmosHttpResponseError as exc:
+            if exc.status_code != 412:
+                raise
+            continue
+        return UNREAD_MARKED, stored if isinstance(stored, dict) else candidate
+    return UNREAD_CONFLICT, conversation_item

application/single_app/functions_notifications.py

@@ -34,6 +34,9 @@
 KEY_VAULT_SECRET_REMINDER_NOTIFICATION_TYPE = 'key_vault_secret_expiring'
 M365_APPROVAL_PENDING_NOTIFICATION_TYPE = 'm365_approval_pending'
 M365_APPROVAL_UPDATED_NOTIFICATION_TYPE = 'm365_approval_updated'
+# Pinned by a parity test to functions_workflow_chat_delivery.NOTIFICATION_TYPE, which this module
+# does not import so its import graph stays unchanged.
+WORKFLOW_CHAT_DELIVERY_NOTIFICATION_TYPE = 'workflow_chat_delivery'
 MAX_NOTIFICATION_IDEMPOTENCY_KEY_LENGTH = 512
 WORKFLOW_ALERT_PRIORITY_CONFIG = {
     'info': {
@@ -209,6 +212,10 @@
         'icon': 'bi-bell',
         'color': 'secondary'
     },
+    WORKFLOW_CHAT_DELIVERY_NOTIFICATION_TYPE: {
+        'icon': 'bi-activity',
+        'color': 'info'
+    },
     KEY_VAULT_SECRET_REMINDER_NOTIFICATION_TYPE: {
         'icon': 'bi-safe',
         'color': 'warning'
@@ -715,8 +722,15 @@ def create_chat_response_notification(
     message_id,
     conversation_title='',
     response_preview='',
+    *,
+    idempotency_key=None,
+    strict=False,
 ):
-    """Create a personal notification when a chat response completes."""
+    """Create a personal notification when a chat response completes.
+
+    ``idempotency_key`` and ``strict`` pass through to ``create_notification``, so a server
+    retry with the same key returns the existing notice rather than creating a second one.
+    """
     normalized_title = str(conversation_title or '').strip() or 'Conversation'
     normalized_preview = str(response_preview or '').strip()
     if len(normalized_preview) > 160:
@@ -740,7 +754,9 @@ def create_chat_response_notification(
         metadata={
             'conversation_id': conversation_id,
             'message_id': message_id,
-        }
+        },
+        idempotency_key=idempotency_key,
+        strict=strict,
     )
 
 

application/single_app/functions_orchestration_workflow_runs.py

@@ -73,6 +73,14 @@
     workflow_run_ready,
     workflow_run_settings_gate,
 )
+from functions_workflow_chat_delivery import (
+    WORKFLOW_RUN_DELIVERY_FOLLOW_UP,
+    WORKFLOW_RUN_DELIVERY_FOLLOW_UP_MANY,
+    build_chat_delivery_seed,
+    chat_delivery_applies,
+    normalize_model_selection,
+    normalize_requester_roles,
+)
 
 
 WORKFLOW_RUN_OUTPUT = 'run'
@@ -200,8 +208,9 @@
 }
 _DEFAULT_WORKFLOW_NAME = 'Workflow'
 
-# The reply's note about the saved workflows a plan started. It says where a run's results appear
-# and never promises them in the chat, which only a later delivery step can do.
+# The reply's note about the saved workflows a plan started. It says where a run's results appear.
+# It promises them in the chat only when every run it started recorded a chat delivery, because only
+# then does the server post each run's result back into this conversation when the run finishes.
 WORKFLOW_RUN_NOTE_HEADING = 'Saved workflows:'
 WORKFLOW_RUN_FOLLOW_UP = (
     "Follow the run's progress and results in the workflow's run history in Workflows. Results also "
@@ -585,6 +594,40 @@ def _started_status(run):
     return WORKFLOW_RUN_STATUS_QUEUED
 
 
+def _chat_delivery_seed(settings, context, planning):
+    """Return the record that has the run's result posted back to this chat, or None.
+
+    The cheap settings check comes first, so a deployment with Use Workflow Results In Chat off
+    never reaches the role-aware gate. A failure here only means the run is not delivered: it
+    never fails the start.
+    """
+    if not isinstance(settings, dict) or not settings.get('enable_chat_workflow_results'):
+        return None
+    try:
+        # Settings initialize application storage, so they are imported only once a gate is reached.
+        from functions_settings import is_chat_workflow_results_enabled_for_user
+
+        user_roles = getattr(context, 'user_roles', None)
+        if not is_chat_workflow_results_enabled_for_user(settings, user_roles=user_roles):
+            return None
+        time_zone = getattr(context, 'time_zone', None)
+        if not time_zone and isinstance(planning, dict):
+            time_zone = planning.get('time_zone')
+        return build_chat_delivery_seed(
+            time_zone=time_zone,
+            model_selection=normalize_model_selection(
+                getattr(context, 'seeds', None), getattr(context, 'active_group_ids', None),
+            ),
+            requester_roles=normalize_requester_roles(user_roles),
+        )
+    except Exception as exc:
+        _log(
+            'Chat delivery was not recorded for a started workflow run.', logging.WARNING,
+            run_id=getattr(context, 'run_id', None), error_type=type(exc).__name__,
+        )
+        return None
+
+
 def _start(step, context, *, settings, user_id, recheck, requested_at):
     """Start or link the step's workflow run. Returns an outcome; raises only to fail the step.
 
@@ -651,6 +694,7 @@ def unavailable(reason, workflow_id=None):
         return unavailable(RULE_NOT_DURABLE, workflow_id)
     if workflow.get('status') in M365_ACTIVE_STATES:
         return unavailable(REASON_WAITING_FOR_MICROSOFT_365, workflow_id)
+    chat_delivery = _chat_delivery_seed(settings, context, planning)
 
     recheck()
     runtime_unavailable, runtime_conflict = _runtime_errors()
@@ -671,6 +715,7 @@ def unavailable(reason, workflow_id=None):
                 'requested_by': user_id,
                 'requested_at': requested_at,
             },
+            **({'chat_delivery': chat_delivery} if chat_delivery is not None else {}),
         )
     except runtime_conflict as exc:
         code = getattr(exc, 'code', None)
@@ -684,11 +729,19 @@ def unavailable(reason, workflow_id=None):
     except AzureError as exc:
         raise _WorkflowRunStepFailure(FAILURE_RUNTIME_UNAVAILABLE) from exc
     run = queued.get('run') if isinstance(queued, dict) else None
-    return _outcome(_started_status(run), name=name, workflow_id=workflow_id, run_id=run_id, queued=True)
+    outcome = _outcome(_started_status(run), name=name, workflow_id=workflow_id, run_id=run_id, queued=True)
+    # Only a run this call started, carrying this chat's delivery record, is posted back here.
+    if (
+        chat_delivery is not None
+        and outcome['status'] in (WORKFLOW_RUN_STATUS_QUEUED, WORKFLOW_RUN_STATUS_RUNNING)
+        and chat_delivery_applies(run, context.conversation_id)
+    ):
+        outcome['chat_delivery'] = True
+    return outcome
 
 
 def _sidecar(step, context, *, user_id, producer_run_id, outcome):
-    return {
+    sidecar = {
         'version': WORKFLOW_RUN_VERSION,
         'step_id': step['step_id'],
         'orchestration_run_id': producer_run_id,
@@ -702,6 +755,10 @@ def _sidecar(step, context, *, user_id, producer_run_id, outcome):
         'status': outcome.get('status'),
         'reason': outcome.get('reason'),
     }
+    # Set only by a start that recorded this chat's delivery, never for a linked or rebuilt run.
+    if outcome.get('chat_delivery') is True:
+        sidecar['chat_delivery'] = True
+    return sidecar
 
 
 def _failure(exc):
@@ -879,7 +936,7 @@ def workflow_run_note(plan, execution_steps, *, stopped=False):
         record['step_id']: record for record in (execution_steps if isinstance(execution_steps, list) else ())
         if isinstance(record, dict) and isinstance(record.get('step_id'), str)
     }
-    lines, started = [], 0
+    lines, started, delivered = [], 0, 0
     for step in steps:
         if not isinstance(step, dict) or step.get('capability_id') != CAPABILITY_WORKFLOW_RUN:
             continue
@@ -896,6 +953,8 @@ def workflow_run_note(plan, execution_steps, *, stopped=False):
             continue
         if sidecar['status'] in WORKFLOW_RUN_STARTED_STATUSES:
             started += 1
+            if sidecar.get('chat_delivery') is True:
+                delivered += 1
         lines.append(_step_line(sidecar))
     notes = plan.get('workflow_run_notes') if isinstance(plan.get('workflow_run_notes'), list) else []
     for note in notes:
@@ -909,7 +968,11 @@ def workflow_run_note(plan, execution_steps, *, stopped=False):
         return ''
     parts = ['\n'.join([WORKFLOW_RUN_NOTE_HEADING, *lines])]
     if started:
-        parts.append(WORKFLOW_RUN_FOLLOW_UP_MANY if started > 1 else WORKFLOW_RUN_FOLLOW_UP)
+        # The chat promises the result only when every run started here will be posted back to it.
+        if delivered == started:
+            parts.append(WORKFLOW_RUN_DELIVERY_FOLLOW_UP_MANY if started > 1 else WORKFLOW_RUN_DELIVERY_FOLLOW_UP)
+        else:
+            parts.append(WORKFLOW_RUN_FOLLOW_UP_MANY if started > 1 else WORKFLOW_RUN_FOLLOW_UP)
         if stopped:
             parts.append(WORKFLOW_RUN_STOPPED)
     return '\n\n'.join(parts)
@@ -919,6 +982,8 @@ def workflow_run_note(plan, execution_steps, *, stopped=False):
     'FAILURE_RUNTIME_UNAVAILABLE',
     'NO_WORKFLOW_STARTED',
     'REASON_INVALID',
+    'WORKFLOW_RUN_DELIVERY_FOLLOW_UP',
+    'WORKFLOW_RUN_DELIVERY_FOLLOW_UP_MANY',
     'WORKFLOW_RUN_FOLLOW_UP',
     'WORKFLOW_RUN_FOLLOW_UP_MANY',
     'WORKFLOW_RUN_NOTE_HEADING',