Plastic Labs Internal Wiki

Recently Updated

reasoning-traces

32 min read

Reasoning Traces Storage

Status: Draft | Owner: vineeth | Last updated: 2026-03-25

1. Problem Statement

Honcho’s three background agents (Deriver, Dialectic, Dreamer) and the Summarizer all make LLM calls that produce reasoning traces — the full input/output of each call including thinking content, tool calls, token counts, and model parameters. Today these traces are logged to a JSONL file via log_reasoning_trace() in src/telemetry/reasoning_traces.py and then effectively discarded:

  1. Not queryable. The JSONL file is append-only with file-level locking. There is no way to search, filter, or retrieve specific traces.

  2. No database storage. Traces live on local disk, are lost on container restart (Fly.io machines are ephemeral), and cannot be accessed by other processes.

  3. No provenance chain. When a Deriver call produces observations (documents), there is no link from those observations back to the LLM call that created them. Developers cannot answer “why did Honcho conclude X?” without manually correlating timestamps across logs.

  4. Not accessible to agents. The Dialectic and Dreamer agents cannot inspect past reasoning. When the Dreamer consolidates observations, it cannot see what reasoning led to an existing deductive observation. When the Dialectic answers questions, it cannot reference the reasoning chain behind a conclusion.

  5. No retention management. Traces accumulate indefinitely on disk with no cleanup mechanism.

This spec designs a PostgreSQL-backed reasoning trace storage system with full provenance linking, developer-facing API, agent tool integration, size-based storage tiering, configurable retention, and CloudEvents billing support.

2. Goals / Non-Goals

Goals

  • G1: Store reasoning traces in PostgreSQL with a dedicated reasoning_traces table, replacing the JSONL file approach.
  • G2: Establish a bidirectional provenance chain: Message → Trace → Observation. Given any observation, a developer can retrieve the trace that produced it. Given any message, they can see all traces and observations it triggered.
  • G3: Expose traces via a read-only REST API with filtering by agent type, task type, time range, workspace, and peer.
  • G4: Add a get_reasoning_trace tool to the Dialectic and Dreamer agents so they can inspect past reasoning when making decisions.
  • G5: Implement size-based storage tiering — inline JSONB for traces under 50KB, S3 reference for larger traces.
  • G6: Provide configurable retention with indefinite as the default, plus options for time-based and count-based pruning.
  • G7: Emit CloudEvents for trace storage operations to support billing in the managed SaaS platform.
  • G8: Maintain backward compatibility — the JSONL file logging continues to work as a fallback when database storage is disabled.

Non-Goals

  • Exposing traces as a writable API (traces are system-generated, read-only to developers).
  • Real-time streaming of traces as they are produced (traces are written after LLM calls complete).
  • Full-text search over trace content (JSONB containment queries and provenance-chain lookups are sufficient).
  • Storing intermediate tool call results separately — these are captured within the trace’s content JSONB.
  • Modifying the LLM client interface (honcho_llm_call) beyond adding context parameters for trace linking.

3. Design

3.1 Schema: reasoning_traces Table

3.1.1 DDL

CREATE TABLE {schema}.reasoning_traces (
    -- Primary key: nanoid, consistent with all other Honcho tables
    id          TEXT PRIMARY KEY,
 
    -- Workspace scoping (foreign key to workspaces.name)
    workspace_name TEXT NOT NULL REFERENCES {schema}.workspaces(name),
 
    -- Session context (nullable -- dialectic global queries have no session)
    session_name   TEXT,
 
    -- Peer context: which peer relationship produced this trace
    observer    TEXT NOT NULL,
    observed    TEXT NOT NULL,
 
    -- Agent identification
    agent_type  TEXT NOT NULL,   -- 'deriver', 'dialectic', 'dreamer_deduction', 'dreamer_induction', 'summarizer'
    task_type   TEXT NOT NULL,   -- 'minimal_deriver', 'dialectic_chat', 'dreamer_deduction', 'dreamer_induction', 'short_summary', 'long_summary'
 
    -- Model metadata
    provider    TEXT NOT NULL,   -- 'anthropic', 'google', 'openai', 'groq', 'custom', 'vllm'
    model       TEXT NOT NULL,   -- e.g. 'claude-haiku-4-5', 'gemini-2.5-flash-lite'
 
    -- Token accounting
    input_tokens                  INTEGER NOT NULL DEFAULT 0,
    output_tokens                 INTEGER NOT NULL DEFAULT 0,
    cache_creation_input_tokens   INTEGER NOT NULL DEFAULT 0,
    cache_read_input_tokens       INTEGER NOT NULL DEFAULT 0,
 
    -- Execution metadata
    duration_ms     DOUBLE PRECISION,   -- wall-clock time for the full LLM call (including tool loops)
    iteration_count INTEGER NOT NULL DEFAULT 1,  -- number of LLM call iterations (1 = single call, 2+ = tool loop)
 
    -- Trace content: inline JSONB for small traces, S3 reference for large ones
    -- Structure when inline:
    --   {
    --     "settings": { "max_tokens": 4096, "thinking_budget_tokens": 1024, ... },
    --     "input": { "tokens": 1234, "prompt": "..." | "messages": [...] },
    --     "output": { "content": "...", "thinking_content": "...", "tool_calls": [...], "finish_reasons": [...] }
    --   }
    -- Structure when externalized:
    --   { "_ref": "s3", "bucket": "honcho-traces", "key": "ws/session/trace_id.json.gz", "size_bytes": 123456 }
    content     JSONB NOT NULL,
 
    -- Content size tracking (always set, regardless of storage location)
    content_size_bytes INTEGER NOT NULL DEFAULT 0,
 
    -- Provenance linkage: observations produced by this trace
    -- Array of document IDs (nanoid format). Updated after observations are created.
    observation_ids JSONB DEFAULT '[]'::jsonb,
 
    -- Provenance linkage: source message IDs that triggered this trace
    -- For deriver: the message IDs from the queue batch
    -- For dialectic: empty (triggered by API call, not by specific messages)
    -- For dreamer: empty (triggered by schedule/threshold)
    -- For summarizer: the message IDs covered by the summary
    message_ids JSONB DEFAULT '[]'::jsonb,
 
    -- Timestamps
    created_at  TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(),
 
    -- Composite foreign keys for peer validation
    CONSTRAINT fk_reasoning_traces_observer
        FOREIGN KEY (observer, workspace_name)
        REFERENCES {schema}.peers(name, workspace_name),
    CONSTRAINT fk_reasoning_traces_observed
        FOREIGN KEY (observed, workspace_name)
        REFERENCES {schema}.peers(name, workspace_name),
    -- Session foreign key (nullable)
    CONSTRAINT fk_reasoning_traces_session
        FOREIGN KEY (session_name, workspace_name)
        REFERENCES {schema}.sessions(name, workspace_name),
 
    -- Constraints
    CONSTRAINT ck_reasoning_traces_id_length CHECK (length(id) = 21),
    CONSTRAINT ck_reasoning_traces_id_format CHECK (id ~ '^[A-Za-z0-9_-]+$'),
    CONSTRAINT ck_reasoning_traces_agent_type CHECK (agent_type IN (
        'deriver', 'dialectic', 'dreamer_deduction', 'dreamer_induction', 'summarizer'
    )),
    CONSTRAINT ck_reasoning_traces_content_size CHECK (content_size_bytes >= 0)
);
 
-- Indexes
CREATE INDEX ix_reasoning_traces_workspace_name ON {schema}.reasoning_traces(workspace_name);
CREATE INDEX ix_reasoning_traces_created_at ON {schema}.reasoning_traces(created_at);
CREATE INDEX ix_reasoning_traces_agent_type ON {schema}.reasoning_traces(agent_type);
CREATE INDEX ix_reasoning_traces_observer_observed ON {schema}.reasoning_traces(observer, observed);
CREATE INDEX ix_reasoning_traces_session_name ON {schema}.reasoning_traces(session_name)
    WHERE session_name IS NOT NULL;
 
-- GIN index for observation_ids lookups ("find trace by observation ID")
CREATE INDEX ix_reasoning_traces_observation_ids_gin ON {schema}.reasoning_traces
    USING gin (observation_ids);
 
