fix(deep-research): require writer-committed final reports

docs/source/architecture/agents/deep-researcher.md

@@ -40,12 +40,10 @@ graph TD
     I --> J[Concurrent reusable researcher workers<br/>one worker per ResearchQuery]
     J --> K[Return structured ResearchNotes<br/>and persist /shared/research_note_*.json]
     K --> L[Normative task: writer-agent]
-    L --> M[Writer reads plan, notes, and captured sources<br/>then writes /shared/output.md]
-    M --> O[Runtime loads writer output]
-    K -.->|Defensive path if writer delegation is missed| S[Orchestrator emits inline Markdown]
-    S --> U{Passes substantive report gate?}
-    U -->|yes| O
-    U -->|no| V[Error: no final Markdown]
+    L --> M[Writer commits /shared/output.md<br/>through the shared-state backend]
+    M --> N{Current bytes match<br/>this run's writer digest?}
+    N -->|yes| O[Runtime loads writer output]
+    N -->|no| V[Error: writer_output_not_committed]
     O --> T{Citation verification enabled?}
     T -->|yes| P[Verify citations against captured sources]
     T -->|no| Q[Skip citation verification]
@@ -90,16 +88,25 @@ sequenceDiagram
     end
     B-->>O: Notes plus /shared/research_note_*.json
     O->>W: task(plan, notes, captured sources)
-    W-->>O: Write /shared/output.md
+    W-->>O: Commit /shared/output.md
     O-->>X: Writer completion marker
+    X->>X: Verify current bytes against the run-local writer digest
     X->>X: Load, optionally verify citations, and sanitize
 ```
 
 The orchestrator serializes dependent stages and tracks progress. It does not
-call source tools directly. The normative flow delegates final synthesis to
-the writer; source access is delegated to the planner and researcher workers.
-If the writer output file is missing, the runtime has a defensive fallback for
-a substantive report emitted inline by the orchestrator.
+call source tools directly. Final synthesis is delegated exclusively to the
+writer; source access is delegated to the planner and researcher workers. A
+non-empty file is not sufficient proof of completion: the runtime accepts the
+report only when its exact UTF-8 bytes match the digest recorded after a
+successful writer mutation in the current run. Missing, stale, or modified
+output fails closed with ``writer_output_not_committed``.
+
+The commit proof is intentionally run-local and is not restored from a
+checkpoint after a process restart. The guarantee is one overwrite-capable
+shared-state backend update followed by byte-exact digest verification; it does
+not claim cross-provider filesystem atomicity. A resumed run without its proof
+therefore fails closed rather than trusting pre-existing output.
 
 ## Runtime Roles
 
@@ -147,11 +154,11 @@ The roles use middleware appropriate to their contracts:
 
 | Role | Relevant behavior |
 | ---- | ----------------- |
-| Orchestrator | DeepAgents task, todo, and filesystem support; source-routing guard; tool-name validation restricted to helpers and `run_research_batch`; tool and model retry handling |
-| Source router | Minimal filesystem and retry middleware; catalog lookup and `write_file` only |
-| Planner | Source capture, retries, filesystem access, todo suppression, structured `ResearchPlan` validation, and automatic plan persistence |
-| Researcher worker | Filesystem context, optional skills, summarization, source capture, retries, and structured `ResearchNotes` validation |
-| Writer | Filesystem context, optional synthesis skills, source-registry access, retries, and todo suppression; no source-search tools |
+| Orchestrator | DeepAgents task, todo, and filesystem support; source-routing and final-report ownership guards; tool-name validation restricted to helpers and `run_research_batch`; tool and model retry handling |
+| Source router | Minimal filesystem and retry middleware; catalog lookup and `write_file` only; final-report mutation denied |
+| Planner | Source capture, retries, filesystem access, todo suppression, structured `ResearchPlan` validation, automatic plan persistence, and final-report mutation denial |
+| Researcher worker | Filesystem context, optional skills, summarization, source capture, retries, structured `ResearchNotes` validation, and final-report mutation denial |
+| Writer | Filesystem context, optional synthesis skills, source-registry access, retries, todo suppression, overwrite-safe `/shared/output.md` commit, and run-local digest verification; no source-search tools |
 
 The root graph is constructed with `create_deep_agent`. The reusable
 researcher runnable is constructed separately with `create_agent`, which is
@@ -288,11 +295,18 @@ edit. The writer performs no new research, writes the complete final answer to
 `/shared/output.md`, and returns a short completion marker. The runtime loads
 the Markdown from that file.
 
-This is the normative synthesis contract. As a defensive compatibility path,
-`_salvage_inline_report()` accepts the orchestrator's final message when the
-output file is missing, but only if the message is substantive Markdown: at
-least 400 characters with a Markdown heading and not merely the writer
-completion marker. Otherwise, missing writer output remains an error.
+This is the only synthesis contract. The runtime accepts only non-empty writer
+output whose exact UTF-8 bytes match the digest recorded after a successful
+writer mutation in the current run. After one bounded corrective turn, missing,
+stale, or mismatched output fails closed with
+``writer_output_not_committed``; inline orchestrator messages are not salvaged
+as final reports.
+
+``/shared/output.md`` is the sole writer-facing path. When
+``CompositeBackend`` routes ``/shared/`` through ``StateBackend``, raw graph
+state may represent that file under the internal route-stripped key
+``/output.md``. Ownership and digest checks recognize that internal alias, but
+agents must not target it directly.
 
 ### Phase 5: Citation Verification (Post-Processing)
 

src/aiq_agent/agents/deep_researcher/agent.py

@@ -35,6 +35,7 @@
 from aiq_agent.common.citation_verification import source_entries_from_parent_context
 from aiq_agent.common.citation_verification import verify_citations
 
+from .custom_middleware import FinalReportCommitTracker
 from .custom_middleware import SourceRegistryMiddleware
 from .deepagents_runtime import DeepAgentsRuntime
 from .deepagents_runtime import DeepResearchSandboxConfig
@@ -54,14 +55,6 @@
 # Path to this agent's directory (for loading prompts)
 AGENT_DIR = Path(__file__).parent
 
-# Salvage gate: when the orchestrator synthesizes the report inline instead of delegating to
-# writer-agent, no /shared/output.md is written. We accept the final message as the report only
-# when it is clearly a substantive report (long + has a markdown heading), so workflow chatter is
-# still rejected and the strict file-first contract is preserved.
-_WRITER_COMPLETION_MARKER = "Wrote /shared/output.md"
-_MIN_INLINE_REPORT_CHARS = 400
-_MD_HEADING_RE = re.compile(r"(?m)^#{1,6}\s")
-
 
 class DeepResearcherAgent:
     """
@@ -181,7 +174,12 @@ def _load_prompts(self) -> dict[str, str]:
 
         return prompts
 
-    def _build_orchestrator_agent(self, state: DeepResearchAgentState) -> Any:
+    def _build_orchestrator_agent(
+        self,
+        state: DeepResearchAgentState,
+        *,
+        final_report_tracker: FinalReportCommitTracker,
+    ) -> Any:
         """Build the orchestrator graph for the current state."""
         return build_deep_research_graph(
             llm_provider=self.llm_provider,
@@ -196,51 +194,25 @@ def _build_orchestrator_agent(self, state: DeepResearchAgentState) -> Any:
             domain_catalog_path=self.domain_catalog_path,
             enable_source_router=self.enable_source_router,
             max_research_concurrency=self.max_research_concurrency,
+            final_report_tracker=final_report_tracker,
         )
 
-    def _extract_final_markdown(self, result: dict | Any, files: dict[str, Any] | None = None) -> str | None:
-        """Extract final Markdown from output files."""
-        output_paths = ("/shared/output.md", "/output.md")
+    def _extract_final_markdown(
+        self,
+        result: dict | Any,
+        files: dict[str, Any] | None = None,
+        *,
+        final_report_tracker: FinalReportCommitTracker,
+    ) -> str | None:
+        """Extract only Markdown committed by the writer during this run."""
         # Resolve result files first, then fall back to the passed-in files (state.files) and
         # finally an empty dict. Without the explicit grouping, `or files or {}` bound only to
-        # the else branch, so a dict result lacking a usable "files" key silently discarded the
-        # fallback and skipped straight to inline salvage even when output files existed.
+        # the else branch, so a dict result lacking a usable "files" key silently discarded
+        # the fallback even when output files existed.
         result_files = result.get("files", None) if isinstance(result, dict) else getattr(result, "files", None)
         files = result_files or files or {}
-        if isinstance(files, dict):
-            for output_path in output_paths:
-                output_entry = files.get(output_path)
-                if isinstance(output_entry, dict):
-                    output_entry = output_entry.get("content")
-                if isinstance(output_entry, bytes):
-                    output_entry = output_entry.decode("utf-8")
-                if isinstance(output_entry, str) and output_entry.strip():
-                    return output_entry.strip()
-        return self._salvage_inline_report(result)
-
-    @staticmethod
-    def _salvage_inline_report(result: dict | Any) -> str | None:
-        """Salvage a report the orchestrator wrote inline instead of via writer-agent.
-
-        When the orchestrator skips the writer-agent delegation and emits the full report in its
-        final message, no output file exists. Accept that message only when it is clearly a
-        substantive report so plain workflow chatter is still rejected.
-        """
-        messages = result.get("messages") if isinstance(result, dict) else getattr(result, "messages", None)
-        if not messages:
-            return None
-        content = getattr(messages[-1], "content", None)
-        if not isinstance(content, str):
-            return None
-        stripped = content.strip()
-        if (
-            not stripped
-            or stripped == _WRITER_COMPLETION_MARKER
-            or len(stripped) < _MIN_INLINE_REPORT_CHARS
-            or not _MD_HEADING_RE.search(stripped)
-        ):
-            return None
-        return stripped
+        committed = final_report_tracker.committed_text(files)
+        return committed.strip() if committed is not None else None
 
     @staticmethod
     def _read_seed_file_text(files: dict[str, Any], path: str) -> str | None:
@@ -281,7 +253,8 @@ async def run(self, state: DeepResearchAgentState) -> DeepResearchAgentState:
         if prepared_files != state.files:
             state = state.model_copy(update={"files": prepared_files})
         self._seed_parent_sources(state.files)
-        agent = self._build_orchestrator_agent(state)
+        final_report_tracker = FinalReportCommitTracker()
+        agent = self._build_orchestrator_agent(state, final_report_tracker=final_report_tracker)
 
         messages = state.messages
         if messages:
@@ -295,9 +268,13 @@ async def run(self, state: DeepResearchAgentState) -> DeepResearchAgentState:
         try:
             result = await agent.ainvoke(state, config={"callbacks": self.callbacks} if self.callbacks else None)
 
-            final_message = self._extract_final_markdown(result, state.files)
+            final_message = self._extract_final_markdown(
+                result,
+                state.files,
+                final_report_tracker=final_report_tracker,
+            )
             if final_message is None:
-                raise ValueError("writer-agent did not produce a final Markdown answer")
+                raise RuntimeError("writer_output_not_committed")
 
             # Post-process: verify citations against source registry
             if self.enable_citation_verification and self.source_registry_middleware.has_sources():