Merge 451a864096 into 9677f3d80c

docs/docs.json

@@ -74,6 +74,7 @@
                       "v3/documentation/features/advanced/search",
                       "v3/documentation/features/advanced/using-filters",
                       "v3/documentation/features/advanced/structured-outputs",
+                      "v3/documentation/features/advanced/evidence",
                       "v3/documentation/features/advanced/streaming-response",
                       "v3/documentation/features/advanced/file-uploads",
                       "v3/documentation/features/advanced/deleting-data"

docs/v3/documentation/features/advanced/evidence.mdx

@@ -0,0 +1,209 @@
+---
+title: "Evidence"
+description: "See what a chat answer was built from"
+icon: "list-check"
+---
+
+By default the [chat endpoint](/v3/documentation/features/chat) returns an answer and nothing else, which makes it hard to check. Pass `include_evidence` and the response also carries the conclusions and messages the dialectic read while answering, plus the tools it called.
+
+## Basic Usage
+
+<CodeGroup>
+```python Python
+from honcho import Honcho
+
+honcho = Honcho()
+peer = honcho.peer("user-123")
+
+result = peer.chat(
+    "What coffee does this user prefer?",
+    include_evidence=True,
+)
+
+print(result.content)
+# "The user prefers dark roast, usually from local roasters."
+
+for conclusion in result.evidence.conclusions:
+    print(f"[{conclusion.level}] {conclusion.id}: {conclusion.content}")
+# [explicit] hK3mZq...: User prefers dark roast coffee
+# [inductive] pR8xLw...: User buys from local roasters
+
+for call in result.evidence.tool_calls:
+    print(call.tool_name, call.tool_input)
+# search_memory {'query': 'coffee preference', 'top_k': 20}
+```
+
+```typescript TypeScript
+import { Honcho } from '@honcho-ai/sdk';
+
+const honcho = new Honcho({});
+const peer = await honcho.peer("user-123");
+
+const { content, evidence } = await peer.chat(
+  "What coffee does this user prefer?",
+  { includeEvidence: true }
+);
+
+console.log(content);
+
+for (const conclusion of evidence?.conclusions ?? []) {
+  console.log(`[${conclusion.level}] ${conclusion.id}: ${conclusion.content}`);
+}
+```
+</CodeGroup>
+
+Without `include_evidence`, `chat` returns the answer on its own exactly as before, and the server collects nothing — asking for evidence is the only thing that turns collection on.
+
+## What evidence is, and is not
+
+Evidence is **collated from what the agent read**, not reported by the model. As the dialectic runs its tool loop, every conclusion and message a read path returns is recorded, and the finished list is returned alongside the answer.
+
+That has a consequence worth being clear about: **evidence over-reports**. A conclusion appears because the agent saw it, which is not proof the answer relied on it. A query that prefetches twenty-five conclusions and answers from three will list all twenty-five.
+
+The alternative — asking the model which sources it used — reads better but fails quietly. Weaker models and lower reasoning levels produce incomplete citations, invented IDs, or none at all, and you cannot tell a sparse citation list from a sparse answer. Collation is deterministic, costs no model tokens, and behaves identically at every reasoning level. Treat evidence as *what was available to the answer*, and audit within it.
+
+Evidence is built for auditing and analytics — working out why an answer looks the way it does, or measuring what recall actually reaches the agent. It is not a read API, and it is not meant to sit in a hot path.
+
+Two further limits:
+
+- `tool_calls` records **successful** invocations. A tool call that errored is retried or worked around by the agent and does not appear, so this is not a complete execution trace.
+- Tool **results** are omitted. They are large, and what they returned is already in `conclusions` and `messages`.
+
+## Response shape
+
+<CodeGroup>
+```json Response
+{
+  "content": "The user prefers dark roast, usually from local roasters.",
+  "evidence": {
+    "conclusions": [
+      {
+        "id": "hK3mZqPvN2wRtY8bXcLdA",
+        "level": "explicit",
+        "content": "User prefers dark roast coffee",
+        "created_at": "2026-03-20T10:15:00Z",
+        "session_id": "session-xyz",
+        "source_ids": []
+      },
+      {
+        "id": "pR8xLwGtH4vKmN6cZqBfE",
+        "level": "inductive",
+        "content": "User buys from local roasters",
+        "created_at": "2026-03-22T14:30:00Z",
+        "session_id": null,
+        "source_ids": ["hK3mZqPvN2wRtY8bXcLdA"]
+      }
+    ],
+    "messages": [
+      {
+        "id": "m1N2o3P4q5R6s7T8u9V0w",
+        "session_id": "session-xyz",
+        "peer_id": "user-123",
+        "created_at": "2026-03-20T10:12:00Z"
+      }
+    ],
+    "tool_calls": [
+      {"tool_name": "search_memory", "tool_input": {"query": "coffee preference", "top_k": 20}}
+    ],
+    "reasoning_trace_id": null
+  }
+}
+```
+</CodeGroup>
+
+**Conclusions** carry the ID, level, and text of each conclusion read. `source_ids` names the conclusions a derived one was reasoned from, so you can walk a chain back toward the explicit statements at its base; explicit conclusions have none, since they derive from messages rather than from other conclusions. `session_id` is null for a conclusion that was reasoned across sessions and so belongs to none. Timestamps are when a conclusion was derived, taken from its source messages where that is recorded.
+
+**Messages** carry identity and provenance only — no content. Fetch a message by its `id` when you need the text.
+
+That asymmetry with conclusions is deliberate. A conclusion's text is written by the deriver, is short, and *is* the thing you are auditing, so it comes along. A message's content is whatever a caller sent, up to the 25,000-character ingest limit, and one answer can touch a few hundred messages — carrying it would let a single response drag megabytes behind it, and would turn evidence into a way to read messages in bulk. Evidence is for auditing and analytics, not a substitute for the message endpoints.
+
+`reasoning_trace_id` is a placeholder for stored reasoning traces and is currently always null.
+
+## Empty is not absent
+
+The two are different and worth distinguishing:
+
+- `evidence` is **absent or null** — you did not ask for it.
+- `evidence` is **present with empty lists** — you asked, and the agent read nothing. This is the honest answer for a peer with no history yet.
+
+So checking that evidence exists tells you nothing about whether anything was found; check the lists.
+
+## Streaming
+
+Evidence can only be known once the answer is complete, so a streaming response sends it on the stream's final event. The SDKs surface it on the stream object after it has been fully consumed:
+
+<CodeGroup>
+```python Python
+stream = peer.chat_stream(
+    "What coffee does this user prefer?",
+    include_evidence=True,
+)
+
+for chunk in stream:
+    print(chunk, end="", flush=True)
+
+# Available only after the stream has drained
+for conclusion in stream.evidence.conclusions:
+    print(conclusion.id, conclusion.content)
+```
+
+```typescript TypeScript
+const stream = await peer.chatStream(
+  "What coffee does this user prefer?",
+  { includeEvidence: true }
+);
+
+for await (const chunk of stream) {
+  process.stdout.write(chunk);
+}
+
+// Available only after the stream has drained
+for (const conclusion of stream.evidence?.conclusions ?? []) {
+  console.log(conclusion.id, conclusion.content);
+}
+```
+</CodeGroup>
+
+Reading `evidence` mid-stream returns null.
+
+## Workspace chat
+
+Workspace-level chat takes the same option:
+
+<CodeGroup>
+```python Python
+result = honcho.chat(
+    "What do people here have in common?",
+    include_evidence=True,
+)
+print(result.evidence.messages)
+```
+
+```typescript TypeScript
+const { content, evidence } = await honcho.chat(
+  "What do people here have in common?",
+  { includeEvidence: true }
+);
+```
+</CodeGroup>
+
+Because workspace chat opens with a statistical overview of the workspace rather than a conclusion prefetch, its evidence is usually weighted toward messages and whatever its tools went on to find.
+
+## Scoped queries
+
+Evidence never widens what a query could see. It reports only rows a permitted read actually returned, so a query confined by [`scope`](/v3/documentation/features/advanced/scopes) or a session allowlist yields evidence confined the same way — a scope with no member sessions recalls nothing and cites nothing.
+
+## Combining with structured outputs
+
+`include_evidence` and [`response_format`](/v3/documentation/features/advanced/structured-outputs) are independent. With both, the answer is parsed to your schema and the evidence sits beside it:
+
+```python
+result = peer.chat(
+    "What are this user's top 3 food preferences?",
+    response_format=FoodPreferences,
+    include_evidence=True,
+)
+
+result.content      # a FoodPreferences instance
+result.evidence     # what the answer was built from
+```

docs/v3/documentation/features/chat.mdx

@@ -158,6 +158,29 @@ const status = await peer.chat(
 
 The agent runs its full reasoning loop either way — only the final answer is formatted to your schema. See [Structured Outputs](/v3/documentation/features/advanced/structured-outputs) for the supported schema subset, streaming behavior, and best practices.
 
+## Evidence
+
+Pass `include_evidence` to get back the conclusions and messages the agent read while answering, alongside the tools it called:
+
+<CodeGroup>
+```python Python
+result = peer.chat("What coffee does this user prefer?", include_evidence=True)
+
+print(result.content)
+for conclusion in result.evidence.conclusions:
+    print(conclusion.id, conclusion.content)
+```
+
+```typescript TypeScript
+const { content, evidence } = await peer.chat(
+  "What coffee does this user prefer?",
+  { includeEvidence: true }
+);
+```
+</CodeGroup>
+
+Evidence is collated from what the agent read rather than reported by the model, so it lists everything that was available to the answer rather than only what the answer used. See [Evidence](/v3/documentation/features/advanced/evidence) for the full shape, streaming behavior, and what the trade-off costs you.
+
 ## Integration Patterns
 
 ### Dynamic Prompt Enhancement

src/dialectic/chat.py

@@ -17,6 +17,7 @@ from src.dialectic.core import DialecticAgent
 from src.dialectic.workspace import WorkspaceDialecticAgent
 from src.exceptions import ValidationException
 from src.utils.config_helpers import get_configuration
+from src.utils.evidence import EvidenceAccumulator
 from src.utils.scopes import is_scope_peer
 
 logger = logging.getLogger(__name__)
@@ -55,6 +56,7 @@ async def agentic_chat(
     reasoning_level: ReasoningLevel = "low",
     session_allowlist: list[str] | None = None,
     response_model: type[BaseModel] | None = None,
+    evidence: EvidenceAccumulator | None = None,
 ) -> str:
     """
     Answer a query about a peer using the agentic dialectic.
@@ -121,6 +123,7 @@ async def agentic_chat(
         observed_peer_card=observed_peer_card,
         reasoning_level=reasoning_level,
         session_allowlist=session_allowlist,
+        evidence=evidence,
     )
 
     return await agent.answer(query, response_model=response_model)
@@ -135,6 +138,7 @@ async def agentic_chat_stream(
     reasoning_level: ReasoningLevel = "low",
     session_allowlist: list[str] | None = None,
     response_model: type[BaseModel] | None = None,
+    evidence: EvidenceAccumulator | None = None,
 ) -> AsyncIterator[str]:
     """
     Stream an answer to a query about a peer using the agentic dialectic.
@@ -202,6 +206,7 @@ async def agentic_chat_stream(
         observed_peer_card=observed_peer_card,
         reasoning_level=reasoning_level,
         session_allowlist=session_allowlist,
+        evidence=evidence,
     )
 
     async for chunk in agent.answer_stream(query, response_model=response_model):
@@ -214,6 +219,7 @@ async def workspace_chat(
     query: str,
     reasoning_level: ReasoningLevel = "low",
     response_model: type[BaseModel] | None = None,
+    evidence: EvidenceAccumulator | None = None,
     session_allowlist: list[str] | None = None,
 ) -> str:
     """Answer a query across all peers in a workspace."""
@@ -233,6 +239,7 @@ async def workspace_chat(
         session_id=session_id,
         reasoning_level=reasoning_level,
         session_allowlist=session_allowlist,
+        evidence=evidence,
     )
     return await agent.answer(query, response_model=response_model)
 
@@ -243,6 +250,7 @@ async def workspace_chat_stream(
     query: str,
     reasoning_level: ReasoningLevel = "low",
     response_model: type[BaseModel] | None = None,
+    evidence: EvidenceAccumulator | None = None,
     session_allowlist: list[str] | None = None,
 ) -> AsyncIterator[str]:
     """Streaming variant of :func:`workspace_chat`."""
@@ -261,6 +269,7 @@ async def workspace_chat_stream(
         session_id=session_id,
         reasoning_level=reasoning_level,
         session_allowlist=session_allowlist,
+        evidence=evidence,
     )
     async for chunk in agent.answer_stream(query, response_model=response_model):
         yield chunk

src/dialectic/core.py

@@ -13,7 +13,7 @@ from typing import Any, cast
 from nanoid import generate as generate_nanoid
 from pydantic import BaseModel
 
-from src import crud
+from src import crud, models
 from src.config import (
     ConfiguredModelSettings,
     DialecticLevelSettings,
@@ -43,6 +43,7 @@ from src.utils.agent_tools import (
     create_tool_executor,
     search_memory,
 )
+from src.utils.evidence import EvidenceAccumulator
 from src.utils.formatting import format_new_turn_with_timestamp
 from src.utils.types import embedding_call_purpose
 
@@ -76,6 +77,7 @@ class DialecticAgent:
         reasoning_level: ReasoningLevel = "low",
         session_id: str | None = None,
         session_allowlist: list[str] | None = None,
+        evidence: EvidenceAccumulator | None = None,
     ):
         """
         Initialize the dialectic agent.
@@ -93,6 +95,9 @@ class DialecticAgent:
             session_allowlist: Optional session allowlist restricting all recall
                 (conclusions and messages) to these sessions; empty list
                 fails closed
+            evidence: Optional accumulator collecting the conclusions and
+                messages this run reads, for callers that asked for evidence.
+                Passing None collects nothing.
         """
         self.workspace_name: str = workspace_name
         self.session_name: str | None = session_name
@@ -124,6 +129,7 @@ class DialecticAgent:
         ]
         self._session_history_initialized: bool = False
         self._prefetched_conclusion_count: int = 0