-- GIN index for message_ids lookups ("find traces triggered by message")
CREATE INDEX ix_reasoning_traces_message_ids_gin ON {schema}.reasoning_traces
    USING gin (message_ids);
 
-- Composite index for common query pattern: workspace + agent_type + time ordering
CREATE INDEX ix_reasoning_traces_workspace_agent_created
    ON {schema}.reasoning_traces(workspace_name, agent_type, created_at DESC);

3.1.2 SQLAlchemy Model

Add to src/models.py:

@final
class ReasoningTrace(Base):
    __tablename__: str = "reasoning_traces"
 
    id: Mapped[str] = mapped_column(TEXT, default=generate_nanoid, primary_key=True)
    workspace_name: Mapped[str] = mapped_column(
        ForeignKey("workspaces.name"), nullable=False, index=True
    )
    session_name: Mapped[str | None] = mapped_column(TEXT, nullable=True, index=True)
    observer: Mapped[str] = mapped_column(TEXT, nullable=False)
    observed: Mapped[str] = mapped_column(TEXT, nullable=False)
 
    agent_type: Mapped[str] = mapped_column(TEXT, nullable=False, index=True)
    task_type: Mapped[str] = mapped_column(TEXT, nullable=False)
 
    provider: Mapped[str] = mapped_column(TEXT, nullable=False)
    model: Mapped[str] = mapped_column(TEXT, nullable=False)
 
    input_tokens: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    output_tokens: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    cache_creation_input_tokens: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0")
    )
    cache_read_input_tokens: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0")
    )
 
    duration_ms: Mapped[float | None] = mapped_column(
        sa.Float, nullable=True
    )
    iteration_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=1, server_default=text("1")
    )
 
    content: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
    content_size_bytes: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0")
    )
 
    observation_ids: Mapped[list[str]] = mapped_column(
        JSONB, default=list, server_default=text("'[]'::jsonb")
    )
    message_ids: Mapped[list[str]] = mapped_column(
        JSONB, default=list, server_default=text("'[]'::jsonb")
    )
 
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), index=True
    )
 
    __table_args__ = (
        CheckConstraint("length(id) = 21", name="id_length"),
        CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'", name="id_format"),
        CheckConstraint(
            "agent_type IN ('deriver', 'dialectic', 'dreamer_deduction', 'dreamer_induction', 'summarizer')",
            name="agent_type_valid",
        ),
        CheckConstraint("content_size_bytes >= 0", name="content_size_non_negative"),
        ForeignKeyConstraint(
            ["observer", "workspace_name"],
            ["peers.name", "peers.workspace_name"],
        ),
        ForeignKeyConstraint(
            ["observed", "workspace_name"],
            ["peers.name", "peers.workspace_name"],
        ),
        ForeignKeyConstraint(
            ["session_name", "workspace_name"],
            ["sessions.name", "sessions.workspace_name"],
        ),
        Index(
            "ix_reasoning_traces_observer_observed",
            "observer", "observed",
        ),
        Index(
            "ix_reasoning_traces_observation_ids_gin",
            "observation_ids",
            postgresql_using="gin",
        ),
        Index(
            "ix_reasoning_traces_message_ids_gin",
            "message_ids",
            postgresql_using="gin",
        ),
        Index(
            "ix_reasoning_traces_workspace_agent_created",
            "workspace_name", "agent_type", created_at.desc(),
        ),
    )
 
    def __repr__(self) -> str:
        return (
            f"ReasoningTrace(id={self.id}, agent_type={self.agent_type}, "
            f"task_type={self.task_type}, workspace_name={self.workspace_name})"
        )

3.1.3 content JSONB Structure

When stored inline (content_size_bytes < 50KB):

{
    "settings": {
        "max_tokens": 4096,
        "thinking_budget_tokens": 1024,
        "reasoning_effort": null,
        "json_mode": true,
        "stop_seqs": ["   \n", "\n\n\n\n"],
        "temperature": null
    },
    "input": {
        "tokens": 5432,
        "prompt": "You are a minimal deriver..."
    },
    "output": {
        "content": "{\"explicit\": [...], \"deductive\": [...]}",
        "tokens": 892,
        "thinking_content": "Let me analyze these messages...",
        "finish_reasons": ["end_turn"],
        "tool_calls": []
    }
}

For multi-turn/agentic calls (dialectic, dreamer), input.messages replaces input.prompt:

{
    "settings": { ... },
    "input": {
        "tokens": 12345,
        "messages": [
            {"role": "system", "content": "You are a dialectic agent..."},
            {"role": "user", "content": "Query: What does Alice prefer?"},
            {"role": "assistant", "content": "...", "tool_calls": [...]},
            {"role": "tool", "content": "..."},
            ...
        ]
    },
    "output": {
        "content": "Based on the observations...",
        "tokens": 1543,
        "thinking_content": "I should search for...",
        "finish_reasons": ["end_turn"],
        "tool_calls": [
            {"name": "search_memory", "input": {"query": "Alice preferences"}, "iteration": 1},
            {"name": "get_observation_context", "input": {"message_ids": ["abc"]}, "iteration": 2}
        ]
    }
}

When externalized to S3 (content_size_bytes >= 50KB):

{
    "_ref": "s3",
    "bucket": "honcho-traces",
    "key": "traces/my_workspace/2026/03/25/abc123XYZ.json.gz",
    "size_bytes": 156789,
    "compressed_bytes": 34567
}

The externalized file contains the same structure as the inline version, gzip-compressed.

3.2 Provenance Chain

The provenance chain connects three entities:

Message(s)  --->  ReasoningTrace  --->  Observation(s) (Documents)
   ^                   |                      |
   |                   v                      v
message_ids[]     observation_ids[]     internal_metadata.trace_id

3.2.1 Forward Links (Trace → Observations)

When the Deriver, Dialectic, or Dreamer creates observations, the trace’s observation_ids array is updated with the IDs of the created documents.

Deriver flow:

  1. honcho_llm_call() completes, producing HonchoLLMCallResponse
  2. save_reasoning_trace() writes the trace to the database, returns trace_id
  3. RepresentationManager.save_representation() creates documents
  4. After document creation, link_trace_to_observations(trace_id, document_ids) updates the trace

Dialectic flow:

  1. DialecticAgent.answer() calls honcho_llm_call() with tools
  2. During tool execution, create_observations tool may create documents
  3. Tool executor accumulates created document IDs
  4. After the agent loop completes, save_reasoning_trace() is called with accumulated observation IDs

Dreamer flow:

  1. Each specialist calls honcho_llm_call() with tools
  2. Tool executor accumulates created/deleted document IDs
  3. After the specialist completes, save_reasoning_trace() is called with accumulated observation IDs

3.2.2 Backward Links (Observation → Trace)

Each document’s internal_metadata gains a trace_id field pointing back to the reasoning trace that created it.

Updated DocumentMetadata in src/schemas/internal.py:

class DocumentMetadata(BaseModel):
    message_ids: list[int] = Field(...)
    message_created_at: str = Field(...)
    source_ids: list[str] | None = Field(default=None, ...)
    premises: list[str] | None = Field(default=None, ...)
    sources: list[str] | None = Field(default=None, ...)
    pattern_type: str | None = Field(default=None, ...)
    confidence: str | None = Field(default=None, ...)
    # NEW: Link back to the reasoning trace that produced this observation
    trace_id: str | None = Field(
        default=None,
        description="ID of the reasoning trace that produced this observation",
    )

For observations created by the tool-based agents (dialectic, dreamer), the trace_id is set at document creation time by the tool executor. For deriver observations, it is set in RepresentationManager.save_representation() after the trace is persisted.

3.2.3 Provenance Queries

“Given this observation, show me the trace that produced it”:

-- Option A: via internal_metadata.trace_id on the document
SELECT rt.*
FROM reasoning_traces rt
JOIN documents d ON d.internal_metadata->>'trace_id' = rt.id
WHERE d.id = :document_id;
 
