Fix orchestrated email review flow and Cosmos response contracts

.github/instructions/cosmos-response-boundaries.instructions.md

@@ -0,0 +1,80 @@
+---
+applyTo: '**/*.py'
+---
+
+# Cosmos document response boundaries
+
+Point reads and returned create/replace/upsert/patch documents in the application's
+pinned Cosmos SDK are `CosmosDict`, a mapping subclass. Query rows and nested JSON
+objects are plain dictionaries. `deepcopy(CosmosDict)` preserves its subclass.
+
+Storage owners must normalize a point response **before copying or passing it to
+an exact-dictionary/strict JSON contract**, using the bootstrap-free
+`normalize_cosmos_document` in `cosmos_document_utils.py`. This helper accepts
+`Mapping`, returns a plain top-level dictionary, preserves all fields and `_etag`,
+and rejects invalid types with `TypeError`. Translate invalid responses through
+the owner's existing failure path. Never normalize arbitrary iterables into a
+document, substitute `{}`, strip ETags, or recursively repair domain values.
+
+Keep domain validators strict. Normalization is not authorization, sanitization,
+metadata stripping, or permission to replay an uncertain external effect. Raw SDK
+responses are compatible with mapping consumers; an exposed wrapper by itself is
+not a product defect. Prove a strict downstream mismatch with a runtime regression
+before changing otherwise-compatible production code.
+
+New Cosmos test containers use the **installed application's SDK `CosmosDict`**
+for point reads and returned write documents by default. Query rows stay plain.
+Plain point responses require an explicit compatibility mode. Preserve ETags,
+partition routing, conditional updates, failure injection and transaction atomicity.
+Domain-level repository mocks do not need to impersonate Cosmos.
+
+## Reviewed inventory and deterministic CI
+
+`functional_tests/cosmos_response_boundaries.json` records point/query calls,
+helper references, consumer classifications and evidence by storage function.
+It separately inventories Cosmos-shaped inline/shared test containers. It is a
+review/change guard, not an interprocedural type checker or proof that every
+external service behaves correctly.
+
+After adding/changing a storage boundary or double:
+
+1. Run `python scripts/check_cosmos_response_boundaries.py --refresh`. Changed
+   implementations become `review-required`; regeneration never approves them.
+2. Review every changed entry's actual consumer. Classify it as
+   `normalized-strict`, `sdk-compatible`, `response-unused`, or `not-cosmos`
+   with a specific contract and runtime test reference. A `not-cosmos` entry
+   needs a concrete explanation of the similarly named non-Cosmos API.
+3. New/changed Cosmos doubles must use `sdk-default`; `legacy-inline` records
+   document unchanged pre-migration doubles, not permission to introduce another.
+   Add a real SDK runtime regression to the inventory's `tests` evidence for
+   changed owners. The gate automatically executes owner-runtime and SDK-default
+   evidence, in addition to its core contracts, so references are not merely
+   checked for existence.
+4. Run `python scripts/check_cosmos_response_boundaries.py --run-contracts`.
+   This checks the inventory and runs its selected suites with
+   `-p test_support.cosmos_network_guard`. Set `PYTHONPATH` to the application
+   and functional test directories using the platform's path separator.
+
+Do not copy a classification blindly, point every owner at a utility-only test,
+or treat a grep match/count as evidence of a defect. Reviews and references are
+invalidated when their implementations disappear or change. Legacy inline
+doubles remain explicitly recorded for staged, owner-specific migration.
+Unchanged legacy mapping consumers distinguish source review plus SDK API
+evidence from owner-runtime SDK coverage. Their source fingerprints form a
+closed baseline; a new or changed owner cannot borrow that limited assurance.
+Replace its baseline entry with real owner-runtime evidence, rather than
+expanding the source-only baseline.
+Owner-runtime evidence executes representative real-owner behavior; it does not
+claim exhaustive branch coverage for every function in that storage owner.
+Existing test-local compatibility seams also have a closed fingerprint baseline.
+A new or changed point override needs explicit `fault-injection` runtime evidence,
+which the gate executes; it cannot silently retain an old easy response override.
+
+The dedicated `cosmos-response-contract-check.yml` installs the existing pinned
+application/test requirements, uses no cloud credentials, and blocks socket/DNS
+network access before test collection. New application storage modules and test
+doubles cannot bypass it by choosing a new filename.
+Tokenizer data needed by the pinned application dependency is warmed during
+dependency setup into a workspace-local cache, before network access is blocked.
+Tests keep the real context-budget calculation; they do not download assets or
+replace budget/token behavior merely to run offline.