+        self.evidence: EvidenceAccumulator | None = evidence
         self._run_id: str = generate_nanoid()  # Always generate for event correlation
 
     def _select_tools(self) -> list[dict[str, Any]]:
@@ -237,6 +243,13 @@ class DialecticAgent:
             ):
                 query_embedding = await embedding_client.embed(query)
 
+            # Prefetched conclusions never pass through the tool executor, so
+            # they are recorded here or not at all -- and on a query that
+            # answers without a tool call they are the whole of what was read.
+            prefetched: list[models.Document] | None = (
+                [] if self.evidence is not None else None
+            )
+
             # search_memory manages its own short-lived DB sessions so no
             # connection is held during external vector-store calls.
             explicit_repr = await search_memory(
@@ -248,6 +261,7 @@ class DialecticAgent:
                 levels=["explicit"],
                 embedding=query_embedding,
                 session_allowlist=self.session_allowlist,
+                documents_out=prefetched,
             )
 
             derived_repr = await search_memory(
@@ -259,6 +273,7 @@ class DialecticAgent:
                 levels=["deductive", "inductive", "contradiction"],
                 embedding=query_embedding,
                 session_allowlist=self.session_allowlist,
+                documents_out=prefetched,
             )
 
             if explicit_repr.is_empty() and derived_repr.is_empty():
@@ -280,6 +295,12 @@ class DialecticAgent:
                 # Include IDs for derived so agent can use get_reasoning_chain
                 parts.append(derived_repr.format_as_markdown(include_ids=True))
 
+            # Recorded last: everything above can still fail into the handler
+            # below, which drops the whole block from the prompt. Evidence should
+            # name what the agent saw, not what was fetched for it.
+            if self.evidence is not None and prefetched:
+                self.evidence.add_documents(prefetched)
+
             return "\n".join(parts)
 
         except Exception as e:
@@ -361,6 +382,7 @@ class DialecticAgent:
             run_id=self._run_id,
             agent_type="dialectic",
             parent_category="dialectic",
+            evidence=self.evidence,
         )
 
     def _prefetch_heading(self) -> str:
@@ -541,6 +563,9 @@ class DialecticAgent:
         if isinstance(content, BaseModel):
             content = content.model_dump_json(by_alias=True)
 