-- Option B: via observation_ids GIN index on the trace
SELECT rt.*
FROM reasoning_traces rt
WHERE rt.observation_ids @> :document_id_jsonb;
-- where :document_id_jsonb = '["abc123XYZ"]'::jsonb

“Given this message, show me all traces and observations”:

-- Find all traces triggered by a message (deriver traces)
SELECT rt.*, rt.observation_ids
FROM reasoning_traces rt
WHERE rt.message_ids @> :message_public_id_jsonb;
-- where :message_public_id_jsonb = '["msg_abc123"]'::jsonb

“Show the full provenance chain for an observation”:

-- 1. Get the observation
SELECT d.id, d.content, d.internal_metadata, d.source_ids
FROM documents d WHERE d.id = :observation_id;
 
-- 2. Get the trace that produced it
SELECT rt.*
FROM reasoning_traces rt
WHERE rt.observation_ids @> to_jsonb(ARRAY[:observation_id]);
 
-- 3. Get the source messages from the trace
-- (message_ids in the trace are internal IDs; join to get public_ids and content)
SELECT m.public_id, m.content, m.peer_name, m.created_at
FROM messages m
WHERE m.public_id = ANY(
    SELECT jsonb_array_elements_text(rt.message_ids)
    FROM reasoning_traces rt
    WHERE rt.observation_ids @> to_jsonb(ARRAY[:observation_id])
);

3.3 API Surface

All trace endpoints are read-only. Traces are created by the system, not by developers.

3.3.1 Router: src/routers/traces.py

router = APIRouter(
    prefix="/workspaces/{workspace_id}/traces",
    tags=["traces"],
    dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)

3.3.2 Endpoints

List traces with filters:

POST /v3/workspaces/{workspace_id}/traces/list

Request body:

{
    "filters": {
        "agent_type": "deriver",
        "observer": "honcho",
        "observed": "alice",
        "session_id": "my_session",
        "created_after": "2026-03-01T00:00:00Z",
        "created_before": "2026-03-25T23:59:59Z"
    }
}

Response (paginated via fastapi_pagination):

{
    "items": [
        {
            "id": "abc123XYZ...",
            "workspace_id": "my_workspace",
            "session_id": "my_session",
            "observer_id": "honcho",
            "observed_id": "alice",
            "agent_type": "deriver",
            "task_type": "minimal_deriver",
            "provider": "google",
            "model": "gemini-2.5-flash-lite",
            "input_tokens": 5432,
            "output_tokens": 892,
            "cache_creation_input_tokens": 0,
            "cache_read_input_tokens": 3200,
            "duration_ms": 1243.5,
            "iteration_count": 1,
            "content_size_bytes": 12345,
            "observation_ids": ["obs_001", "obs_002"],
            "message_ids": ["msg_abc", "msg_def"],
            "created_at": "2026-03-25T10:30:00Z"
        }
    ],
    "total": 150,
    "page": 1,
    "size": 50,
    "pages": 3
}

Note: The list endpoint does NOT include content to keep response sizes manageable. Use the detail endpoint to retrieve content.

Get full trace (with content):

GET /v3/workspaces/{workspace_id}/traces/{trace_id}

Response:

{
    "id": "abc123XYZ...",
    "workspace_id": "my_workspace",
    "session_id": "my_session",
    "observer_id": "honcho",
    "observed_id": "alice",
    "agent_type": "deriver",
    "task_type": "minimal_deriver",
    "provider": "google",
    "model": "gemini-2.5-flash-lite",
    "input_tokens": 5432,
    "output_tokens": 892,
    "cache_creation_input_tokens": 0,
    "cache_read_input_tokens": 3200,
    "duration_ms": 1243.5,
    "iteration_count": 1,
    "content_size_bytes": 12345,
    "content": {
        "settings": { "max_tokens": 4096, "thinking_budget_tokens": 1024 },
        "input": { "tokens": 5432, "prompt": "..." },
        "output": { "content": "...", "thinking_content": "...", "finish_reasons": ["end_turn"] }
    },
    "observation_ids": ["obs_001", "obs_002"],
    "message_ids": ["msg_abc", "msg_def"],
    "created_at": "2026-03-25T10:30:00Z"
}

If the trace content is externalized to S3, the server fetches it transparently and returns the full content. The content._ref field is never exposed to the API consumer.

Get trace for a specific conclusion:

GET /v3/workspaces/{workspace_id}/conclusions/{conclusion_id}/trace

Response: Same shape as the single trace response, or 404 if no trace is linked.

Implementation: Query documents.internal_metadata->>'trace_id' then fetch the trace by ID. Falls back to the GIN index query on reasoning_traces.observation_ids.

Get traces for a message:

GET /v3/workspaces/{workspace_id}/messages/{message_id}/traces

Response: Array of trace summaries (without content), same shape as list items.

Implementation: Query reasoning_traces.message_ids @> '["<message_public_id>"]'::jsonb.

3.3.3 Pydantic Schemas

Add to src/schemas/api.py:

# ---------------------------------------------------------------------------
# Trace schemas
# ---------------------------------------------------------------------------
 
class TraceListFilters(BaseModel):
    """Filters for listing reasoning traces."""
    agent_type: str | None = Field(default=None, description="Filter by agent type")
    task_type: str | None = Field(default=None, description="Filter by task type")
    observer: str | None = Field(default=None, alias="observer_id")
    observed: str | None = Field(default=None, alias="observed_id")
    session_name: str | None = Field(default=None, alias="session_id")
    created_after: datetime.datetime | None = None
    created_before: datetime.datetime | None = None
 
    model_config = ConfigDict(populate_by_name=True)
 
 
class TraceListRequest(BaseModel):
    """Request body for listing traces."""
    filters: TraceListFilters | None = None
 
 
class TraceSummary(BaseModel):
    """Trace summary returned in list responses (no content)."""
    id: str
    workspace_name: str = Field(serialization_alias="workspace_id")
    session_name: str | None = Field(default=None, serialization_alias="session_id")
    observer: str = Field(serialization_alias="observer_id")
    observed: str = Field(serialization_alias="observed_id")
    agent_type: str
    task_type: str
    provider: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_creation_input_tokens: int
    cache_read_input_tokens: int
    duration_ms: float | None
    iteration_count: int
    content_size_bytes: int
    observation_ids: list[str]
    message_ids: list[str]
    created_at: datetime.datetime
 
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
 
 
class TraceDetail(TraceSummary):
    """Full trace with content, returned by detail endpoints."""
    content: dict[str, Any]

3.4 Agent Tool Integration

3.4.1 New Tool: get_reasoning_trace

Add to TOOLS dict in src/utils/agent_tools.py:

TOOLS["get_reasoning_trace"] = {
    "name": "get_reasoning_trace",
    "description": (
        "Retrieve the reasoning trace that produced a specific observation. "
        "Use this to understand WHY an observation exists -- what evidence "
        "the agent considered and what reasoning it applied. Pass the observation ID "
        "from search results (the [id:xxx] field)."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "observation_id": {
                "type": "string",
                "description": "The ID of the observation to get the reasoning trace for",
            },
        },
        "required": ["observation_id"],
    },
}

3.4.2 Tool Handler

Add to src/utils/agent_tools.py:

async def _handle_get_reasoning_trace(
    ctx: ToolContext, tool_input: dict[str, Any]
) -> str:
    """Handle get_reasoning_trace tool."""
    observation_id = tool_input.get("observation_id", "")
    if not observation_id:
        return "Error: observation_id is required"
 
    try:
        trace = await crud.get_trace_for_observation(
            ctx.db,
            workspace_name=ctx.workspace_name,
            observation_id=observation_id,
        )
    except Exception as e:
        logger.warning(f"Failed to get reasoning trace for {observation_id}: {e}")
        return f"No reasoning trace found for observation {observation_id}"
 
    if trace is None:
        return f"No reasoning trace found for observation {observation_id}"
 
    # Format trace for the agent -- include thinking content and key metadata
    parts = [
        f"## Reasoning Trace for Observation [{observation_id}]",
        f"Agent: {trace.agent_type} ({trace.task_type})",
        f"Model: {trace.provider}/{trace.model}",
        f"Created: {trace.created_at.isoformat()}",
        f"Iterations: {trace.iteration_count}",
    ]
 
    # Extract thinking content from trace
    content = trace.content
    if isinstance(content, dict) and "_ref" not in content:
        output = content.get("output", {})
        thinking = output.get("thinking_content")
        if thinking:
            parts.append(f"\n### Reasoning:\n{thinking}")
 
        # Include tool calls if present
        tool_calls = output.get("tool_calls", [])
        if tool_calls:
            tool_names = [tc.get("name", "unknown") for tc in tool_calls]
            parts.append(f"\n### Tools Used: {', '.join(tool_names)}")
 
        # Include the output content (the actual conclusion/response)
        output_content = output.get("content", "")
        if output_content:
            # Truncate if very long
            truncated = _truncate_tool_output(str(output_content), max_chars=3000)
            parts.append(f"\n### Output:\n{truncated}")
    else:
        parts.append("\n(Trace content stored externally -- summary not available)")
 
    # Include sibling observations from same trace
    sibling_ids = [
        oid for oid in (trace.observation_ids or []) if oid != observation_id
    ]
    if sibling_ids:
        parts.append(f"\n### Other observations from this trace: {', '.join(sibling_ids)}")
 
    return _truncate_tool_output("\n".join(parts))

Register in _TOOL_HANDLERS:

_TOOL_HANDLERS["get_reasoning_trace"] = _handle_get_reasoning_trace

3.4.3 Tool Set Updates

Add get_reasoning_trace to the tool sets that should have access:

# Dialectic tools -- add to DIALECTIC_TOOLS (not DIALECTIC_TOOLS_MINIMAL)
DIALECTIC_TOOLS = [
    TOOLS["get_recent_history"],
    TOOLS["get_recent_observations"],
    TOOLS["get_most_derived_observations"],
    TOOLS["search_memory"],
    TOOLS["search_messages"],
    TOOLS["get_observation_context"],
    TOOLS["grep_messages"],
    TOOLS["get_messages_by_date_range"],
    TOOLS["search_messages_temporal"],
    TOOLS["get_session_summary"],
    TOOLS["get_peer_card"],
    TOOLS["get_reasoning_trace"],  # NEW
]
 
# Dreamer deduction tools
DEDUCTION_SPECIALIST_TOOLS = [
    TOOLS["get_recent_observations"],
    TOOLS["get_most_derived_observations"],
    TOOLS["search_memory"],
    TOOLS["get_peer_card"],
    TOOLS["create_observations"],
    TOOLS["delete_observations"],
    TOOLS["update_peer_card"],
    TOOLS["search_messages"],
    TOOLS["get_observation_context"],
    TOOLS["get_reasoning_trace"],  # NEW
]
 
# Dreamer induction tools
INDUCTION_SPECIALIST_TOOLS = [
    TOOLS["get_recent_observations"],
    TOOLS["get_most_derived_observations"],
    TOOLS["search_memory"],
    TOOLS["search_messages"],
    TOOLS["create_observations"],
    TOOLS["update_peer_card"],
    TOOLS["get_reasoning_trace"],  # NEW
]

The DIALECTIC_TOOLS_MINIMAL set does NOT include this tool (minimal reasoning should stay cheap).

3.5 Storage Tiering

3.5.1 Size Threshold

The threshold for externalizing trace content is 50KB (51,200 bytes), measured as the byte length of the JSON-serialized content before compression. This is configurable via settings.

Rationale: A typical deriver trace is 5-15KB. Dialectic traces with multi-turn tool loops can reach 30-80KB. Dreamer traces with many tool iterations can exceed 100KB. The 50KB threshold keeps the majority of traces inline while preventing PostgreSQL row bloat for the largest traces.

3.5.2 Configuration

Add to src/config.py:

class TraceStorageSettings(HonchoSettings):
    """Settings for reasoning trace storage."""
    model_config = SettingsConfigDict(env_prefix="TRACE_", extra="ignore")
 
    # Master toggle for database storage of traces
    ENABLED: bool = False
 
    # Size threshold (bytes) for externalizing content to S3
    # Traces smaller than this are stored inline as JSONB
    EXTERNALIZE_THRESHOLD_BYTES: Annotated[
        int, Field(default=51200, gt=0, le=10_000_000)
    ] = 51200  # 50KB
 
    # S3 configuration for externalized traces
    S3_BUCKET: str | None = None
    S3_PREFIX: str = "traces"
    S3_REGION: str | None = None
    S3_ENDPOINT_URL: str | None = None  # For S3-compatible stores (MinIO, R2, etc.)
    S3_ACCESS_KEY_ID: str | None = None
    S3_SECRET_ACCESS_KEY: str | None = None
 
    # Retention policy
    RETENTION_POLICY: Literal["indefinite", "time_based", "count_based"] = "indefinite"
    RETENTION_DAYS: Annotated[int, Field(default=90, gt=0, le=3650)] = 90  # only used if time_based
    RETENTION_MAX_PER_WORKSPACE: Annotated[
        int, Field(default=10000, gt=0, le=1_000_000)
    ] = 10000  # only used if count_based
 
    # Cleanup interval
    CLEANUP_INTERVAL_HOURS: Annotated[int, Field(default=24, gt=0, le=168)] = 24
 
    @model_validator(mode="after")
    def _require_s3_for_externalization(self) -> "TraceStorageSettings":
        """Warn if externalization is implicitly disabled due to missing S3 config."""
        # S3 is optional -- traces that exceed the threshold will still be stored
        # inline with a warning if S3 is not configured
        return self

Add to AppSettings:

class AppSettings(HonchoSettings):
    ...
    TRACE: TraceStorageSettings = Field(default_factory=TraceStorageSettings)

Add "TRACE": "trace" to TomlConfigSettingsSource.SECTION_MAP.

3.5.3 TOML Configuration Example

[trace]
enabled = true
externalize_threshold_bytes = 51200
s3_bucket = "honcho-traces"
s3_region = "us-east-1"
s3_prefix = "traces"
retention_policy = "indefinite"

3.5.4 S3 Client

Add src/telemetry/trace_storage.py:

"""
Reasoning trace storage backend.
 
Handles writing traces to PostgreSQL with optional S3 externalization
for large trace content.
"""
 
import gzip
import json
import logging
from typing import Any
 
import boto3
from botocore.config import Config as BotoConfig
from nanoid import generate as generate_nanoid
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
 
from src.config import settings
from src.models import ReasoningTrace
 
logger = logging.getLogger(__name__)
 
# Lazy-initialized S3 client
_s3_client = None
 
 
def _get_s3_client():
    """Get or create S3 client."""
    global _s3_client
    if _s3_client is not None:
        return _s3_client
 
    trace_settings = settings.TRACE
    if not trace_settings.S3_BUCKET:
        return None
 
    kwargs: dict[str, Any] = {}
    if trace_settings.S3_REGION:
        kwargs["region_name"] = trace_settings.S3_REGION
    if trace_settings.S3_ENDPOINT_URL:
        kwargs["endpoint_url"] = trace_settings.S3_ENDPOINT_URL
    if trace_settings.S3_ACCESS_KEY_ID:
        kwargs["aws_access_key_id"] = trace_settings.S3_ACCESS_KEY_ID
    if trace_settings.S3_SECRET_ACCESS_KEY:
        kwargs["aws_secret_access_key"] = trace_settings.S3_SECRET_ACCESS_KEY
 
    _s3_client = boto3.client(
        "s3",
        config=BotoConfig(
            retries={"max_attempts": 3, "mode": "adaptive"},
            connect_timeout=5,
            read_timeout=10,
        ),
        **kwargs,
    )
    return _s3_client
 
 
def _build_s3_key(workspace_name: str, trace_id: str) -> str:
    """Build S3 object key for a trace."""
    from datetime import datetime, UTC
    now = datetime.now(UTC)
    prefix = settings.TRACE.S3_PREFIX.rstrip("/")
    return f"{prefix}/{workspace_name}/{now.year}/{now.month:02d}/{now.day:02d}/{trace_id}.json.gz"
 
 