.github/workflows/cosmos-response-contract-check.yml

@@ -0,0 +1,44 @@
+name: Cosmos Response Contracts
+
+on:
+  pull_request:
+    paths:
+      - 'application/**/*.py'
+      - 'scripts/**/*.py'
+      - 'deployers/**/*.py'
+      - 'functional_tests/**/*.py'
+      - 'functional_tests/cosmos_response_boundaries.json'
+      - 'ui_tests/**/*.py'
+      - '**/requirements*.txt'
+      - '.github/instructions/cosmos-response-boundaries.instructions.md'
+      - '.github/workflows/cosmos-response-contract-check.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  cosmos-contracts:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    env:
+      PYTHONPATH: application/single_app:functional_tests
+      PYTHONDONTWRITEBYTECODE: '1'
+      TIKTOKEN_CACHE_DIR: ${{ github.workspace }}/.cosmos-contract-assets/tiktoken
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: '3.12'
+          cache: pip
+          cache-dependency-path: |
+            application/single_app/requirements.txt
+            requirements-dev.txt
+      - name: Install pinned application and test dependencies
+        run: python -m pip install -r application/single_app/requirements.txt -r requirements-dev.txt
+      - name: Prepare pinned tokenizer dependency assets before network blocking
+        run: python -c "import tiktoken; tiktoken.get_encoding('cl100k_base'); tiktoken.get_encoding('o200k_base')"
+      - name: Require reviewed storage owners and doubles
+        run: python scripts/check_cosmos_response_boundaries.py
+      - name: Run contracts with external network blocked before collection
+        run: python scripts/check_cosmos_response_boundaries.py --run-contracts

application/single_app/config.py

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

application/single_app/cosmos_document_utils.py

@@ -0,0 +1,15 @@
+# cosmos_document_utils.py
+"""Bootstrap-free normalization of Cosmos point document responses."""
+
+from collections.abc import Mapping
+
+
+def normalize_cosmos_document(value):
+    """Return a plain top-level document, preserving fields and conditional ETags.
+
+    Nested values belong to the domain contract and are deliberately unchanged.
+    This is not a JSON validator, authorization check, or metadata sanitizer.
+    """
+    if not isinstance(value, Mapping):
+        raise TypeError("A Cosmos document response must be a mapping.")
+    return dict(value)

application/single_app/functions_orchestration_actions.py

@@ -681,7 +681,13 @@ async def guard_function(invocation, next):
                         level=logging.ERROR,
                         extra={'action_ref': action_ref, 'error_type': type(exc).__name__},
                     )
-                    failure = ActionExecutionError('An action function could not complete.')
+                    journal = current_operation_journal()
+                    failure = (
+                        journal.failure if journal is not None and journal.failure is not None
+                        else ActionExecutionError('An action function could not complete.')
+                    )
+                    if failure is exc:
+                        raise
                     raise failure from exc
                 refusal = result_refusal(invocation.result.value) if m365_step and invocation.result is not None else None
                 if refusal is not None:

application/single_app/functions_orchestration_adapters.py

@@ -479,6 +479,19 @@ def _m365_step_identity(step, context):
     return f'{root}\x00{step_id}', origin
 
 