+        if self.evidence is not None:
+            self.evidence.record_tool_calls(response.tool_calls_made)
+
         self._log_response_metrics(
             task_name=task_name,
             run_id=run_id,
@@ -617,6 +642,9 @@ class DialecticAgent:
                 accumulated_content.append(chunk.content)
                 yield chunk.content
 
+        if self.evidence is not None:
+            self.evidence.record_tool_calls(response.tool_calls_made)
+
         self._log_response_metrics(
             task_name=task_name,
             run_id=run_id,

src/dialectic/workspace.py

@@ -31,6 +31,7 @@ from src.utils.agent_tools import (
     create_workspace_tool_executor,
     format_workspace_stats,
 )
+from src.utils.evidence import EvidenceAccumulator
 
 logger = logging.getLogger(__name__)
 
@@ -51,6 +52,7 @@ class WorkspaceDialecticAgent(DialecticAgent):
         reasoning_level: ReasoningLevel = "low",
         session_id: str | None = None,
         session_allowlist: list[str] | None = None,
+        evidence: EvidenceAccumulator | None = None,
     ) -> None:
         super().__init__(
             workspace_name=workspace_name,
@@ -61,6 +63,7 @@ class WorkspaceDialecticAgent(DialecticAgent):
             reasoning_level=reasoning_level,
             session_id=session_id,
             session_allowlist=session_allowlist,
+            evidence=evidence,
         )
         # Replace the pair-oriented system prompt with the workspace one.
         self.messages[0] = {
@@ -194,6 +197,7 @@ class WorkspaceDialecticAgent(DialecticAgent):
             run_id=self._run_id,
             agent_type="workspace_dialectic",
             parent_category="dialectic",
+            evidence=self.evidence,
         )
 
     # Workspace chat shares the base "dialectic_chat" Langfuse trace name;

src/routers/peers.py

@@ -1,8 +1,6 @@
 """FastAPI routes for peer resources and peer-scoped operations."""
 
-import json
 import logging
-from collections.abc import AsyncIterator
 from contextlib import suppress
 from time import perf_counter
 
@@ -28,6 +26,7 @@ from src.exceptions import (
 from src.security import JWTParams, require_auth
 from src.telemetry import prometheus_metrics
 from src.telemetry.events import EmbeddingCallPurpose, GetContextEvent, emit
+from src.utils.evidence import EvidenceAccumulator
 from src.utils.filter import MAX_SESSION_ALLOWLIST_ENTRIES, extract_session_allowlist
 from src.utils.schema_conversion import json_response_schema_to_pydantic
 from src.utils.scopes import (
@@ -37,6 +36,7 @@ from src.utils.scopes import (
     validate_scope_read_option,
 )
 from src.utils.search import search
+from src.utils.sse import format_dialectic_sse_stream
 from src.utils.types import embedding_call_purpose
 
 logger = logging.getLogger(__name__)
@@ -358,17 +358,9 @@ async def chat(
         await peer_db.commit()
     await peers_result.post_commit()
 
+    evidence = EvidenceAccumulator() if options.include_evidence else None
+
     if options.stream:
-        # Stream the response using Server-Sent Events
-
-        async def format_sse_stream(
-            chunks: AsyncIterator[str],
-        ) -> AsyncIterator[str]:
-            """Format chunks as SSE events."""
-            async for chunk in chunks:
-                yield f"data: {json.dumps({'delta': {'content': chunk}, 'done': False})}\n\n"
-            yield f"data: {json.dumps({'done': True})}\n\n"
-
         # Prometheus metrics
         if settings.METRICS.ENABLED:
             prometheus_metrics.record_dialectic_call(
@@ -377,7 +369,7 @@ async def chat(
             )
 
         return StreamingResponse(
-            format_sse_stream(
+            format_dialectic_sse_stream(
                 agentic_chat_stream(
                     workspace_name=workspace_id,
                     session_name=options.session_id,
@@ -387,7 +379,9 @@ async def chat(
                     reasoning_level=options.reasoning_level,
                     session_allowlist=session_allowlist,
                     response_model=response_model,
-                )
+                    evidence=evidence,
+                ),
+                evidence,
             ),
             media_type="text/event-stream",
         )
@@ -404,6 +398,7 @@ async def chat(
         reasoning_level=options.reasoning_level,
         session_allowlist=session_allowlist,
         response_model=response_model,
+        evidence=evidence,
     )
 
     # Prometheus metrics
@@ -413,7 +408,10 @@ async def chat(
             reasoning_level=options.reasoning_level,
         )
 
-    return schemas.DialecticResponse(content=response if response else None)
+    return schemas.DialecticResponse(
+        content=response if response else None,
+        evidence=evidence.build() if evidence is not None else None,
+    )
 
 
 @router.post(

src/routers/workspaces.py

@@ -1,8 +1,6 @@
 """FastAPI routes for workspace resources and workspace-scoped operations."""
 
-import json
 import logging
-from collections.abc import AsyncIterator
 
 from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Response
 from fastapi.responses import StreamingResponse
@@ -20,10 +18,12 @@ from src.dialectic.chat import workspace_chat, workspace_chat_stream
 from src.exceptions import AuthenticationException, ValidationException
 from src.security import JWTParams, require_auth
 from src.telemetry import prometheus_metrics
+from src.utils.evidence import EvidenceAccumulator
 from src.utils.filter import MAX_SESSION_ALLOWLIST_ENTRIES
 from src.utils.schema_conversion import json_response_schema_to_pydantic
 from src.utils.scopes import validate_scope_read_option
 from src.utils.search import search
+from src.utils.sse import format_dialectic_sse_stream
 
 logger = logging.getLogger(__name__)
 
@@ -343,16 +343,11 @@ async def chat(
             reasoning_level=options.reasoning_level,
         )
 
+    evidence = EvidenceAccumulator() if options.include_evidence else None
+
     if options.stream:
-
-        async def format_sse_stream(chunks: AsyncIterator[str]) -> AsyncIterator[str]:
-            """Format chunks as SSE events."""
-            async for chunk in chunks:
-                yield f"data: {json.dumps({'delta': {'content': chunk}, 'done': False})}\n\n"
-            yield f"data: {json.dumps({'done': True})}\n\n"
-
         return StreamingResponse(
-            format_sse_stream(
+            format_dialectic_sse_stream(
                 workspace_chat_stream(
                     workspace_name=workspace_id,
                     session_name=options.session_id,
@@ -360,7 +355,9 @@ async def chat(
                     reasoning_level=options.reasoning_level,
                     response_model=response_model,
                     session_allowlist=session_allowlist,
-                )
+                    evidence=evidence,
+                ),
+                evidence,
             ),
             media_type="text/event-stream",
         )
@@ -372,5 +369,9 @@ async def chat(
         reasoning_level=options.reasoning_level,
         response_model=response_model,
         session_allowlist=session_allowlist,
+        evidence=evidence,
+    )
+    return schemas.DialecticResponse(
+        content=response if response else None,
+        evidence=evidence.build() if evidence is not None else None,
     )
-    return schemas.DialecticResponse(content=response if response else None)

src/schemas/__init__.py

@@ -15,6 +15,10 @@ from src.schemas.api import (
     DialecticResponse,
     DialecticStreamChunk,
     DialecticStreamDelta,
+    Evidence,
+    EvidenceMessageRef,
+    EvidenceObservation,
+    EvidenceToolCall,
     Message,
     MessageBase,
     MessageBatchCreate,
@@ -118,6 +122,10 @@ __all__ = [
     "DialecticResponse",
     "WorkspaceChatOptions",
     "DialecticStreamChunk",
+    "Evidence",
+    "EvidenceMessageRef",
+    "EvidenceObservation",
+    "EvidenceToolCall",
     "DialecticStreamDelta",
     "Message",
     "MessageBase",

src/schemas/api.py

@@ -736,6 +736,15 @@ class WorkspaceMessageSearchOptions(MessageSearchOptions):
 # ---------------------------------------------------------------------------
 
 
+_INCLUDE_EVIDENCE_DESCRIPTION = (
+    "When true, the response includes an `evidence` object listing the"
+    " conclusions and messages the agent read while answering, plus the tool"
+    " calls it made. Evidence is collated from what the agent accessed; the"
+    " model is never asked to cite anything, so evidence may over-report"
+    " (accessed is not the same as used)."
+)
+
+
 class DialecticOptions(BaseModel):
     session_id: str | None = Field(
         None, description="ID of the session to scope the representation to"
@@ -787,6 +796,9 @@ class DialecticOptions(BaseModel):
             " maxLength, ...) are hints to the model, not enforced server-side."
         ),
     )
+    include_evidence: bool = Field(
+        default=False, description=_INCLUDE_EVIDENCE_DESCRIPTION
+    )
 
 
 class WorkspaceChatOptions(BaseModel):
@@ -821,10 +833,108 @@ class WorkspaceChatOptions(BaseModel):
             "with `session_id`. Requires a workspace- or admin-level key."
         ),
     )
+    include_evidence: bool = Field(
+        default=False, description=_INCLUDE_EVIDENCE_DESCRIPTION
+    )
+
+
+class EvidenceObservation(BaseModel):
+    """A conclusion the dialectic agent read while answering."""
+
+    id: str = Field(description="Conclusion (document) ID")
+    level: DocumentLevel = Field(
+        description="Conclusion level: explicit, deductive, inductive, or contradiction"
+    )
+    content: str = Field(
+        description="The conclusion text (the derived conclusion, for non-explicit levels)"
+    )
+    created_at: datetime.datetime = Field(
+        description="When the conclusion was derived, from its source messages when known"
+    )
+    session_id: str | None = Field(
+        default=None, description="Session the conclusion is scoped to, if any"
+    )
+    source_ids: list[str] = Field(
+        default_factory=list,
+        description=(
+            "IDs of the conclusions this one was derived from. Empty for explicit"
+            " conclusions, which derive from messages rather than from other"
+            " conclusions."
+        ),
+    )
+
+
+class EvidenceMessageRef(BaseModel):
+    """A message the dialectic agent read while answering.
+
+    Identity and provenance only -- no content. Message content is
+    caller-supplied and unbounded, so carrying it would let one answer drag
+    megabytes behind it, and would invite callers to read messages out of
+    evidence in bulk rather than asking for the ones they want. Fetch the
+    message by `id` when the text is needed.
+    """
+
+    id: str = Field(description="Message ID")
+    session_id: str = Field(description="Session the message belongs to")
+    peer_id: str = Field(description="Peer who sent the message")
+    created_at: datetime.datetime = Field(description="When the message was sent")
+
+
+class EvidenceToolCall(BaseModel):
+    """A tool the dialectic agent invoked while answering."""
+
+    tool_name: str = Field(description="Name of the tool")
+    tool_input: dict[str, Any] = Field(
+        default_factory=dict, description="Arguments the agent passed to the tool"
+    )
+
+
+class Evidence(BaseModel):
+    """What the dialectic agent read and did while answering.
+
+    Collated from the agent's own reads rather than reported by the model, so
+    it is deterministic but over-reports: it lists what the agent accessed,
+    which is not necessarily what the answer relied on.
+
+    Meant for auditing and analytics -- inspecting why an answer looks the way
+    it does, or measuring what recall actually reaches the agent. It is not a
+    read API: conclusions carry their text because that text is the thing being
+    audited and the deriver keeps it short, while messages carry identity alone
+    (see `EvidenceMessageRef`).
+    """
+
+    conclusions: list[EvidenceObservation] = Field(
+        default_factory=list,
+        description="Conclusions the agent read, whether prefetched or found via its tools",
+    )
+    messages: list[EvidenceMessageRef] = Field(
+        default_factory=list,
+        description=(
+            "Messages the agent read via its search and grep tools, by ID and"
+            " provenance only. Fetch a message to read its content."
+        ),
+    )
+    tool_calls: list[EvidenceToolCall] = Field(
+        default_factory=list,
+        description=(
+            "Tools the agent invoked, in order, with their arguments. Results are"
+            " omitted (they are reflected in `conclusions` and `messages`), and so"
+            " are calls that failed, so this is a record of successful invocations"
+            " rather than a complete reasoning trace."
+        ),
+    )
+    reasoning_trace_id: str | None = Field(
+        default=None,
+        description="ID of the stored reasoning trace for this call, when trace storage is enabled",
+    )
 
 
 class DialecticResponse(BaseModel):
     content: str | None
+    evidence: Evidence | None = Field(
+        default=None,
+        description="What the answer was built from. Present only when `include_evidence` is true.",
+    )
 
 
 class DialecticStreamDelta(BaseModel):
@@ -842,6 +952,13 @@ class DialecticStreamChunk(BaseModel):
 
     delta: DialecticStreamDelta
     done: bool = False
+    evidence: Evidence | None = Field(
+        default=None,
+        description=(
+            "What the answer was built from. Set only on the final chunk"
+            " (`done` is true) and only when `include_evidence` is true."
+        ),
+    )
 
 
 # ---------------------------------------------------------------------------

src/utils/agent_tools.py

@@ -25,6 +25,7 @@ from src.telemetry.events import (
     emit,
 )
 from src.utils import summarizer
+from src.utils.evidence import EvidenceAccumulator
 from src.utils.formatting import (
     format_datetime_utc,
     format_new_turn_with_timestamp,
@@ -1206,6 +1207,7 @@ async def search_memory(
     levels: list[str] | None = None,
     embedding: list[float] | None = None,
     session_allowlist: list[str] | None = None,
+    documents_out: list[models.Document] | None = None,
 ) -> Representation:
     """
     Search for observations in memory using semantic similarity.
@@ -1222,6 +1224,9 @@ async def search_memory(
         levels: Optional list of observation levels to filter by
                 (e.g., ["explicit"], ["deductive", "inductive", "contradiction"])
         embedding: Optional pre-computed embedding to avoid redundant API calls
+        documents_out: Optional list the matched documents are appended to, for
+                callers that need the rows and not just the representation
+                built from them
 
     Returns:
         Representation object containing relevant observations
@@ -1254,6 +1259,9 @@ async def search_memory(
         embedding=embedding,
     )
 
+    if documents_out is not None:
+        documents_out.extend(documents)
+
     return Representation.from_documents(documents)
 
 
@@ -1464,6 +1472,10 @@ class ToolContext:
     run_id: str | None = None
     agent_type: str | None = None  # "dialectic", "deriver", "dreamer"
     parent_category: str | None = None  # Parent category for CloudEvents
+    # Set only when the caller asked for evidence. Read handlers append the rows
+    # they loaded; `dataclasses.replace` copies of this context share the same
+    # accumulator, so delegating handlers reach it without extra wiring.
+    evidence: EvidenceAccumulator | None = None
 
 
 def _normalize_observation_id(obs_id: str) -> str:
@@ -1897,6 +1909,36 @@ async def _handle_get_recent_history(
     return _maybe_truncated_result(output)
 
 
+def _record_conclusion_evidence(
+    ctx: ToolContext, documents: Sequence[models.Document]
+) -> None:
+    """Record conclusions a read returned, when evidence was asked for."""
+    if ctx.evidence is not None:
+        ctx.evidence.add_documents(documents)
+
+
+def _record_message_evidence(
+    ctx: ToolContext, messages: Sequence[models.Message]
+) -> None:
+    """Record messages a read returned, when evidence was asked for."""
+    if ctx.evidence is not None:
+        ctx.evidence.add_messages(messages)
+
+
+def _record_snippet_evidence(
+    ctx: ToolContext,
+    snippets: Sequence[tuple[list[models.Message], list[models.Message]]],
+) -> None:
+    """Record every message a snippet search surfaced.
+
+    Both halves of a snippet count as read: the surrounding context reaches the
+    prompt the same way the matches do.
+    """
+    for matches, context in snippets:
+        _record_message_evidence(ctx, matches)
+        _record_message_evidence(ctx, context)
+
+
 async def _handle_search_memory(
     ctx: ToolContext, tool_input: dict[str, Any]
 ) -> "str | ToolResult":
@@ -1953,6 +1995,7 @@ async def _handle_search_memory(
             if ctx.session_allowlist is not None
             else None,
         )
+    _record_conclusion_evidence(ctx, documents)
     mem = Representation.from_documents(documents)
     total_count = mem.len()
     if total_count == 0:
@@ -1976,6 +2019,7 @@ async def _handle_search_memory(
                 observer=ctx.observer,
                 session_allowlist=ctx.session_allowlist,
             )
+            _record_snippet_evidence(ctx, snippets)
             if snippets:
                 message_output = _format_message_snippets(
                     snippets, f"for query '{query}'"
@@ -2018,6 +2062,7 @@ async def _handle_get_observation_context(
             observer=ctx.observer or None,
             session_allowlist=ctx.session_allowlist,
         )
+        _record_message_evidence(ctx, messages)
         if not messages:
             return f"No messages found for IDs {tool_input['message_ids']}"
         messages_text = "\n".join(
@@ -2061,6 +2106,7 @@ async def _handle_search_messages(
         observer=ctx.observer or None,
         session_allowlist=ctx.session_allowlist,
     )
+    _record_snippet_evidence(ctx, snippets)
     search_meta: dict[str, Any] = {
         "top_k": limit,
         "used_embedding": True,
@@ -2096,6 +2142,7 @@ async def _handle_grep_messages(
         observer=ctx.observer or None,
         session_allowlist=ctx.session_allowlist,
     )
+    _record_snippet_evidence(ctx, snippets)
     if not snippets:
         return f"No messages found containing '{text}'"
 
@@ -2161,6 +2208,7 @@ async def _handle_get_messages_by_date_range(
             observer=ctx.observer or None,
             session_allowlist=ctx.session_allowlist,
         )
+        _record_message_evidence(ctx, messages)
         msg_count = len(messages)
         messages_text = (
             "\n".join(
@@ -2236,6 +2284,7 @@ async def _handle_search_messages_temporal(
         embedding=query_embedding,
         observer=ctx.observer or None,
     )
+    _record_snippet_evidence(ctx, snippets)
     date_filter: list[str] = []
     if after_date_str:
         date_filter.append(f"after {after_date_str}")
@@ -2523,6 +2572,7 @@ async def _handle_get_reasoning_chain(
             return f"ERROR: Observation '{observation_id}' not found"
 
         doc: Document = docs[0]
+        _record_conclusion_evidence(ctx, [doc])
 
         output_parts: list[str] = []
 
@@ -2536,6 +2586,7 @@ async def _handle_get_reasoning_chain(
                 premises = await crud.get_documents_by_ids(
                     db, ctx.workspace_name, doc.source_ids
                 )
+                _record_conclusion_evidence(ctx, premises)
                 if premises:
                     premise_lines: list[Any] = []
                     for p in premises:
@@ -2553,6 +2604,7 @@ async def _handle_get_reasoning_chain(
                 sources = await crud.get_documents_by_ids(
                     db, ctx.workspace_name, doc.source_ids
                 )
+                _record_conclusion_evidence(ctx, sources)
                 if sources:
                     source_lines: list[Any] = []
                     for s in sources:
@@ -2581,6 +2633,7 @@ async def _handle_get_reasoning_chain(
                 observer=ctx.observer,
                 observed=ctx.observed,
             )
+            _record_conclusion_evidence(ctx, children)
             if children:
                 child_lines: list[Any] = []
                 for c in children:
@@ -2634,6 +2687,7 @@ async def create_tool_executor(
     parent_category: str | None = None,
     session_allowlist: list[str] | None = None,
     handler_resolver: Callable[[str], Any] | None = None,
+    evidence: EvidenceAccumulator | None = None,
 ) -> Callable[[str, dict[str, Any]], Any]:
     """
     Create a unified tool executor function for all agent operations.
@@ -2661,6 +2715,9 @@ async def create_tool_executor(
         handler_resolver: Optional callback that replaces the default
             handler-table lookup for resolving tool names to handlers.
             Returning None takes the "Unknown tool" path.
+        evidence: Optional accumulator that read handlers record the
+            conclusions and messages they load into. None means evidence was
+            not requested and nothing is collected.
 
     Returns:
         An async callable that executes tools with the captured context
@@ -2683,6 +2740,7 @@ async def create_tool_executor(
         run_id=run_id,
         agent_type=agent_type,
         parent_category=parent_category,
+        evidence=evidence,
     )
 
     async def execute_tool(tool_name: str, tool_input: dict[str, Any]) -> str:
@@ -3077,6 +3135,7 @@ async def create_workspace_tool_executor(
     run_id: str | None = None,
     agent_type: str | None = None,
     parent_category: str | None = None,
+    evidence: EvidenceAccumulator | None = None,
 ) -> Callable[[str, dict[str, Any]], Any]:
     """Tool executor for workspace-level operations (no bound peer pair).
 
@@ -3099,4 +3158,5 @@ async def create_workspace_tool_executor(
         agent_type=agent_type,
         parent_category=parent_category,
         handler_resolver=_workspace_handler_resolver,
+        evidence=evidence,
     )

src/utils/evidence.py

@@ -0,0 +1,203 @@
+"""Collation of what the dialectic agent read while answering.
+
+Evidence is gathered passively: read paths hand the rows they already loaded to
+an accumulator, and nothing is re-queried when the response is built. The model
+is never asked to cite its sources, which keeps collection deterministic and
+free of model load, at the cost of over-reporting -- the agent may read a
+conclusion it does not end up leaning on.
+
+Because nothing is re-queried, evidence inherits the scoping of the reads that
+produced it: whatever the workspace, observer/observed pair and session
+allowlist permitted the agent to see is exactly what can appear here.
+
+What it is for shapes what it carries. Evidence is an audit and analytics
+surface -- for asking why an answer looks the way it does, or measuring what
+recall reaches the agent -- not a bulk read API. So conclusions carry their
+text, which is short, model-written, and the thing being audited, while
+messages carry identity alone: message content is caller-supplied and
+unbounded, and including it would both inflate every response and invite
+callers to read messages out of evidence instead of asking for the ones they
+want.
+"""
+
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import Any, cast
+
+from src import models
+from src.schemas.api import (
+    Evidence,
+    EvidenceMessageRef,
+    EvidenceObservation,
+    EvidenceToolCall,
+)
+from src.utils.representation import (
+    ContradictionObservation,
+    DeductiveObservation,
+    ExplicitObservation,
+    InductiveObservation,
+    Representation,
+)
+from src.utils.types import DocumentLevel
+
+# The four shapes `Representation` sorts observations into. They agree on
+# identity and timing but disagree on where the text lives and whether there
+# are source conclusions, which is what the two helpers below reconcile.
+_Observation = (
+    ExplicitObservation
+    | DeductiveObservation
+    | InductiveObservation
+    | ContradictionObservation
+)
+
+
+def _observation_text(observation: _Observation) -> str:
+    """Read an observation's text, whatever its level calls it.
+
+    A derived observation reached its text by reasoning, so it is a
+    `conclusion`; an explicit or contradiction observation just has `content`.
+    """
+    if isinstance(observation, DeductiveObservation | InductiveObservation):
+        return observation.conclusion
+    return observation.content
+
+
+def _observation_source_ids(observation: _Observation) -> list[str]:
+    """Read the conclusions an observation was derived from.
+
+    Explicit observations derive from messages rather than from other
+    conclusions, so they have no source ids to report.
+    """
+    if isinstance(observation, ExplicitObservation):
+        return []
+    return observation.source_ids
+
+
+def _restore_utc_marker(timestamp: datetime) -> datetime:
+    """Put back the tzinfo a representation timestamp dropped.
+
+    `Representation` renders observations into prompts, so it strips tzinfo to
+    keep those timestamps short. The underlying columns are stored UTC, so
+    naming UTC again recovers what was dropped rather than shifting the
+    instant. This repairs that one lossy step; it is not an offset conversion,
+    and an already-aware timestamp is left as it is.
+    """
+    return timestamp if timestamp.tzinfo is not None else timestamp.replace(tzinfo=UTC)
+
+
+def _tool_call_from_log_entry(entry: dict[str, Any]) -> EvidenceToolCall:
+    """Read one entry of a tool loop's call log.
+
+    The log carries two shapes depending on where it came from: the accumulated
+    loop history uses ``tool_name``/``tool_input``, a raw provider response uses
+    ``name``/``input``. Accept either. Results are deliberately dropped -- they
+    are already reflected in the conclusions and messages, and they are large.
+    """
+    name = entry.get("tool_name") or entry.get("name") or ""
+    raw_input = entry.get("tool_input")
+    if raw_input is None:
+        raw_input = entry.get("input")
+    return EvidenceToolCall(
+        tool_name=str(name),
+        tool_input=cast("dict[str, Any]", raw_input)
+        if isinstance(raw_input, dict)
+        else {},
+    )
+
+
+@dataclass
+class EvidenceAccumulator:
+    """Collects the rows a dialectic agent reads, for one chat call.
+
+    Created by the router when the caller asks for evidence and threaded down
+    through the agent into ``ToolContext``. Tool handlers reach it through the
+    context, which they copy with ``dataclasses.replace`` -- a shallow copy, so
+    every copy appends to this same instance.
+
+    Conclusions and messages are keyed by ID so that a row two tools both
+    returned is recorded once. Keying on ID rather than content matters:
+    ``Representation``'s own deduplication ignores IDs, which would collapse
+    distinct conclusions that happen to read the same.
+    """
+
+    conclusions: dict[str, models.Document] = field(default_factory=dict)
+    messages: dict[str, models.Message] = field(default_factory=dict)
+    tool_calls: list[EvidenceToolCall] = field(default_factory=list)
+
+    def add_documents(self, documents: Iterable[models.Document]) -> None:
+        """Record conclusions a read path returned."""
+        for document in documents:
+            self.conclusions.setdefault(document.id, document)
+
+    def add_messages(self, messages: Iterable[models.Message]) -> None:
+        """Record messages a read path returned."""
+        for message in messages:
+            self.messages.setdefault(message.public_id, message)
+
+    def record_tool_calls(self, tool_calls_made: Sequence[dict[str, Any]]) -> None:
+        """Replace the tool call log with a completed loop's history.
+
+        The tool loop rewrites its log wholesale on every exit path, so this
+        overwrites rather than appends. Failed calls never reach the log, so the
+        result records successful invocations only.
+        """
+        self.tool_calls = [
+            _tool_call_from_log_entry(entry) for entry in tool_calls_made
+        ]
+
+    def build(self) -> Evidence:
+        """Flatten what was collected into the API shape."""
+        return Evidence(
+            conclusions=_flatten_conclusions(self.conclusions.values()),
+            messages=sorted(
+                (
+                    EvidenceMessageRef(
+                        id=message.public_id,
+                        session_id=message.session_name,
+                        peer_id=message.peer_name,
+                        created_at=message.created_at,
+                    )
+                    for message in self.messages.values()
+                ),
+                key=lambda ref: (ref.created_at, ref.id),
+            ),
+            tool_calls=list(self.tool_calls),
+        )
+
+
+def _flatten_conclusions(
+    documents: Iterable[models.Document],
+) -> list[EvidenceObservation]:
+    """Turn conclusion rows into a flat, level-tagged list.
+
+    Goes through ``Representation.from_documents`` rather than reading the rows
+    directly so that evidence resolves ``source_ids`` and derivation timestamps
+    the same way every other reader of a conclusion does -- both have fallbacks
+    for older rows that are easy to get wrong twice.
+
+    An observation's ``message_ids`` are dropped: they are internal row ids,
+    and Honcho identifies messages by their public id everywhere it faces a
+    caller.
+    """
+    representation = Representation.from_documents(list(documents))
+    by_level: tuple[tuple[DocumentLevel, Sequence[_Observation]], ...] = (
+        ("explicit", representation.explicit),
+        ("deductive", representation.deductive),
+        ("inductive", representation.inductive),
+        ("contradiction", representation.contradiction),
+    )
+    observations = [
+        EvidenceObservation(
+            id=observation.id,
+            level=level,
+            content=_observation_text(observation),
+            created_at=_restore_utc_marker(observation.created_at),
+            session_id=observation.session_name,
+            source_ids=_observation_source_ids(observation),
+        )
+        for level, observations_at_level in by_level
+        for observation in observations_at_level
+    ]
+    observations.sort(key=lambda observation: (observation.created_at, observation.id))
+    return observations

src/utils/sse.py

@@ -0,0 +1,36 @@
+"""Server-sent event framing for streamed dialectic answers.
+
+Shared by the peer and workspace chat routes so the two stay in step; the
+frames they emit are part of the public API and the SDKs parse them.
+"""
+
+import json
+from collections.abc import AsyncIterator
+
+from src.schemas.api import Evidence
+from src.utils.evidence import EvidenceAccumulator
+
+
+async def format_dialectic_sse_stream(
+    chunks: AsyncIterator[str],
+    evidence: EvidenceAccumulator | None = None,
+) -> AsyncIterator[str]:
+    """Frame answer chunks as SSE events, then a terminal event.
+
+    Evidence can only be known once the answer has finished streaming, so it
+    rides on the terminal event rather than a frame of its own. `evidence` is
+    the accumulator the agent filled in while answering; None means the caller
+    did not ask for evidence and the terminal event carries only `done`.
+    """
+    async for chunk in chunks:
+        yield f"data: {json.dumps({'delta': {'content': chunk}, 'done': False})}\n\n"
+
+    final: dict[str, object] = {"done": True}
+    if evidence is not None:
+        final["evidence"] = _serializable(evidence.build())
+    yield f"data: {json.dumps(final)}\n\n"
+
+
+def _serializable(evidence: Evidence) -> object:
+    """Round-trip through Pydantic's JSON encoder for the datetime fields."""
+    return json.loads(evidence.model_dump_json())

tests/dialectic/test_evidence.py

@@ -0,0 +1,396 @@
+"""Tests that a dialectic run collates what it read.
+
+The accumulator is unit-tested in tests/utils/test_evidence.py and the tool
+handlers in tests/utils/test_agent_tools.py. What these add is the agent: a
+real run against a real database, with only the LLM call mocked, so that the
+paths outside the tool loop are covered too.
+"""
+
+from collections.abc import AsyncIterator, Mapping
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from nanoid import generate as generate_nanoid
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src import crud, models, schemas
+from src.config import settings
+from src.dialectic.core import DialecticAgent
+from src.dialectic.workspace import WorkspaceDialecticAgent
+from src.llm import (
+    HonchoLLMCallResponse,
+    HonchoLLMCallStreamChunk,
+    StreamingResponseWithMetadata,
+)
+from src.utils.evidence import EvidenceAccumulator
+from src.utils.representation import Representation
+
+
+def llm_call_kwargs(mock_llm_call: AsyncMock) -> Mapping[str, Any]:
+    """The arguments of the last LLM call, once one has actually happened."""
+    assert mock_llm_call.await_args is not None, "the agent never called the LLM"
+    return mock_llm_call.await_args.kwargs
+
+
+def make_llm_response(
+    tool_calls_made: list[dict[str, Any]] | None = None,
+) -> HonchoLLMCallResponse[str]:
+    return HonchoLLMCallResponse(
+        content="The user drinks coffee.",
+        input_tokens=10,
+        output_tokens=5,
+        finish_reasons=["end_turn"],
+        tool_calls_made=tool_calls_made or [],
+    )
+
+
+@pytest.fixture
+async def dialectic_test_data(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+    monkeypatch: pytest.MonkeyPatch,
+) -> Any:
+    """A peer with conclusions and messages the agent can actually recall.
+
+    Returns (workspace, observer, observed, session, messages, documents).
+    """
+    # The documents are written straight to postgres, so recall has to go
+    # through pgvector rather than the migrated vector store.
+    monkeypatch.setattr(settings.VECTOR_STORE, "MIGRATED", False)
+
+    workspace, observer = sample_data
+    observed = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name)
+    db_session.add(observed)
+    await db_session.flush()
+
+    session = (
+        await crud.get_or_create_session(
+            db_session,
+            schemas.SessionCreate(
+                name=str(generate_nanoid()),
+                peers={observed.name: schemas.SessionPeerConfig(observe_me=True)},
+            ),
+            workspace.name,
+        )
+    ).resource
+    db_session.add(
+        models.Collection(
+            workspace_name=workspace.name,
+            observer=observer.name,
+            observed=observed.name,
+        )
+    )
+    await db_session.flush()
+
+    messages: list[models.Message] = []
+    for i, content in enumerate(
+        ["I drink a lot of coffee", "Mornings are my best time"]
+    ):
+        message = models.Message(
+            workspace_name=workspace.name,
+            session_name=session.name,
+            peer_name=observed.name,
+            content=content,
+            seq_in_session=i + 1,
+            token_count=10,
+        )
+        db_session.add(message)
+        messages.append(message)
+    await db_session.flush()
+
+    documents: list[models.Document] = []
+    for i, (content, level) in enumerate(
+        [
+            ("User drinks coffee", "explicit"),
+            ("User is a morning person", "explicit"),
+            ("User drinks coffee in the morning", "deductive"),
+        ]
+    ):
+        document = models.Document(
+            workspace_name=workspace.name,
+            observer=observer.name,
+            observed=observed.name,
+            content=content,
+            embedding=[0.1 * (i + 1)] * 1536,
+            session_name=session.name,
+            level=level,
+        )
+        db_session.add(document)
+        documents.append(document)
+    await db_session.flush()
+
+    for row in (*messages, *documents):
+        await db_session.refresh(row)
+    await db_session.commit()
+
+    return workspace, observer, observed, session, messages, documents
+
+
+def make_agent(
+    dialectic_test_data: Any, evidence: EvidenceAccumulator | None = None, **kwargs: Any
+) -> DialecticAgent:
+    """A dialectic agent pointed at the fixture's peer pair."""
+    _, observer, observed, session, _, _ = dialectic_test_data
+    return DialecticAgent(
+        workspace_name=observer.workspace_name,
+        session_name=session.name,
+        observer=observer.name,
+        observed=observed.name,
+        evidence=evidence,
+        **kwargs,
+    )
+
+
+async def run_answer(
+    agent: DialecticAgent, response: Any = None
+) -> tuple[str, AsyncMock]:
+    """Answer a query with the LLM call mocked out.
+
+    Returns the answer and the mock, so a test can reach the arguments the
+    agent handed the LLM -- the tool executor especially.
+    """
+    mock_llm_call = AsyncMock(return_value=response or make_llm_response())
+    with patch("src.dialectic.core.honcho_llm_call", new=mock_llm_call):
+        answer = await agent.answer("What does the user drink?")
+    return answer, mock_llm_call
+
+
+@pytest.mark.asyncio
+class TestPrefetchEvidence:
+    """Prefetched conclusions never pass through the tool executor.
+
+    The agent reads them before its first LLM call, so on a query that answers
+    without calling a tool they are the whole of what it read. If evidence only
+    watched the tool loop it would look empty on exactly those queries.
+    """
+
+    async def test_records_conclusions_the_prefetch_loaded(
+        self, dialectic_test_data: Any
+    ):
+        *_, documents = dialectic_test_data
+        evidence = EvidenceAccumulator()
+        await run_answer(make_agent(dialectic_test_data, evidence))
+
+        recorded = set(evidence.conclusions)
+        assert recorded, "prefetched conclusions were not recorded"
+        assert recorded <= {document.id for document in documents}
+
+    async def test_prefetch_covers_derived_conclusions_too(
+        self, dialectic_test_data: Any
+    ):
+        """Prefetch searches explicit and derived levels separately."""
+        *_, documents = dialectic_test_data
+        deductive_id = next(d.id for d in documents if d.level == "deductive")
+        evidence = EvidenceAccumulator()
+        await run_answer(make_agent(dialectic_test_data, evidence))
+
+        assert deductive_id in evidence.conclusions
+
+    async def test_built_evidence_reports_the_prefetched_conclusions(
+        self, dialectic_test_data: Any
+    ):
+        """The API shape carries them, not just the accumulator."""
+        evidence = EvidenceAccumulator()
+        await run_answer(make_agent(dialectic_test_data, evidence))
+
+        built = evidence.build()
+        assert {c.content for c in built.conclusions} >= {"User drinks coffee"}
+        assert all(c.id for c in built.conclusions)
+
+    async def test_records_nothing_when_the_prefetch_block_is_dropped(
+        self, dialectic_test_data: Any, monkeypatch: pytest.MonkeyPatch
+    ):
+        """A prefetch that fails after the search reaches the agent with nothing.
+
+        `_prefetch_relevant_observations` swallows any failure and returns
+        None, so `_prepare_query` builds a prompt with no prefetch block. The
+        rows the search returned were never shown to the agent, and evidence
+        has to say so.
+        """
+
+        def explode(*_args: object, **_kwargs: object) -> str:
+            raise RuntimeError("formatting blew up")
+
+        monkeypatch.setattr(Representation, "format_as_markdown", explode)
+        evidence = EvidenceAccumulator()
+        agent = make_agent(dialectic_test_data, evidence)
+
+        await run_answer(agent)
+
+        assert evidence.conclusions == {}
+        # The agent still answered, just without the prefetched context.
+        assert agent.messages[-1]["content"].startswith("Query:")
+        assert "Relevant Observations" not in agent.messages[-1]["content"]
+
+
+@pytest.mark.asyncio
+class TestToolLoopEvidence:
+    async def test_the_agents_tool_executor_records_into_the_same_accumulator(
+        self, dialectic_test_data: Any
+    ):
+        """Capture the executor the agent handed the LLM and drive it.
+
+        The mocked LLM call never invokes tools, so this checks the wiring the
+        way the tool loop would use it.
+        """
+        *_, documents = dialectic_test_data
+        evidence = EvidenceAccumulator()
+        _, mock_llm_call = await run_answer(make_agent(dialectic_test_data, evidence))
+
+        evidence.conclusions.clear()
+        tool_executor = llm_call_kwargs(mock_llm_call)["tool_executor"]
+        await tool_executor("search_memory", {"query": "coffee"})
+
+        assert set(evidence.conclusions) <= {d.id for d in documents}
+        assert evidence.conclusions, "the tool executor did not record anything"
+
+    async def test_records_the_tool_calls_the_loop_made(self, dialectic_test_data: Any):
+        evidence = EvidenceAccumulator()
+        await run_answer(
+            make_agent(dialectic_test_data, evidence),
+            make_llm_response(
+                tool_calls_made=[
+                    {
+                        "tool_name": "search_memory",
+                        "tool_input": {"query": "coffee"},
+                        "tool_result": "Found 3 observations",
+                    }
+                ]
+            ),
+        )
+
+        built = evidence.build()
+        assert [(c.tool_name, c.tool_input) for c in built.tool_calls] == [
+            ("search_memory", {"query": "coffee"})
+        ]
+        assert "Found 3 observations" not in built.model_dump_json()
+
+    async def test_records_the_tool_calls_a_streamed_answer_made(
+        self, dialectic_test_data: Any
+    ):
+        """Evidence is only complete once the stream has drained."""
+        evidence = EvidenceAccumulator()
+        agent = make_agent(dialectic_test_data, evidence)
+
+        async def chunks() -> AsyncIterator[HonchoLLMCallStreamChunk]:
+            for text in ("The user ", "drinks coffee."):
+                yield HonchoLLMCallStreamChunk(content=text, is_done=False)
+            yield HonchoLLMCallStreamChunk(content="", is_done=True)
+
+        streaming = StreamingResponseWithMetadata(
+            chunks(),
+            tool_calls_made=[{"tool_name": "search_messages", "tool_input": {}}],
+            input_tokens=10,
+            output_tokens=5,
+            cache_creation_input_tokens=0,
+            cache_read_input_tokens=0,
+            iterations=1,
+        )
+
+        with patch(
+            "src.dialectic.core.honcho_llm_call",
+            new=AsyncMock(return_value=streaming),
+        ):
+            streamed = [
+                chunk
+                async for chunk in agent.answer_stream("What does the user drink?")
+            ]
+
+        assert "".join(streamed) == "The user drinks coffee."
+        assert [c.tool_name for c in evidence.build().tool_calls] == ["search_messages"]
+
+
+@pytest.mark.asyncio
+class TestEvidenceOptedOut:
+    async def test_collects_nothing_and_answers_the_same(
+        self, dialectic_test_data: Any
+    ):
+        agent = make_agent(dialectic_test_data)
+        mock_llm_call = AsyncMock(return_value=make_llm_response())
+
+        with patch("src.dialectic.core.honcho_llm_call", new=mock_llm_call):
+            answer = await agent.answer("What does the user drink?")
+
+        assert answer == "The user drinks coffee."
+        assert agent.evidence is None
+        tool_executor = llm_call_kwargs(mock_llm_call)["tool_executor"]
+        # The executor still works; it just has nowhere to record.
+        await tool_executor("search_memory", {"query": "coffee"})
+
+
+@pytest.mark.asyncio
+class TestScopedEvidence:
+    async def test_stays_empty_when_recall_fails_closed(self, dialectic_test_data: Any):
+        """An empty session allowlist recalls nothing, so it cites nothing.
+
+        Evidence reports only rows a permitted read returned, so it cannot
+        become a way around the allowlist.
+        """
+        evidence = EvidenceAccumulator()
+        await run_answer(
+            make_agent(dialectic_test_data, evidence, session_allowlist=[])
+        )
+
+        assert evidence.conclusions == {}
+        assert evidence.messages == {}
+
+    async def test_only_reports_conclusions_inside_the_allowlist(
+        self, dialectic_test_data: Any, db_session: AsyncSession
+    ):
+        _, observer, observed, session, _, documents = dialectic_test_data
+        other_session = (
+            await crud.get_or_create_session(
+                db_session,
+                schemas.SessionCreate(name=str(generate_nanoid())),
+                observer.workspace_name,
+            )
+        ).resource
+        outsider = models.Document(
+            workspace_name=observer.workspace_name,
+            observer=observer.name,
+            observed=observed.name,
+            content="User dislikes tea",
+            embedding=[0.9] * 1536,
+            session_name=other_session.name,
+            level="explicit",
+        )
+        db_session.add(outsider)
+        await db_session.flush()
+        await db_session.refresh(outsider)
+        await db_session.commit()
+
+        evidence = EvidenceAccumulator()
+        await run_answer(
+            make_agent(dialectic_test_data, evidence, session_allowlist=[session.name])
+        )
+
+        assert outsider.id not in evidence.conclusions
+        assert set(evidence.conclusions) <= {d.id for d in documents}
+
+
+@pytest.mark.asyncio
+class TestWorkspaceAgentEvidence:
+    async def test_its_tool_executor_records_into_the_same_accumulator(
+        self, dialectic_test_data: Any
+    ):
+        """The workspace agent's prefetch is a stats overview, not conclusions.
+
+        So its evidence has to come from the tool loop, through the workspace
+        executor's delegating handlers.
+        """
+        workspace, _, _, session, messages, _ = dialectic_test_data
+        evidence = EvidenceAccumulator()
+        agent = WorkspaceDialecticAgent(
+            workspace_name=workspace.name,
+            session_name=session.name,
+            evidence=evidence,
+        )
+        _, mock_llm_call = await run_answer(agent)
+
+        assert evidence.conclusions == {}, "the stats prefetch has nothing to cite"
+
+        tool_executor = llm_call_kwargs(mock_llm_call)["tool_executor"]
+        await tool_executor("grep_messages", {"text": "coffee", "context_window": 0})
+
+        assert messages[0].public_id in evidence.messages

tests/routes/test_dialectic_evidence.py

@@ -0,0 +1,283 @@
+"""Tests for the `include_evidence` request option on both chat endpoints.
+
+These cover the request/response contract: whether evidence is asked for,
+whether an accumulator reaches the dialectic, and how the result is
+serialized -- including onto the stream, where evidence can only ride on the
+terminal event.
+
+Note what these deliberately do NOT cover. The autouse
+`mock_llm_call_functions` fixture replaces `agentic_chat` and `workspace_chat`
+wholesale, so nothing here exercises collection; a test that asserted on
+evidence content while the dialectic is mocked would only be reading back its
+own fixture. Collection is covered in tests/dialectic/test_evidence.py and
+tests/utils/test_agent_tools.py.
+"""
+
+import json
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from nanoid import generate as generate_nanoid
+
+from src.models import Peer, Workspace
+from src.utils.evidence import EvidenceAccumulator
+
+TOOL_CALL = {"tool_name": "search_memory", "tool_input": {"query": "coffee"}}
+
+
+@dataclass(frozen=True)
+class Endpoint:
+    """One of the two chat endpoints, and the mock standing in for its agent."""
+
+    name: str
+    path: str
+    query: str
+    mock_key: str
+
+    def url(self, workspace: Workspace, peer: Peer) -> str:
+        return self.path.format(workspace=workspace.name, peer=peer.name)
+
+
+PEER_CHAT = Endpoint(
+    name="peer",
+    path="/v3/workspaces/{workspace}/peers/{peer}/chat",
+    query="What does this user drink?",
+    mock_key="agentic_chat",
+)
+WORKSPACE_CHAT = Endpoint(
+    name="workspace",
+    path="/v3/workspaces/{workspace}/chat",
+    query="What do people here drink?",
+    mock_key="workspace_chat",
+)
+BOTH_ENDPOINTS = pytest.mark.parametrize(
+    "endpoint", [PEER_CHAT, WORKSPACE_CHAT], ids=lambda e: e.name
+)
+
+
+def _record_sample_evidence(evidence: EvidenceAccumulator) -> None:
+    """Stand in for what a real run would have collated."""
+    evidence.record_tool_calls([TOOL_CALL])
+
+
+def _stub_chat(mock: Any, content: str) -> None:
+    """Have the mocked dialectic fill in the accumulator it was handed."""
+
+    async def _chat(*_args: object, **kwargs: Any) -> str:
+        evidence = kwargs.get("evidence")
+        if evidence is not None:
+            _record_sample_evidence(evidence)
+        return content
+
+    mock.side_effect = _chat
+
+
+def _stub_chat_stream(mock: Any) -> None:
+    def _chat_stream(*_args: object, **kwargs: Any) -> AsyncIterator[str]:
+        evidence = kwargs.get("evidence")
+
+        async def _chunks() -> AsyncIterator[str]:
+            if evidence is not None:
+                _record_sample_evidence(evidence)
+            for chunk in ("Test ", "streaming ", "response"):
+                yield chunk
+
+        return _chunks()
+
+    mock.side_effect = _chat_stream
+
+
+def _sse_events(text: str) -> list[dict[str, Any]]:
+    return [
+        json.loads(line.removeprefix("data: "))
+        for line in text.splitlines()
+        if line.startswith("data: ")
+    ]
+
+
+@BOTH_ENDPOINTS
+class TestBothEndpoints:
+    """The two chat endpoints have to behave identically here."""
+
+    def test_evidence_is_absent_by_default(
+        self,
+        endpoint: Endpoint,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        workspace, peer = sample_data
+
+        response = client.post(
+            endpoint.url(workspace, peer), json={"query": endpoint.query}
+        )
+
+        assert response.status_code == 200
+        assert response.json()["evidence"] is None
+
+    def test_serializes_what_the_run_collated(
+        self,
+        endpoint: Endpoint,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        mock_llm_call_functions: dict[str, Any],
+    ):
+        workspace, peer = sample_data
+        _stub_chat(mock_llm_call_functions[endpoint.mock_key], "They drink coffee.")
+
+        response = client.post(
+            endpoint.url(workspace, peer),
+            json={"query": endpoint.query, "include_evidence": True},
+        )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["content"] == "They drink coffee."
+        assert body["evidence"]["tool_calls"] == [TOOL_CALL]
+
+    def test_evidence_rides_on_the_terminal_stream_event(
+        self,
+        endpoint: Endpoint,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        mock_llm_call_functions: dict[str, Any],
+    ):
+        workspace, peer = sample_data
+        _stub_chat_stream(mock_llm_call_functions[f"{endpoint.mock_key}_stream"])
+
+        response = client.post(
+            endpoint.url(workspace, peer),
+            json={
+                "query": endpoint.query,
+                "stream": True,
+                "include_evidence": True,
+            },
+        )
+
+        assert response.status_code == 200
+        events = _sse_events(response.text)
+        content_events, final = events[:-1], events[-1]
+        assert "".join(e["delta"]["content"] for e in content_events) == (
+            "Test streaming response"
+        )
+        assert all(e["done"] is False for e in content_events)
+        assert all("evidence" not in e for e in content_events)
+        assert final["done"] is True
+        assert final["evidence"]["tool_calls"] == [TOOL_CALL]
+
+    def test_the_stream_is_unchanged_when_evidence_is_not_requested(
+        self,
+        endpoint: Endpoint,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        workspace, peer = sample_data
+
+        response = client.post(
+            endpoint.url(workspace, peer),
+            json={"query": endpoint.query, "stream": True},
+        )
+
+        assert _sse_events(response.text)[-1] == {"done": True}
+
+
+class TestOptingIn:
+    """Whether an accumulator reaches the dialectic at all."""
+
+    @pytest.mark.parametrize(
+        ("body", "expect_accumulator"),
+        [
+            ({}, False),
+            ({"include_evidence": False}, False),
+            ({"include_evidence": True}, True),
+        ],
+        ids=["omitted", "false", "true"],
+    )
+    def test_an_accumulator_is_created_only_on_request(
+        self,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        mock_llm_call_functions: dict[str, Any],
+        body: dict[str, Any],
+        expect_accumulator: bool,
+    ):
+        """Opting out has to cost nothing, so nothing is collected at all."""
+        workspace, peer = sample_data
+
+        client.post(
+            PEER_CHAT.url(workspace, peer), json={"query": PEER_CHAT.query, **body}
+        )
+
+        await_args = mock_llm_call_functions["agentic_chat"].await_args
+        assert await_args is not None
+        accumulator = await_args.kwargs["evidence"]
+        assert isinstance(accumulator, EvidenceAccumulator) is expect_accumulator
+
+    def test_reports_empty_evidence_when_nothing_was_read(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        """Asked-for-but-empty is not the same as not asked for.
+
+        The default mocked dialectic reads nothing, so this is the shape a run
+        that found nothing produces -- and it is distinguishable from `null`.
+        """
+        workspace, peer = sample_data
+
+        response = client.post(
+            PEER_CHAT.url(workspace, peer),
+            json={"query": PEER_CHAT.query, "include_evidence": True},
+        )
+
+        assert response.json()["evidence"] == {
+            "conclusions": [],
+            "messages": [],
+            "tool_calls": [],
+            "reasoning_trace_id": None,
+        }
+
+
+class TestEvidenceSchemaContract:
+    def test_evidence_is_documented_on_the_response_schema(self, client: TestClient):
+        """The chat routes hand-inject their 200 schema, so it can drift."""
+        schema = client.get("/openapi.json").json()
+        path = "/v3/workspaces/{workspace_id}/peers/{peer_id}/chat"
+        chat_schema = schema["paths"][path]["post"]["responses"]["200"]["content"][
+            "application/json"
+        ]["schema"]
+
+        assert "evidence" in chat_schema["properties"]
+
+    @pytest.mark.parametrize(
+        "options_schema", ["DialecticOptions", "WorkspaceChatOptions"]
+    )
+    def test_the_request_schema_advertises_the_toggle(
+        self, client: TestClient, options_schema: str
+    ):
+        schema = client.get("/openapi.json").json()
+        properties = schema["components"]["schemas"][options_schema]["properties"]
+
+        assert properties["include_evidence"]["default"] is False
+
+
+def test_a_bad_request_still_fails_when_evidence_is_requested(
+    client: TestClient, sample_data: tuple[Workspace, Peer]
+):
+    """Asking for evidence must not reorder the validation that runs first.
+
+    The accumulator is built in the handler, so it must not be constructed
+    ahead of the checks that reject the request outright.
+    """
+    workspace, _ = sample_data
+
+    response = client.post(
+        f"/v3/workspaces/{workspace.name}/peers/{generate_nanoid()}/chat",
+        json={
+            "query": "anything",
+            "include_evidence": True,
+            "response_format": {"type": "array"},
+        },
+    )
+
+    assert response.status_code == 422

tests/utils/test_agent_tools.py

@@ -1,9 +1,12 @@
 """Tests for agent tools in src/utils/agent_tools.py"""
 
 import asyncio
+import inspect
+import re
 from collections.abc import Callable
-from datetime import datetime, timedelta, timezone
-from typing import Any
+from dataclasses import replace
+from datetime import UTC, datetime, timedelta
+from typing import Any, ClassVar
 from unittest.mock import AsyncMock
 
 import pytest
@@ -14,9 +17,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from src import crud, models, schemas
 from src.config import settings
 from src.utils.agent_tools import (
+    _TOOL_HANDLERS,  # pyright: ignore[reportPrivateUsage]
+    _WORKSPACE_TOOL_HANDLERS,  # pyright: ignore[reportPrivateUsage]
+    DIALECTIC_TOOLS,
+    DIALECTIC_TOOLS_MINIMAL,
     MAX_PEER_CARD_ENTRY_LENGTH,
     MAX_PEER_CARD_FACTS,
     PEER_CARD_ALLOWED_PREFIXES,
+    WORKSPACE_DIALECTIC_TOOLS,
+    WORKSPACE_TOOLS_MINIMAL,
     ObservationsCreatedResult,
     ToolContext,
     _bounded_int,  # pyright: ignore[reportPrivateUsage]
@@ -27,6 +36,7 @@ from src.utils.agent_tools import (
     _handle_get_messages_by_date_range,  # pyright: ignore[reportPrivateUsage]
     _handle_get_observation_context,  # pyright: ignore[reportPrivateUsage]
     _handle_get_peer_card,  # pyright: ignore[reportPrivateUsage]
+    _handle_get_reasoning_chain,  # pyright: ignore[reportPrivateUsage]
     _handle_get_recent_history,  # pyright: ignore[reportPrivateUsage]
     _handle_get_recent_observations,  # pyright: ignore[reportPrivateUsage]
     _handle_get_session_summary,  # pyright: ignore[reportPrivateUsage]
@@ -43,6 +53,7 @@ from src.utils.agent_tools import (
     get_observation_context,
     get_recent_history,
 )
+from src.utils.evidence import EvidenceAccumulator
 
 # =============================================================================
 # Fixtures
@@ -81,7 +92,7 @@ async def tool_test_data(
     await db_session.flush()
 
     # Create messages in the session
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
     messages: list[models.Message] = []
     for i in range(5):
         peer_name = peer2.name if i % 2 == 0 else peer1.name
@@ -151,6 +162,7 @@ def make_tool_context(tool_test_data: Any) -> Callable[..., ToolContext]:
         run_id: str | None = None,
         agent_type: str | None = None,
         parent_category: str | None = None,
+        evidence: EvidenceAccumulator | None = None,
     ) -> ToolContext:
         return ToolContext(
             workspace_name=workspace.name,
@@ -164,6 +176,7 @@ def make_tool_context(tool_test_data: Any) -> Callable[..., ToolContext]:
             run_id=run_id,
             agent_type=agent_type,
             parent_category=parent_category,
+            evidence=evidence,
         )
 
     return _make_context
@@ -380,7 +393,7 @@ class TestCreateObservations:
             session_name=session.name,
             workspace_name=workspace.name,
             message_ids=[],
-            message_created_at=str(datetime.now(timezone.utc)),
+            message_created_at=str(datetime.now(UTC)),
         )
 
         assert isinstance(result, ObservationsCreatedResult)
@@ -443,7 +456,7 @@ class TestCreateObservations:
             session_name=session.name,
             workspace_name=workspace.name,
             message_ids=[],
-            message_created_at=str(datetime.now(timezone.utc)),
+            message_created_at=str(datetime.now(UTC)),
         )
 
         assert isinstance(result, ObservationsCreatedResult)
@@ -502,7 +515,7 @@ class TestCreateObservations:
             session_name=session.name,
             workspace_name=workspace.name,
             message_ids=[],
-            message_created_at=str(datetime.now(timezone.utc)),
+            message_created_at=str(datetime.now(UTC)),
         )
 
         assert isinstance(result, ObservationsCreatedResult)
@@ -557,7 +570,7 @@ class TestCreateObservations:
             session_name=session.name,
             workspace_name=workspace.name,
             message_ids=[],
-            message_created_at=str(datetime.now(timezone.utc)),
+            message_created_at=str(datetime.now(UTC)),
         )
 
         assert isinstance(result, ObservationsCreatedResult)
@@ -593,7 +606,7 @@ class TestCreateObservations:
             session_name=session.name,
             workspace_name=workspace.name,
             message_ids=[],
-            message_created_at=str(datetime.now(timezone.utc)),
+            message_created_at=str(datetime.now(UTC)),
         )
 
         assert isinstance(result, ObservationsCreatedResult)
@@ -652,7 +665,7 @@ class TestCreateObservations:
             session_name=session.name,
             workspace_name=workspace.name,
             message_ids=[],
-            message_created_at=str(datetime.now(timezone.utc)),
+            message_created_at=str(datetime.now(UTC)),
         )
 
         assert isinstance(result, ObservationsCreatedResult)
@@ -931,7 +944,7 @@ class TestSearchMemory:
                 content="Relevant fallback message",
                 seq_in_session=1,
                 token_count=5,
-                created_at=datetime.now(timezone.utc),
+                created_at=datetime.now(UTC),
             )
             return [([msg], [msg])]
 
@@ -1097,7 +1110,7 @@ class TestSearchMessagesTemporal:
                 content="Relevant temporal fallback message",
                 seq_in_session=1,
                 token_count=5,
-                created_at=datetime.now(timezone.utc),
+                created_at=datetime.now(UTC),
             )
             return [([msg], [msg])]
 
@@ -1132,7 +1145,7 @@ class TestGetMessagesByDateRange:
         ctx = make_tool_context()
 
         # Get messages from today
-        today = datetime.now(timezone.utc).date().isoformat()
+        today = datetime.now(UTC).date().isoformat()
         result = await _handle_get_messages_by_date_range(
             ctx, {"after_date": today, "limit": 10}
         )
@@ -1633,7 +1646,7 @@ class TestExtractPreferences:
             content="I prefer brief responses and always include code examples",
             seq_in_session=100,
             token_count=20,
-            created_at=datetime.now(timezone.utc),
+            created_at=datetime.now(UTC),
         )
         db_session.add(preference_msg)
         await db_session.flush()
@@ -1682,7 +1695,7 @@ class TestExtractPreferences:
                 content=f"Relevant from {query}",
                 seq_in_session=1,
                 token_count=5,
-                created_at=datetime.now(timezone.utc),
+                created_at=datetime.now(UTC),
             )
             return [([msg], [])]
 
@@ -2078,3 +2091,361 @@ class TestSessionAllowlistFailClosed:
             session_allowlist=["some-other-session"],
         )
         assert blocked == []
+
+
+# =============================================================================
+# Evidence Collection
+# =============================================================================
+
+
+def _stub_message_search(
+    monkeypatch: pytest.MonkeyPatch,
+    crud_function: str,
+    snippets: list[tuple[list[models.Message], list[models.Message]]],
+) -> None:
+    """Make a message search return known rows.
+
+    The real searches are embedding-backed and the test embedding client
+    returns a fixed vector, so they match nothing. Stubbing keeps these tests
+    about whether the handler records what it got back.
+    """
+
+    async def fake_search(**_kwargs: Any) -> Any:
+        return snippets
+
+    monkeypatch.setattr(f"src.utils.agent_tools.crud.{crud_function}", fake_search)
+
+
+@pytest.mark.asyncio
+class TestEvidenceCollection:
+    """Tests that read handlers record what they loaded.
+
+    The accumulator itself is tested in tests/utils/test_evidence.py. What
+    matters here is the wiring: a handler that returns rows to the model but
+    never records them makes evidence silently incomplete, and nothing else
+    catches that.
+    """
+
+    async def test_search_memory_records_the_conclusions_it_returned(
+        self,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        monkeypatch.setattr(settings.VECTOR_STORE, "MIGRATED", False)
+        *_, documents = tool_test_data
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_search_memory(ctx, {"query": "coffee preferences"})
+
+        # Assert on the IDs, not on evidence merely being populated: an empty
+        # accumulator would satisfy a "collected something" check.
+        assert set(evidence.conclusions) == {doc.id for doc in documents}
+
+    async def test_search_memory_records_nothing_without_an_accumulator(
+        self,
+        make_tool_context: Callable[..., ToolContext],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """The handler still answers when evidence was not requested."""
+        monkeypatch.setattr(settings.VECTOR_STORE, "MIGRATED", False)
+        ctx = make_tool_context()
+
+        result = await _handle_search_memory(ctx, {"query": "coffee preferences"})
+
+        assert ctx.evidence is None
+        assert "Found" in getattr(result, "content", result)
+
+    async def test_search_memory_records_messages_on_its_empty_memory_fallback(
+        self,
+        db_session: AsyncSession,
+        sample_data: tuple[models.Workspace, models.Peer],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """With no conclusions to find, search_memory falls back to messages."""
+        monkeypatch.setattr(settings.VECTOR_STORE, "MIGRATED", False)
+        workspace, peer1 = sample_data
+        peer2 = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name)
+        session = models.Session(
+            name=str(generate_nanoid()), workspace_name=workspace.name
+        )
+        db_session.add_all([peer2, session])
+        await db_session.flush()
+        db_session.add(
+            models.Collection(
+                workspace_name=workspace.name, observer=peer1.name, observed=peer2.name
+            )
+        )
+        message = models.Message(
+            workspace_name=workspace.name,
+            session_name=session.name,
+            peer_name=peer2.name,
+            content="I drink a lot of coffee",
+            seq_in_session=1,
+            token_count=10,
+        )
+        db_session.add(message)
+        await db_session.flush()
+        await db_session.refresh(message)
+        await db_session.commit()
+
+        _stub_message_search(monkeypatch, "search_messages", [([message], [message])])
+
+        evidence = EvidenceAccumulator()
+        ctx = ToolContext(
+            workspace_name=workspace.name,
+            observer=peer1.name,
+            observed=peer2.name,
+            session_name=session.name,
+            current_messages=None,
+            include_observation_ids=False,
+            history_token_limit=8192,
+            db_lock=asyncio.Lock(),
+            agent_type="dialectic",
+            evidence=evidence,
+        )
+
+        await _handle_search_memory(ctx, {"query": "coffee"})
+
+        assert evidence.conclusions == {}
+        assert message.public_id in evidence.messages
+
+    async def test_search_messages_records_matches_and_their_context(
+        self,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """Context messages count as read: they reach the prompt too."""
+        _, _, _, _, messages, _ = tool_test_data
+        _stub_message_search(
+            monkeypatch,
+            "search_messages",
+            [([messages[2]], [messages[1], messages[3]])],
+        )
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_search_messages(ctx, {"query": "Test message"})
+
+        assert set(evidence.messages) == {
+            messages[1].public_id,
+            messages[2].public_id,
+            messages[3].public_id,
+        }
+
+    async def test_grep_messages_records_the_messages_it_matched(
+        self,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+    ):
+        _, _, _, _, messages, _ = tool_test_data
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_grep_messages(ctx, {"text": "Test message 1"})
+
+        assert messages[1].public_id in evidence.messages
+
+    async def test_search_messages_temporal_records_the_messages_it_matched(
+        self,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        _, _, _, _, messages, _ = tool_test_data
+        _stub_message_search(
+            monkeypatch, "search_messages_temporal", [([messages[0]], [messages[0]])]
+        )
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_search_messages_temporal(ctx, {"query": "Test message"})
+
+        assert set(evidence.messages) == {messages[0].public_id}
+
+    async def test_get_messages_by_date_range_records_the_messages_it_returned(
+        self,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+    ):
+        _, _, _, _, messages, _ = tool_test_data
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_get_messages_by_date_range(ctx, {"limit": 20})
+
+        assert set(evidence.messages) == {message.public_id for message in messages}
+
+    async def test_get_observation_context_records_the_messages_it_returned(
+        self,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+    ):
+        _, _, _, _, messages, _ = tool_test_data
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_get_observation_context(
+            ctx, {"message_ids": [messages[2].public_id]}
+        )
+
+        # The tool returns the named message plus its neighbours.
+        assert messages[2].public_id in evidence.messages
+
+    async def test_get_reasoning_chain_records_the_chain_it_walked(
+        self,
+        db_session: AsyncSession,
+        tool_test_data: Any,
+        make_tool_context: Callable[..., ToolContext],
+    ):
+        workspace, peer1, peer2, session, _, documents = tool_test_data
+        derived = models.Document(
+            workspace_name=workspace.name,
+            observer=peer1.name,
+            observed=peer2.name,
+            content="User is a morning coffee drinker",
+            embedding=[0.5] * 1536,
+            session_name=session.name,
+            level="deductive",
+            source_ids=[documents[0].id, documents[2].id],
+        )
+        db_session.add(derived)
+        await db_session.flush()
+        await db_session.refresh(derived)
+        await db_session.commit()
+
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        await _handle_get_reasoning_chain(ctx, {"observation_id": derived.id})
+
+        assert {derived.id, documents[0].id, documents[2].id} <= set(
+            evidence.conclusions
+        )
+
+    async def test_evidence_stays_empty_when_recall_fails_closed(
+        self,
+        tool_test_data: Any,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """An empty session allowlist recalls nothing, so it cites nothing.
+
+        Evidence must not become a way around the allowlist: it only ever
+        reports rows a permitted read actually returned.
+        """
+        monkeypatch.setattr(settings.VECTOR_STORE, "MIGRATED", False)
+        workspace, peer1, peer2, session, _, _ = tool_test_data
+        evidence = EvidenceAccumulator()
+        ctx = ToolContext(
+            workspace_name=workspace.name,
+            observer=peer1.name,
+            observed=peer2.name,
+            session_name=session.name,
+            current_messages=None,
+            include_observation_ids=False,
+            history_token_limit=8192,
+            db_lock=asyncio.Lock(),
+            session_allowlist=[],
+            agent_type="dialectic",
+            evidence=evidence,
+        )
+
+        await _handle_search_memory(ctx, {"query": "coffee preferences"})
+        await _handle_search_messages(ctx, {"query": "Test message"})
+        await _handle_grep_messages(ctx, {"text": "Test message"})
+
+        assert evidence.conclusions == {}
+        assert evidence.messages == {}
+
+    async def test_a_replaced_context_shares_the_same_accumulator(
+        self,
+        make_tool_context: Callable[..., ToolContext],
+    ):
+        """Workspace handlers delegate through `dataclasses.replace`.
+
+        That is a shallow copy, so the copy must append to the original
+        accumulator rather than to one of its own.
+        """
+        evidence = EvidenceAccumulator()
+        ctx = make_tool_context(evidence=evidence)
+
+        delegated = replace(ctx, observer="someone-else", observed="someone-else")
+
+        assert delegated.evidence is evidence
+
+
+# The recording helpers a read handler is expected to call. Matched against
+# handler source so a newly added tool fails the coverage guard below.
+RECORDING_CALL = re.compile(r"_record_(conclusion|message|snippet)_evidence\(")
+
+
+class TestEvidenceCoverage:
+    """Guards against a dialectic tool being added without evidence wiring.
+
+    A new read tool that never records is invisible: the answer looks right and
+    evidence just quietly under-reports. This asserts the wiring exists for
+    every tool the dialectic can reach.
+    """
+
+    # Tools that read rows and must record them.
+    RECORDING_TOOLS: ClassVar[set[str]] = {
+        "search_memory",
+        "search_messages",
+        "grep_messages",
+        "search_messages_temporal",
+        "get_messages_by_date_range",
+        "get_observation_context",
+        "get_reasoning_chain",
+    }
+    # Tools with nothing citable to record: workspace stats are aggregates and a
+    # peer card is free text, so neither carries conclusion or message identity.
+    NON_RECORDING_TOOLS: ClassVar[set[str]] = {
+        "get_workspace_stats",
+        "get_peer_card",
+    }
+
+    def test_every_dialectic_tool_is_accounted_for(self):
+        reachable = {
+            name
+            for tools in (
+                DIALECTIC_TOOLS,
+                DIALECTIC_TOOLS_MINIMAL,
+                WORKSPACE_DIALECTIC_TOOLS,
+                WORKSPACE_TOOLS_MINIMAL,
+            )
+            for tool in tools
+            if isinstance((name := tool.get("name")), str)
+        }
+
+        unclassified = reachable - self.RECORDING_TOOLS - self.NON_RECORDING_TOOLS
+        assert not unclassified, (
+            "New dialectic tool(s) with no evidence decision: "
+            f"{sorted(unclassified)}. Record what they read in the handler and "
+            "add them to RECORDING_TOOLS, or justify them in NON_RECORDING_TOOLS."
+        )
+
+    @pytest.mark.parametrize("tool_name", sorted(RECORDING_TOOLS))
+    def test_a_recording_tool_reaches_the_accumulator(self, tool_name: str):
+        """Every recording handler must consult `ctx.evidence`.
+
+        A source check rather than a behavioural one so a newly added tool
+        fails here at once, without needing fixture data shaped to make it
+        return rows. One tool name can resolve to two handlers -- the workspace
+        agent overrides some -- and it is enough for one of them to record,
+        since the workspace overrides delegate to the pair handler with a
+        copied context that shares the accumulator.
+        """
+        handlers = [
+            table[tool_name]
+            for table in (_TOOL_HANDLERS, _WORKSPACE_TOOL_HANDLERS)
+            if tool_name in table
+        ]
+        assert handlers, f"{tool_name} has no handler"
+
+        sources = [inspect.getsource(handler) for handler in handlers]
+        assert any(RECORDING_CALL.search(source) for source in sources), (
+            f"{tool_name} returns rows to the model but never records them, so "
+            "they will be missing from evidence"
+        )

tests/utils/test_evidence.py

@@ -0,0 +1,369 @@
+"""Tests for the evidence accumulator in src/utils/evidence.py.
+
+These exercise collation in isolation: rows are built in memory and never
+written, so nothing here depends on what the agent or its tools do. The wiring
+that feeds the accumulator is covered in tests/utils/test_agent_tools.py.
+"""
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import pytest
+
+from src import models
+from src.utils.evidence import EvidenceAccumulator
+
+NOW = datetime(2026, 1, 1, tzinfo=UTC)
+LEVELS = ("explicit", "deductive", "inductive", "contradiction")
+
+
+def make_document(
+    doc_id: str,
+    *,
+    level: str = "explicit",
+    content: str = "User likes coffee",
+    internal_metadata: dict[str, Any] | None = None,
+    source_ids: list[str] | None = None,
+    session_name: str | None = "session-1",
+    created_at: datetime = NOW,
+) -> models.Document:
+    """Build an unpersisted Document.
+
+    Note `internal_metadata=` and not `metadata=`: `metadata` is SQLAlchemy's
+    own class attribute and assigning it shadows that instead of setting the
+    column.
+    """
+    return models.Document(
+        id=doc_id,
+        level=level,
+        content=content,
+        internal_metadata=internal_metadata or {},
+        source_ids=source_ids,
+        session_name=session_name,
+        created_at=created_at,
+        observer="observer",
+        observed="observed",
+        workspace_name="workspace",
+    )
+
+
+def make_message(
+    public_id: str,
+    *,
+    content: str = "I love coffee",
+    peer_name: str = "alice",
+    session_name: str = "session-1",
+    created_at: datetime = NOW,
+) -> models.Message:
+    return models.Message(
+        public_id=public_id,
+        content=content,
+        peer_name=peer_name,
+        session_name=session_name,
+        created_at=created_at,
+        workspace_name="workspace",
+        seq_in_session=1,
+    )
+
+
+class TestConclusionCollection:
+    def test_records_id_level_and_content(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents([make_document("doc-1", content="User likes tea")])
+
+        (conclusion,) = accumulator.build().conclusions
+        assert conclusion.id == "doc-1"
+        assert conclusion.level == "explicit"
+        assert conclusion.content == "User likes tea"
+        assert conclusion.session_id == "session-1"
+
+    def test_deduplicates_by_id_across_tools(self):
+        """A conclusion two tools both returned is reported once."""
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents([make_document("doc-1")])
+        accumulator.add_documents([make_document("doc-1"), make_document("doc-2")])
+
+        assert [c.id for c in accumulator.build().conclusions] == ["doc-1", "doc-2"]
+
+    def test_keeps_distinct_conclusions_that_read_alike(self):
+        """Identical text is not identity.
+
+        `Representation`'s own deduplication keys on content and timestamp and
+        ignores IDs, so building evidence through it without keying on ID first
+        would drop one of these and pick arbitrarily between their IDs.
+        """
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [
+                make_document("doc-1", content="User likes coffee"),
+                make_document("doc-2", content="User likes coffee"),
+            ]
+        )
+
+        assert [c.id for c in accumulator.build().conclusions] == ["doc-1", "doc-2"]
+
+    @pytest.mark.parametrize(
+        ("level", "document_kwargs"),
+        [
+            ("inductive", {"source_ids": ["doc-1", "doc-2"]}),
+            ("deductive", {"internal_metadata": {"premise_ids": ["doc-1", "doc-2"]}}),
+            ("inductive", {"internal_metadata": {"source_ids": ["doc-1", "doc-2"]}}),
+        ],
+        ids=["column", "metadata-premise-ids", "metadata-source-ids"],
+    )
+    def test_resolves_source_ids_from_either_location(
+        self, level: str, document_kwargs: dict[str, Any]
+    ):
+        """Rows predating the `source_ids` column keep their premises in metadata."""
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [make_document("doc-3", level=level, **document_kwargs)]
+        )
+
+        (conclusion,) = accumulator.build().conclusions
+        assert conclusion.level == level
+        assert conclusion.source_ids == ["doc-1", "doc-2"]
+
+    def test_explicit_conclusions_have_no_source_ids(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents([make_document("doc-1")])
+
+        assert accumulator.build().conclusions[0].source_ids == []
+
+    def test_carries_a_null_session_for_unscoped_conclusions(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents([make_document("doc-1", session_name=None)])
+
+        assert accumulator.build().conclusions[0].session_id is None
+
+    def test_reports_every_level(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [
+                make_document("doc-1", level="explicit"),
+                make_document("doc-2", level="deductive"),
+                make_document("doc-3", level="inductive"),
+                make_document("doc-4", level="contradiction"),
+            ]
+        )
+
+        assert {c.level for c in accumulator.build().conclusions} == set(LEVELS)
+
+    @pytest.mark.parametrize("level", LEVELS)
+    def test_reports_the_text_of_every_level(self, level: str):
+        """Derived levels name their text `conclusion`, the others `content`."""
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [make_document("doc-1", level=level, content="User drinks coffee")]
+        )
+
+        (conclusion,) = accumulator.build().conclusions
+        assert conclusion.content == "User drinks coffee"
+
+    def test_orders_conclusions_by_derivation_time(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [
+                make_document("doc-late", created_at=NOW + timedelta(hours=1)),
+                make_document("doc-early", created_at=NOW),
+            ]
+        )
+
+        assert [c.id for c in accumulator.build().conclusions] == [
+            "doc-early",
+            "doc-late",
+        ]
+
+    def test_dates_a_conclusion_from_its_source_messages(self):
+        """The logical timestamp beats the row's insert time when it is recorded."""
+        derived_from = NOW - timedelta(days=30)
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [
+                make_document(
+                    "doc-1",
+                    created_at=NOW,
+                    internal_metadata={"message_created_at": derived_from.isoformat()},
+                )
+            ]
+        )
+
+        # Seconds resolution: representations drop microseconds.
+        assert accumulator.build().conclusions[0].created_at == derived_from
+
+
+class TestMessageCollection:
+    def test_records_identity_and_provenance(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_messages([make_message("msg-1", peer_name="bob")])
+
+        (message,) = accumulator.build().messages
+        assert message.id == "msg-1"
+        assert message.peer_id == "bob"
+        assert message.session_id == "session-1"
+
+    def test_deduplicates_by_id(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_messages([make_message("msg-1"), make_message("msg-1")])
+        accumulator.add_messages([make_message("msg-1")])
+
+        assert [m.id for m in accumulator.build().messages] == ["msg-1"]
+
+    @pytest.mark.parametrize(
+        "content", ["short", "x" * 5000], ids=["short", "pasted-document"]
+    )
+    def test_carries_no_message_content_at_any_length(self, content: str):
+        """Evidence names messages; it does not reproduce them.
+
+        Message content is caller-supplied and unbounded, so a single answer
+        could otherwise drag megabytes behind it. Callers fetch by ID.
+        """
+        accumulator = EvidenceAccumulator()
+        accumulator.add_messages([make_message("msg-1", content=content)])
+
+        (message,) = accumulator.build().messages
+        assert not hasattr(message, "content_preview")
+        assert content not in accumulator.build().model_dump_json()
+
+    def test_orders_messages_chronologically(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.add_messages(
+            [
+                make_message("msg-late", created_at=NOW + timedelta(minutes=1)),
+                make_message("msg-early", created_at=NOW),
+            ]
+        )
+
+        assert [m.id for m in accumulator.build().messages] == [
+            "msg-early",
+            "msg-late",
+        ]
+
+
+class TestToolCallCollection:
+    def test_reads_the_accumulated_loop_history(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.record_tool_calls(
+            [
+                {
+                    "tool_name": "search_memory",
+                    "tool_input": {"query": "coffee"},
+                    "tool_result": "Found 3 observations",
+                    "tool_result_metadata": {"results_count": 3},
+                }
+            ]
+        )
+
+        (call,) = accumulator.build().tool_calls
+        assert call.tool_name == "search_memory"
+        assert call.tool_input == {"query": "coffee"}
+
+    def test_reads_the_raw_provider_shape(self):
+        """A provider response names the same fields `name`/`input`."""
+        accumulator = EvidenceAccumulator()
+        accumulator.record_tool_calls(
+            [{"id": "call-1", "name": "grep_messages", "input": {"text": "coffee"}}]
+        )
+
+        (call,) = accumulator.build().tool_calls
+        assert call.tool_name == "grep_messages"
+        assert call.tool_input == {"text": "coffee"}
+
+    def test_omits_tool_results(self):
+        """Results are large and already reflected in conclusions and messages."""
+        accumulator = EvidenceAccumulator()
+        accumulator.record_tool_calls(
+            [
+                {
+                    "tool_name": "search_memory",
+                    "tool_input": {"query": "coffee"},
+                    "tool_result": "SENTINEL-RESULT-TEXT",
+                }
+            ]
+        )
+
+        assert "SENTINEL-RESULT-TEXT" not in accumulator.build().model_dump_json()
+
+    def test_preserves_call_order(self):
+        accumulator = EvidenceAccumulator()
+        accumulator.record_tool_calls(
+            [
+                {"tool_name": "search_memory", "tool_input": {}},
+                {"tool_name": "search_messages", "tool_input": {}},
+                {"tool_name": "search_memory", "tool_input": {}},
+            ]
+        )
+
+        assert [c.tool_name for c in accumulator.build().tool_calls] == [
+            "search_memory",
+            "search_messages",
+            "search_memory",
+        ]
+
+    def test_overwrites_rather_than_appends(self):
+        """The tool loop rewrites its log wholesale on every exit path."""
+        accumulator = EvidenceAccumulator()
+        accumulator.record_tool_calls(
+            [{"tool_name": "search_memory", "tool_input": {}}]
+        )
+        accumulator.record_tool_calls(
+            [
+                {"tool_name": "search_memory", "tool_input": {}},
+                {"tool_name": "grep_messages", "tool_input": {}},
+            ]
+        )
+
+        assert len(accumulator.build().tool_calls) == 2
+
+    @pytest.mark.parametrize(
+        "entry",
+        [
+            {"tool_name": "search_memory"},
+            {"tool_name": "search_memory", "tool_input": None},
+            {"tool_name": "search_memory", "tool_input": "not-a-dict"},
+        ],
+    )
+    def test_tolerates_a_missing_or_malformed_input(self, entry: dict[str, Any]):
+        accumulator = EvidenceAccumulator()
+        accumulator.record_tool_calls([entry])
+
+        assert accumulator.build().tool_calls[0].tool_input == {}
+
+
+class TestTimestamps:
+    def test_every_timestamp_names_its_timezone(self):
+        """A caller should never have to guess what zone a timestamp is in.
+
+        Conclusion timestamps arrive via `Representation`, which strips tzinfo
+        so observations render compactly into prompts; message timestamps come
+        straight off the column and keep theirs. Evidence has to be consistent.
+        """
+        accumulator = EvidenceAccumulator()
+        accumulator.add_documents(
+            [make_document(f"doc-{level}", level=level) for level in LEVELS]
+        )
+        accumulator.add_messages([make_message("msg-1")])
+        evidence = accumulator.build()
+
+        assert len(evidence.conclusions) == len(LEVELS)
+        for conclusion in evidence.conclusions:
+            assert conclusion.created_at.tzinfo is not None, conclusion.id
+        for message in evidence.messages:
+            assert message.created_at.tzinfo is not None, message.id
+
+
+class TestEmptyEvidence:
+    def test_builds_an_empty_object_rather_than_none(self):
+        """An agent that read nothing still reports evidence, just empty.
+
+        `evidence` being absent means the caller did not ask for it; an empty
+        `evidence` means it was asked for and nothing was read. Callers can
+        tell those apart, so a test asserting only `evidence is not None`
+        proves nothing.
+        """
+        evidence = EvidenceAccumulator().build()
+
+        assert evidence.conclusions == []
+        assert evidence.messages == []
+        assert evidence.tool_calls == []
+        assert evidence.reasoning_trace_id is None