def _externalize_content(
    content: dict[str, Any],
    workspace_name: str,
    trace_id: str,
) -> tuple[dict[str, Any], int]:
    """
    Upload trace content to S3 and return a reference object.
 
    Returns:
        Tuple of (reference_dict, original_size_bytes)
    """
    s3 = _get_s3_client()
    if s3 is None:
        # S3 not configured -- store inline with warning
        logger.warning(
            "Trace %s exceeds externalize threshold (%d bytes) but S3 is not configured. "
            "Storing inline.",
            trace_id,
            len(json.dumps(content).encode()),
        )
        return content, len(json.dumps(content).encode())
 
    # Serialize and compress
    content_bytes = json.dumps(content).encode("utf-8")
    original_size = len(content_bytes)
    compressed = gzip.compress(content_bytes, compresslevel=6)
 
    key = _build_s3_key(workspace_name, trace_id)
    bucket = settings.TRACE.S3_BUCKET
 
    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=compressed,
        ContentType="application/json",
        ContentEncoding="gzip",
    )
 
    ref = {
        "_ref": "s3",
        "bucket": bucket,
        "key": key,
        "size_bytes": original_size,
        "compressed_bytes": len(compressed),
    }
    return ref, original_size
 
 
def fetch_externalized_content(content: dict[str, Any]) -> dict[str, Any]:
    """
    Fetch externalized trace content from S3.
 
    If content is not a reference (no _ref key), returns it as-is.
    """
    if not isinstance(content, dict) or "_ref" not in content:
        return content
 
    if content["_ref"] != "s3":
        raise ValueError(f"Unknown content reference type: {content['_ref']}")
 
    s3 = _get_s3_client()
    if s3 is None:
        raise RuntimeError("S3 client not configured but trace references S3 storage")
 
    response = s3.get_object(Bucket=content["bucket"], Key=content["key"])
    compressed = response["Body"].read()
    decompressed = gzip.decompress(compressed)
    return json.loads(decompressed)
 
 
async def save_reasoning_trace(
    db: AsyncSession,
    *,
    workspace_name: str,
    session_name: str | None,
    observer: str,
    observed: str,
    agent_type: str,
    task_type: str,
    provider: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
    cache_creation_input_tokens: int = 0,
    cache_read_input_tokens: int = 0,
    duration_ms: float | None = None,
    iteration_count: int = 1,
    content: dict[str, Any],
    observation_ids: list[str] | None = None,
    message_ids: list[str] | None = None,
) -> str | None:
    """
    Save a reasoning trace to the database.
 
    Handles size-based tiering: if the serialized content exceeds the
    configured threshold, it is externalized to S3 and a reference
    is stored in the JSONB column instead.
 
    Returns:
        The trace ID if storage is enabled, None otherwise.
    """
    if not settings.TRACE.ENABLED:
        return None
 
    trace_id = generate_nanoid()
 
    # Calculate content size
    content_bytes = json.dumps(content).encode("utf-8")
    content_size = len(content_bytes)
 
    # Externalize if over threshold
    stored_content = content
    if content_size >= settings.TRACE.EXTERNALIZE_THRESHOLD_BYTES:
        stored_content, content_size = _externalize_content(
            content, workspace_name, trace_id
        )
 
    trace = ReasoningTrace(
        id=trace_id,
        workspace_name=workspace_name,
        session_name=session_name,
        observer=observer,
        observed=observed,
        agent_type=agent_type,
        task_type=task_type,
        provider=provider,
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cache_creation_input_tokens=cache_creation_input_tokens,
        cache_read_input_tokens=cache_read_input_tokens,
        duration_ms=duration_ms,
        iteration_count=iteration_count,
        content=stored_content,
        content_size_bytes=content_size,
        observation_ids=observation_ids or [],
        message_ids=message_ids or [],
    )
 
    db.add(trace)
    await db.commit()
 
    return trace_id
 
 
async def link_trace_to_observations(
    db: AsyncSession,
    trace_id: str,
    observation_ids: list[str],
) -> None:
    """
    Update a trace's observation_ids after observations are created.
 
    This is called after the deriver creates observations, since the trace
    is persisted before the observations exist.
    """
    if not settings.TRACE.ENABLED or not trace_id or not observation_ids:
        return
 
    stmt = (
        update(ReasoningTrace)
        .where(ReasoningTrace.id == trace_id)
        .values(observation_ids=observation_ids)
    )
    await db.execute(stmt)
    await db.commit()

3.6 Retention Policy

3.6.1 Policy Types

PolicyBehaviorConfiguration
indefiniteTraces are never automatically deletedDefault
time_basedTraces older than N days are deletedTRACE_RETENTION_DAYS=90
count_basedKeep at most N traces per workspace; delete oldest firstTRACE_RETENTION_MAX_PER_WORKSPACE=10000

3.6.2 Cleanup Job

The cleanup job runs as a reconciler task, reusing the existing queue infrastructure. It is enqueued periodically by the deriver process (which already runs the reconciler).

Add to src/telemetry/trace_storage.py:

async def cleanup_expired_traces(db: AsyncSession) -> int:
    """
    Delete traces according to the configured retention policy.
 
    Returns the number of traces deleted.
    """
    trace_settings = settings.TRACE
    if not trace_settings.ENABLED or trace_settings.RETENTION_POLICY == "indefinite":
        return 0
 
    deleted = 0
 
    if trace_settings.RETENTION_POLICY == "time_based":
        from datetime import datetime, timedelta, UTC
        cutoff = datetime.now(UTC) - timedelta(days=trace_settings.RETENTION_DAYS)
 
        # Find traces to delete (need to clean up S3 objects first)
        stmt = (
            select(ReasoningTrace)
            .where(ReasoningTrace.created_at < cutoff)
            .limit(1000)  # Process in batches to avoid long transactions
        )
        result = await db.execute(stmt)
        traces = result.scalars().all()
 
        for trace in traces:
            _delete_externalized_content(trace.content)
            await db.delete(trace)
            deleted += 1
 
        await db.commit()
 
    elif trace_settings.RETENTION_POLICY == "count_based":
        # For each workspace, keep only the most recent N traces
        from sqlalchemy import func, text
 
        # Get workspaces that exceed the limit
        count_stmt = (
            select(
                ReasoningTrace.workspace_name,
                func.count().label("trace_count"),
            )
            .group_by(ReasoningTrace.workspace_name)
            .having(func.count() > trace_settings.RETENTION_MAX_PER_WORKSPACE)
        )
        result = await db.execute(count_stmt)
        over_limit = result.all()
 
        for workspace_name, count in over_limit:
            excess = count - trace_settings.RETENTION_MAX_PER_WORKSPACE
            # Delete oldest traces for this workspace
            delete_stmt = text(f"""
                DELETE FROM {settings.DB.SCHEMA}.reasoning_traces
                WHERE id IN (
                    SELECT id FROM {settings.DB.SCHEMA}.reasoning_traces
                    WHERE workspace_name = :workspace_name
                    ORDER BY created_at ASC
                    LIMIT :excess
                )
            """)
            await db.execute(
                delete_stmt,
                {"workspace_name": workspace_name, "excess": excess},
            )
            deleted += excess
 
        await db.commit()
 
    if deleted > 0:
        logger.info("Cleaned up %d expired reasoning traces", deleted)
 
    return deleted
 
 
def _delete_externalized_content(content: dict[str, Any]) -> None:
    """Delete externalized content from S3 if applicable."""
    if not isinstance(content, dict) or content.get("_ref") != "s3":
        return
 
    s3 = _get_s3_client()
    if s3 is None:
        return
 
    try:
        s3.delete_object(Bucket=content["bucket"], Key=content["key"])
    except Exception as e:
        logger.warning("Failed to delete S3 object %s: %s", content.get("key"), e)

3.6.3 Reconciler Integration

Add trace_cleanup to the ReconcilerType enum in src/schemas/internal.py:

class ReconcilerType(str, Enum):
    SYNC_VECTORS = "sync_vectors"
    CLEANUP_QUEUE = "cleanup_queue"
    CLEANUP_TRACES = "cleanup_traces"  # NEW