+def _m365_card_callback(step, context, emit):
+    def publish(reference):
+        if not callable(emit):
+            return
+        # The producer logs a failed callback while retaining the saved reference.
+        emit({
+            'type': 'm365_pending_action_reference', 'reference': reference,
+            'step_id': step.get('step_id'), 'run_id': _ctx(context, 'run_id', None),
+            'user_message_id': _ctx(context, 'user_message_id', None),
+        })
+    return publish
+
+
 def _progress(step, capability_id, label, status='running'):
     step = step or {}
     return {
@@ -1936,10 +1949,13 @@ def run_action_invoke(step, context, *, settings, user_id, emit, cancel_requeste
         from functions_orchestration_actions import invoke_action
         from semantic_kernel_plugins.plugin_invocation_logger import sanitize_plugin_invocation_value
         from functions_m365_citations import m365_display_time_zone
+        from functions_orchestration_m365 import orchestration_action_card_events
 
         # Microsoft 365 email and event times are shown in the turn's browser time zone.
-        with m365_display_time_zone(_ctx(context, 'time_zone')), operation_step_scope(
-            step, context, user_id, named_inputs,
+        with (
+            orchestration_action_card_events(_m365_card_callback(step, context, emit)),
+            m365_display_time_zone(_ctx(context, 'time_zone')),
+            operation_step_scope(step, context, user_id, named_inputs),
         ):
             result = asyncio.run(invoke_action(
                 action_ref, task, context, settings=settings, user_id=user_id,
@@ -2103,7 +2119,9 @@ def run_agent_invoke(step, context, *, settings, user_id, emit, cancel_requested
         invocation_settings = deepcopy(settings)
         access('agent', settings=invocation_settings, selector=selector)
         from agent_delegation_runtime import delegation_citations, invoke_scoped_agent
-        from functions_orchestration_m365 import agent_loads_actions, agent_step_scope
+        from functions_orchestration_m365 import (
+            agent_loads_actions, agent_step_scope, orchestration_action_card_events,
+        )
         from semantic_kernel_plugins.plugin_invocation_logger import get_plugin_logger
 
         budget = _ctx(context, 'delegation_budget', None)
@@ -2128,9 +2146,11 @@ def scope(_target):
                 )
         from functions_m365_citations import m365_display_time_zone
 
-        with m365_display_time_zone(_ctx(context, 'time_zone')), operation_step_scope(
-            step, context, user_id, named_inputs,
-        ) as journal:
+        with (
+            orchestration_action_card_events(_m365_card_callback(step, context, emit)),
+            m365_display_time_zone(_ctx(context, 'time_zone')),
+            operation_step_scope(step, context, user_id, named_inputs) as journal,
+        ):
             async def invoke():
                 return await invoke_scoped_agent(
                     agent_cfg, task, identity=execution_identity, budget=budget,

application/single_app/functions_orchestration_bootstrap.py

@@ -56,11 +56,12 @@
     upload_generated_file_artifact_stream_for_user,
 )
 from functions_workflow_result_store import WorkflowResultStore, _quota_bytes
+from cosmos_document_utils import normalize_cosmos_document
 
 
 def _document_response(value):
     """Remove an SDK mapping wrapper; leave invalid types for the owning contract to reject."""
-    return dict(value) if isinstance(value, dict) else value
+    return normalize_cosmos_document(value) if isinstance(value, dict) else value
 
 
 def read_owned_conversation(user_id, conversation_id, *, request_timeout=None):
@@ -75,7 +76,7 @@ def read_owned_conversation(user_id, conversation_id, *, request_timeout=None):
         or conversation.get("orchestration_deleted") or conversation.get("deleted")
     ):
         raise OutputUnavailableError("output_conversation_unavailable")
-    return dict(conversation)
+    return normalize_cosmos_document(conversation)
 
 
 def private_external_configuration_digest(value):

application/single_app/functions_orchestration_execution.py

@@ -1101,6 +1101,9 @@ def _send(self, frame):
                 _log_failure("Execution progress delivery stopped.", self.record, exc)
 
     def _progress(self, event):
+        if type(event) is dict and event.get("type") == "m365_pending_action_reference":
+            self._pending_action_created(event)
+            return
         if type(event) is not dict or event.get("type") != "step":
             return
         extra = {
@@ -1118,6 +1121,72 @@ def _progress(self, event):
             event.get("step_index"), event.get("capability_id"), **extra,
         ))
 
+    def _pending_action_created(self, event):
+        # The producer runs inside a separate request. Only its exact step-owned
+        # reference crosses the bridge; tool/model snapshots are never published.
+        from functions_orchestration_m365 import step_request_id
+
+        reference = event.get("reference")
+        step_id = event.get("step_id")
+        root = self.record.get("attempt_root_run_id") or self.record["id"]
+        if (
+            type(reference) is not dict
+            or event.get("run_id") != self.record["id"]
+            or reference.get("user_id") != self.record["user_id"]
+            or reference.get("conversation_id") != self.record["conversation_id"]
+            or not any(
+                step.get("step_id") == step_id for step in self.record["plan"]["steps"]
+            )
+            or reference.get("request_id") != step_request_id(f"{root}\x00{step_id}")
+            or not isinstance(reference.get("id"), str) or not reference["id"]
+        ):
+            return
+        with self.lease.lock:
+            current = self.lease.read()
+            action_ids = list(dict.fromkeys([
+                *(current.get("m365_pending_action_ids") or []), reference["id"],
+            ]))
+            self.lease.update({"m365_pending_action_ids": action_ids})
+        try:
+            from functions_msgraph_pending_actions import get_chat_pending_action_cards
+
+            cards = get_chat_pending_action_cards(
+                self.record["user_id"], self.record["conversation_id"],
+                request_id=reference["request_id"], action_ids=[reference["id"]],
+            )
+            for card in cards:
+                self._send(serialize_sse({
+                    "type": "m365_pending_action", "pending_action": card,
+                    "conversation_id": self.record["conversation_id"],
+                    "run_id": self.record["id"], "step_id": step_id,
+                    "user_message_id": self.record.get("user_message_id"),
+                    "request_id": reference["request_id"],
+                }))
+        except Exception as error:
+            _log_failure("A saved outgoing action could not be displayed live.", self.record, error)
+
+    def _pending_action_payload(self):
+        try:
+            record = self._final_record or self.lease.read()
+            action_ids = record.get("m365_pending_action_ids") or []
+            if not action_ids:
+                return {}
+            from functions_msgraph_pending_actions import get_chat_pending_action_cards
+
+            cards = []
+            for offset in range(0, len(action_ids), 100):
+                cards.extend(get_chat_pending_action_cards(
+                    self.record["user_id"], self.record["conversation_id"],
+                    action_ids=action_ids[offset:offset + 100],
+                ))
+            return {"m365_pending_actions": cards}
+        except Exception as error:
+            _log_failure("Saved outgoing actions could not be refreshed.", self.record, error)
+            return {"m365_pending_actions_error": {
+                "error": "m365_pending_actions_unavailable",
+                "message": "Outgoing actions could not be refreshed. Reload them; do not repeat the request.",
+            }}
+
     def _persist(self, record_type, payload):
         if record_type == "run":
             try:
@@ -1571,6 +1640,8 @@ def _finalize(self, result, error, *, refreshes=0):
                 if workflow_result_reads and workflow_result_contexts else {}
             ),
         }
+        if current.get("m365_pending_action_ids"):
+            metadata["m365_pending_action_ids"] = list(current["m365_pending_action_ids"])
         thread = self._regeneration_thread()
         if thread:
             metadata["thread_info"] = thread
@@ -1659,6 +1730,7 @@ def _finalize(self, result, error, *, refreshes=0):
         payload.update({
             "metadata": document.get("metadata") or {}, "role": document["role"],
             "replace_content": True, "blocked": blocked,
+            **self._pending_action_payload(),
         })
         return [build_content_event(answer), serialize_sse(strip_private_chat_checks(payload))]
 
@@ -1690,6 +1762,9 @@ def _finish(self, result, error):
                     finalization_status="failed", outputs=[],
                 ),
             ]
+            terminal = json.loads(self._frames[-1].partition("data:")[2].strip())
+            terminal.update(self._pending_action_payload())
+            self._frames[-1] = serialize_sse(terminal)
         if self.record.get("requires_fresh_review"):
             status = (self._final_record or {}).get("status") or "failed"
             state = {"completed": "completed", "failed": "failed", "cancelled": "interrupted"}.get(status)