The deriver’s reconciler loop enqueues a cleanup_traces task every TRACE_CLEANUP_INTERVAL_HOURS hours. The task handler calls cleanup_expired_traces().

3.7 CloudEvents

3.7.1 New Event Type

Add src/telemetry/events/trace.py:

"""CloudEvents for reasoning trace operations."""
 
from pydantic import Field
 
from src.telemetry.events.base import BaseEvent
 
 
class TraceStoredEvent(BaseEvent):
    """Emitted when a reasoning trace is stored to the database."""
 
    _event_type = "honcho.work.trace.stored"
    _schema_version = 1
    _category = "work"
 
    workspace_name: str
    trace_id: str
    agent_type: str
    task_type: str
    provider: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_creation_input_tokens: int = 0
    cache_read_input_tokens: int = 0
    content_size_bytes: int
    externalized: bool = Field(
        default=False,
        description="Whether content was externalized to S3",
    )
    observation_count: int = Field(
        default=0,
        description="Number of observations produced by this trace",
    )
 
    def get_resource_id(self) -> str:
        return f"{self.workspace_name}:{self.trace_id}"

Register in src/telemetry/events/__init__.py:

from src.telemetry.events.trace import TraceStoredEvent
 
__all__ = [
    ...
    "TraceStoredEvent",
]

3.7.2 Event Emission

In save_reasoning_trace(), emit the event after successful database commit:

from src.telemetry.events import TraceStoredEvent, emit
 
# After db.commit():
emit(
    TraceStoredEvent(
        workspace_name=workspace_name,
        trace_id=trace_id,
        agent_type=agent_type,
        task_type=task_type,
        provider=provider,
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cache_creation_input_tokens=cache_creation_input_tokens,
        cache_read_input_tokens=cache_read_input_tokens,
        content_size_bytes=content_size,
        externalized=stored_content is not content,
        observation_count=len(observation_ids or []),
    )
)

This event flows through Xatu for billing: each trace stored counts as a billable operation, and the token counts allow per-token billing if desired.

4. Migration Plan

4.1 Alembic Migration

Create migrations/versions/XXXX_add_reasoning_traces_table.py:

"""add reasoning_traces table
 
Revision ID: <auto-generated>
Revises: <current-head>
Create Date: 2026-XX-XX
"""
 
from collections.abc import Sequence
 
import sqlalchemy as sa
from alembic import op
 
from migrations.utils import table_exists, index_exists
from src.config import settings
 
revision: str = "<auto-generated>"
down_revision: str | None = "<current-head>"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
 
schema = settings.DB.SCHEMA
 
 
def upgrade() -> None:
    op.create_table(
        "reasoning_traces",
        sa.Column("id", sa.TEXT(), primary_key=True, nullable=False),
        sa.Column(
            "workspace_name",
            sa.TEXT(),
            sa.ForeignKey(f"{schema}.workspaces.name"),
            nullable=False,
        ),
        sa.Column("session_name", sa.TEXT(), nullable=True),
        sa.Column("observer", sa.TEXT(), nullable=False),
        sa.Column("observed", sa.TEXT(), nullable=False),
        sa.Column("agent_type", sa.TEXT(), nullable=False),
        sa.Column("task_type", sa.TEXT(), nullable=False),
        sa.Column("provider", sa.TEXT(), nullable=False),
        sa.Column("model", sa.TEXT(), nullable=False),
        sa.Column(
            "input_tokens", sa.Integer(), nullable=False, server_default=sa.text("0")
        ),
        sa.Column(
            "output_tokens", sa.Integer(), nullable=False, server_default=sa.text("0")
        ),
        sa.Column(
            "cache_creation_input_tokens",
            sa.Integer(),
            nullable=False,
            server_default=sa.text("0"),
        ),
        sa.Column(
            "cache_read_input_tokens",
            sa.Integer(),
            nullable=False,
            server_default=sa.text("0"),
        ),
        sa.Column("duration_ms", sa.Float(), nullable=True),
        sa.Column(
            "iteration_count",
            sa.Integer(),
            nullable=False,
            server_default=sa.text("1"),
        ),
        sa.Column(
            "content",
            sa.dialects.postgresql.JSONB(),
            nullable=False,
        ),
        sa.Column(
            "content_size_bytes",
            sa.Integer(),
            nullable=False,
            server_default=sa.text("0"),
        ),
        sa.Column(
            "observation_ids",
            sa.dialects.postgresql.JSONB(),
            nullable=False,
            server_default=sa.text("'[]'::jsonb"),
        ),
        sa.Column(
            "message_ids",
            sa.dialects.postgresql.JSONB(),
            nullable=False,
            server_default=sa.text("'[]'::jsonb"),
        ),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        # Check constraints
        sa.CheckConstraint("length(id) = 21", name="ck_reasoning_traces_id_length"),
        sa.CheckConstraint(
            "id ~ '^[A-Za-z0-9_-]+$'", name="ck_reasoning_traces_id_format"
        ),
        sa.CheckConstraint(
            "agent_type IN ('deriver', 'dialectic', 'dreamer_deduction', 'dreamer_induction', 'summarizer')",
            name="ck_reasoning_traces_agent_type_valid",
        ),
        sa.CheckConstraint(
            "content_size_bytes >= 0",
            name="ck_reasoning_traces_content_size_non_negative",
        ),
        # Composite foreign keys
        sa.ForeignKeyConstraint(
            ["observer", "workspace_name"],
            [f"{schema}.peers.name", f"{schema}.peers.workspace_name"],
            name="fk_reasoning_traces_observer",
        ),
        sa.ForeignKeyConstraint(
            ["observed", "workspace_name"],
            [f"{schema}.peers.name", f"{schema}.peers.workspace_name"],
            name="fk_reasoning_traces_observed",
        ),
        sa.ForeignKeyConstraint(
            ["session_name", "workspace_name"],
            [f"{schema}.sessions.name", f"{schema}.sessions.workspace_name"],
            name="fk_reasoning_traces_session",
        ),
        schema=schema,
    )
 
    # Indexes
    op.create_index(
        "ix_reasoning_traces_workspace_name",
        "reasoning_traces",
        ["workspace_name"],
        schema=schema,
    )
    op.create_index(
        "ix_reasoning_traces_created_at",
        "reasoning_traces",
        ["created_at"],
        schema=schema,
    )
    op.create_index(
        "ix_reasoning_traces_agent_type",
        "reasoning_traces",
        ["agent_type"],
        schema=schema,
    )
    op.create_index(
        "ix_reasoning_traces_observer_observed",
        "reasoning_traces",
        ["observer", "observed"],
        schema=schema,
    )
    op.create_index(
        "ix_reasoning_traces_session_name",
        "reasoning_traces",
        ["session_name"],
        schema=schema,
        postgresql_where=sa.text("session_name IS NOT NULL"),
    )
    op.create_index(
        "ix_reasoning_traces_observation_ids_gin",
        "reasoning_traces",
        ["observation_ids"],
        schema=schema,
        postgresql_using="gin",
    )
    op.create_index(
        "ix_reasoning_traces_message_ids_gin",
        "reasoning_traces",
        ["message_ids"],
        schema=schema,
        postgresql_using="gin",
    )
    op.create_index(
        "ix_reasoning_traces_workspace_agent_created",
        "reasoning_traces",
        ["workspace_name", "agent_type", sa.text("created_at DESC")],
        schema=schema,
    )
 
 
def downgrade() -> None:
    inspector = sa.inspect(op.get_bind())
 
    if table_exists("reasoning_traces", inspector):
        # Drop all indexes first
        for idx_name in [
            "ix_reasoning_traces_workspace_name",
            "ix_reasoning_traces_created_at",
            "ix_reasoning_traces_agent_type",
            "ix_reasoning_traces_observer_observed",
            "ix_reasoning_traces_session_name",
            "ix_reasoning_traces_observation_ids_gin",
            "ix_reasoning_traces_message_ids_gin",
            "ix_reasoning_traces_workspace_agent_created",
        ]:
            if index_exists("reasoning_traces", idx_name, inspector):
                op.drop_index(idx_name, table_name="reasoning_traces", schema=schema)
 
        op.drop_table("reasoning_traces", schema=schema)

4.2 Zero-Downtime Deployment

The migration is purely additive (new table, no existing table changes). It can be run with the existing application version still serving traffic. The feature is gated behind TRACE_ENABLED=false by default, so the migration can be deployed before the application code that writes traces.

The only change to existing tables is adding trace_id to DocumentMetadata (the internal_metadata JSONB on documents). This is a Pydantic schema change, not a database column change — the field is simply written into the existing JSONB column when present. Old documents without trace_id continue to work because the field defaults to None.

5. Implementation Phases

Phase 1: Database Storage (Foundation)

Scope: Replace JSONL logging with PostgreSQL storage. No API, no agent tools, no S3.

Files to modify:

FileChange
src/models.pyAdd ReasoningTrace model
src/config.pyAdd TraceStorageSettings, add to AppSettings and SECTION_MAP
src/telemetry/trace_storage.pyNew file: save_reasoning_trace(), link_trace_to_observations() (S3 stubs that store inline)
src/telemetry/reasoning_traces.pyModify log_reasoning_trace() to call save_reasoning_trace() in addition to JSONL
src/schemas/internal.pyAdd trace_id field to DocumentMetadata
migrations/versions/XXXX_add_reasoning_traces_table.pyNew file: Migration

Deliverables:

  • Traces written to database when TRACE_ENABLED=true
  • JSONL logging still works as before (dual-write)
  • trace_id not yet populated on documents (no provenance linkage yet)

Phase 2: Provenance Linking

Scope: Wire up the Message → Trace → Observation provenance chain in all agent paths.

Files to modify:

FileChange
src/utils/clients.pyAdd trace_context parameter to honcho_llm_call(): workspace_name, observer, observed, session_name, agent_type. After logging trace, return trace_id on the response object.
src/deriver/deriver.pyAfter honcho_llm_call(), capture trace_id. Pass to RepresentationManager.save_representation(). After observations are created, call link_trace_to_observations().
src/crud/representation.pyAccept trace_id parameter in save_representation() and _create_documents(). Set trace_id in DocumentMetadata.
src/dialectic/core.pyAfter honcho_llm_call() completes, save trace with observation IDs accumulated by tool executor.
src/dreamer/specialists.pyAfter specialist honcho_llm_call() completes, save trace with observation IDs accumulated by tool executor.
src/utils/summarizer.pyAfter summarizer honcho_llm_call(), save trace (summarizer traces have no observation_ids).
src/utils/agent_tools.pyAdd accumulated_observation_ids: list[str] to ToolContext. Append created document IDs in _handle_create_observations().

Key integration pattern for honcho_llm_call():

The trace context is passed alongside existing parameters. A new TraceContext dataclass bundles the fields needed to write a trace:

@dataclass
class TraceContext:
    """Context for saving a reasoning trace after an LLM call."""
    workspace_name: str
    session_name: str | None
    observer: str
    observed: str
    agent_type: str
 
# In honcho_llm_call, after the call completes:
# trace_id is set on the HonchoLLMCallResponse as a new optional field

Add to HonchoLLMCallResponse:

class HonchoLLMCallResponse(BaseModel, Generic[T]):
    ...
    trace_id: str | None = None  # Set when trace storage is enabled

Deliverables:

  • All LLM calls produce traces with correct context
  • Deriver traces linked to their observations (bidirectional)
  • Dialectic/Dreamer traces linked to observations created by tools
  • Summarizer traces stored (no observation linkage needed)
  • Documents have trace_id in internal_metadata

Phase 3: API & Agent Tools

Scope: Developer-facing API and agent tool integration.

Files to create/modify:

FileChange
src/routers/traces.pyNew file: API endpoints
src/schemas/api.pyAdd TraceListFilters, TraceListRequest, TraceSummary, TraceDetail
src/schemas/__init__.pyRe-export new schemas
src/crud/trace.pyNew file: CRUD operations for traces
src/crud/__init__.pyExport new functions
src/main.pyRegister traces router
src/utils/agent_tools.pyAdd get_reasoning_trace tool, handler, and register in tool sets

CRUD operations in src/crud/trace.py:

async def get_trace(
    db: AsyncSession,
    workspace_name: str,
    trace_id: str,
) -> ReasoningTrace | None:
    """Get a single trace by ID."""
    ...
 
async def list_traces(
    db: AsyncSession,
    workspace_name: str,
    *,
    agent_type: str | None = None,
    task_type: str | None = None,
    observer: str | None = None,
    observed: str | None = None,
    session_name: str | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
) -> Select:
    """Build a query for listing traces with filters."""
    ...
 
async def get_trace_for_observation(
    db: AsyncSession,
    workspace_name: str,
    observation_id: str,
) -> ReasoningTrace | None:
    """Get the trace that produced a specific observation."""
    # First try internal_metadata.trace_id on the document
    # Fall back to GIN index query on reasoning_traces.observation_ids
    ...
 
async def get_traces_for_message(
    db: AsyncSession,
    workspace_name: str,
    message_public_id: str,
) -> list[ReasoningTrace]:
    """Get all traces triggered by a specific message."""
    ...

Deliverables:

  • All API endpoints functional with pagination
  • Agent tools available to dialectic and dreamer
  • Full provenance chain queryable via API

Phase 4: S3 Tiering & Retention

Scope: S3 externalization for large traces, retention policy enforcement.

Files to modify:

FileChange
src/telemetry/trace_storage.pyImplement _externalize_content(), fetch_externalized_content(), cleanup_expired_traces(), _delete_externalized_content()
src/routers/traces.pyTransparently fetch externalized content in detail endpoints
src/reconciler/Add cleanup_traces task handler
src/schemas/internal.pyAdd CLEANUP_TRACES to ReconcilerType
src/telemetry/events/trace.pyNew file: TraceStoredEvent CloudEvent
src/telemetry/events/__init__.pyRegister new event
pyproject.tomlAdd boto3 dependency (for S3)

Deliverables:

  • Large traces externalized to S3
  • Retention policies enforced by background job
  • CloudEvents emitted for billing

6. Files to Modify (Complete Index)

New Files

PathPurpose
src/telemetry/trace_storage.pyTrace storage backend (save, link, fetch, cleanup, S3)
src/routers/traces.pyAPI router for trace endpoints
src/crud/trace.pyCRUD operations for traces
src/telemetry/events/trace.pyCloudEvents event type for trace operations
migrations/versions/XXXX_add_reasoning_traces_table.pyDatabase migration

Modified Files

PathChange Summary
src/models.pyAdd ReasoningTrace model class
src/config.pyAdd TraceStorageSettings class, add to AppSettings, add to SECTION_MAP
src/schemas/api.pyAdd TraceListFilters, TraceListRequest, TraceSummary, TraceDetail
src/schemas/internal.pyAdd trace_id field to DocumentMetadata; add CLEANUP_TRACES to ReconcilerType
src/schemas/__init__.pyRe-export new schemas
src/main.pyRegister traces.router with /v3 prefix
src/telemetry/reasoning_traces.pyAdd DB write path alongside JSONL; accept and forward trace context
src/telemetry/events/__init__.pyImport and export TraceStoredEvent
src/utils/clients.pyAdd `trace_context: TraceContext
src/utils/agent_tools.pyAdd get_reasoning_trace tool definition, handler, and dispatch; add accumulated_observation_ids to ToolContext; update _handle_create_observations to accumulate IDs; add tool to DIALECTIC_TOOLS, DEDUCTION_SPECIALIST_TOOLS, INDUCTION_SPECIALIST_TOOLS
src/deriver/deriver.pyPass trace context to honcho_llm_call(); capture trace_id; pass to save_representation(); call link_trace_to_observations()
src/crud/representation.pyAccept trace_id in save_representation() and _create_documents(); set trace_id in DocumentMetadata
src/dialectic/core.pySave trace after honcho_llm_call() with accumulated observation IDs from tool executor
src/dreamer/specialists.pySave trace after specialist honcho_llm_call() with accumulated observation IDs
src/utils/summarizer.pySave trace after summarizer LLM call
src/crud/__init__.pyExport new trace CRUD functions
pyproject.tomlAdd boto3 as optional dependency (for S3 tiering)

7. Risk Assessment

7.1 Database Size Growth

Risk: Traces are written for every LLM call. A busy workspace might produce thousands of traces per day.

Mitigation:

  • Traces are opt-in (TRACE_ENABLED=false by default).
  • Large content externalized to S3 (cheap, unbounded storage).
  • Configurable retention with time-based and count-based pruning.
  • The list API excludes content to avoid accidental large reads.
  • Monitoring: the content_size_bytes column allows easy reporting on storage consumption.

Estimate: A typical deriver trace is ~10KB of JSONB. At 100 traces/day/workspace, that is ~1MB/day/workspace inline, or ~30MB/month. For managed SaaS with 1000 workspaces, that is ~30GB/month before retention kicks in. With 90-day retention, steady-state is ~90GB — manageable for PostgreSQL with proper vacuuming.

7.2 Write Latency

Risk: Adding a database write after every LLM call increases latency in the deriver pipeline.

Mitigation:

  • The trace write is a single INSERT with no joins or subqueries.
  • The link_trace_to_observations() UPDATE is a simple indexed update.
  • Both operations are fast (sub-millisecond for typical sizes).
  • S3 writes happen synchronously but only for traces >50KB (rare for deriver, occasional for dreamer).
  • If latency becomes an issue, trace writes can be made asynchronous via a background task queue.

7.3 S3 Dependency

Risk: S3 adds a cloud dependency that may not be available in all self-hosted deployments.

Mitigation:

  • S3 is entirely optional. Without S3 configuration, all traces are stored inline.
  • The S3_ENDPOINT_URL setting supports S3-compatible stores (MinIO, Cloudflare R2, Backblaze B2).
  • A warning is logged when traces exceed the threshold but S3 is not configured.

7.4 Schema Migration

Risk: Adding a new table with foreign keys to workspaces, peers, and sessions requires those tables to exist.

Mitigation:

  • The migration only adds a new table — no existing table modifications.
  • Foreign keys reference existing tables and columns with well-established data.
  • The migration is idempotent (uses table_exists guard in downgrade).
  • Zero-downtime: the table can be created while the old code is still running; the old code simply does not write to it.

7.5 Backward Compatibility

Risk: Changing log_reasoning_trace() could break existing JSONL workflows.

Mitigation:

  • The JSONL path is preserved. The function writes to JSONL first (if configured), then to the database.
  • The TRACE_ENABLED flag is independent of REASONING_TRACES_FILE. Both can be active simultaneously.
  • New parameters use keyword-only arguments with defaults, so existing callers are unaffected.

8. Verification Plan

8.1 Unit Tests

TestFileWhat it validates
test_save_reasoning_tracetests/test_trace_storage.pyTrace saved to DB with correct fields
test_save_trace_disabledtests/test_trace_storage.pyReturns None when TRACE_ENABLED=false
test_link_trace_to_observationstests/test_trace_storage.pyobservation_ids updated after linking
test_externalize_thresholdtests/test_trace_storage.pyContent externalized when over threshold
test_externalize_no_s3tests/test_trace_storage.pyInline storage with warning when S3 not configured
test_fetch_externalized_contenttests/test_trace_storage.pyS3 content fetched and decompressed correctly
test_cleanup_time_basedtests/test_trace_storage.pyOld traces deleted by time-based retention
test_cleanup_count_basedtests/test_trace_storage.pyExcess traces deleted per workspace
test_cleanup_indefinitetests/test_trace_storage.pyNo deletion when policy is indefinite

8.2 Integration Tests

TestFileWhat it validates
test_deriver_produces_tracetests/test_trace_integration.pyDeriver LLM call creates a trace with correct agent_type, message_ids
test_deriver_trace_linked_to_observationstests/test_trace_integration.pyAfter deriver runs, trace.observation_ids contains created document IDs and documents have trace_id in internal_metadata
test_dialectic_produces_tracetests/test_trace_integration.pyDialectic chat creates a trace with tool call data
test_dreamer_produces_tracetests/test_trace_integration.pyDreamer specialist creates a trace with correct specialist type
test_summarizer_produces_tracetests/test_trace_integration.pySummarizer creates a trace without observation_ids

8.3 API Tests

TestFileWhat it validates
test_list_tracestests/test_trace_api.pyPaginated list with filters returns correct results
test_list_traces_agent_type_filtertests/test_trace_api.pyFiltering by agent_type works
test_list_traces_date_rangetests/test_trace_api.pycreated_after/created_before filtering
test_get_trace_detailtests/test_trace_api.pyFull trace returned with content
test_get_trace_404tests/test_trace_api.py404 for non-existent trace
test_get_trace_for_conclusiontests/test_trace_api.pyConclusion → trace lookup works
test_get_traces_for_messagetests/test_trace_api.pyMessage → traces lookup works
test_trace_content_excluded_from_listtests/test_trace_api.pyList endpoint does not include content field

8.4 Manual Verification

  1. Deploy with TRACE_ENABLED=true on a staging instance.
  2. Send messages via the API and verify traces appear in the database.
  3. Use the dialectic API and verify traces include tool calls and iteration data.
  4. Trigger a dream and verify specialist traces are created.
  5. Query the /conclusions/{id}/trace endpoint and verify the provenance chain is complete.
  6. Verify that REASONING_TRACES_FILE still produces JSONL output when configured.
  7. For S3: configure a MinIO instance, send a large multi-turn dialectic query, verify the trace content is externalized and retrievable.

9. Open Questions

Q1: Should the message_ids on the trace use public_id (nanoid) or internal id (bigint)?

Recommendation: Use public_id (nanoid strings). This is what the API exposes, what DocumentMetadata.message_ids currently stores as internal IDs (which should arguably be migrated), and what developers will use when querying. The GIN index on message_ids will use JSONB containment which works with string arrays.

Note: The existing DocumentMetadata.message_ids field uses internal integer IDs. This is an existing inconsistency that should be addressed separately. For traces, we use public IDs from the start for API consistency.

Q2: Should traces capture the full message history for multi-turn agents?

The full message history can be very large (100KB+ for long dialectic sessions). The current log_reasoning_trace() captures it. Recommendation: Capture it in the content.input.messages field. The S3 tiering handles the size concern. For the get_reasoning_trace agent tool, only the thinking content and tool calls are surfaced (not the full message history), keeping the tool output concise.

Q3: Should the agent_type CHECK constraint be enforced at the database level?

Recommendation: Yes, for data integrity. The set of agent types is small and stable. If a new agent type is added, a migration adds it to the constraint. This prevents garbage data from entering the table.

Q4: Should trace creation be synchronous or asynchronous?

Phase 1-3: Synchronous (write in the same request/task context). This is simpler, ensures traces are available immediately for provenance linking, and the latency impact is minimal (single INSERT).

If profiling in production shows trace writes adding unacceptable latency, Phase 5 can introduce an async write path using the existing queue infrastructure.

Q5: boto3 dependency — should it be optional?

Recommendation: Yes. Add it as an optional dependency group: pip install honcho[s3] or uv sync --extra s3. The S3 client initialization checks for the import and logs a clear error if S3 is configured but boto3 is not installed. This avoids adding a heavy dependency for deployments that do not need S3.

# pyproject.toml
[project.optional-dependencies]
s3 = ["boto3>=1.34"]

Q6: Should we add a run_id column for correlating traces across a multi-step operation?

For example, a dream run produces two specialist traces (deduction + induction). Today these are correlated via the DreamRunEvent.run_id in CloudEvents. Adding a run_id TEXT column (nullable) to reasoning_traces would allow easy correlation queries: “show me all traces from this dream run.” This is low-cost and high-value. Recommendation: add it in Phase 2.

-- Additional column
run_id TEXT,  -- Nullable. Correlates traces within a multi-step operation (dream run, etc.)
# Additional index
Index("ix_reasoning_traces_run_id", "run_id", postgresql_where=text("run_id IS NOT NULL")),

Graph View

Backlinks