---
contract_version: "2.0"
module: "core.conversation_api"
purpose: "HTTP/SSE server providing multi-turn conversation sessions with Worldtree agents for external clients"
depends_on:
  - "core.agent_turn"
  - "core.conversation_api.session"
  - "core.conversation_api.store"
  - "core.llm"
  - "core.persona"
  - "core.model_profiles"
  - "core.role_loader"
  - "core.agent_bus"
  - "core.agent_registry"
  - "core.heimdall"
  - "core.integration"
used_by:
  - "web app (external, #15)"
  - "TUI SSH client (external, #50)"
  - "core.matrix_bridge (internal)"
language: "python"
complexity: "high"
estimated_loc: 500
confidence: 0.95
assumptions:
  - "FastAPI + sse-starlette for HTTP/SSE transport"
  - "Single-process deployment (no horizontal scaling yet)"
  - "SQLite conversation store is sufficient for current scale"
  - "Agent configs are static after startup (no hot-reload)"
  - "API key auth (not JWT) — JWT is the web app's concern"
  - "Keys defined in defaults.yaml under conversation_api.api_keys, synced into Heimdall's user store at startup (idempotent)"
  - "All HTTP requests pass through Heimdall: get_security_context FastAPI dependency produces a SecurityContext per request, anonymous in dev mode (no api_keys configured)"
  - "Session ownership enforced via SecurityContext.user_id — cross-user access returns 404 (not 403) to avoid leaking session existence"
open_questions: []
---

## Module structure

| File | Responsibility |
|---|---|
| `service.py` | `ConversationService` class, `AgentContext` dataclass, helper functions (`_get_api_config`, `_resolve_llm_cfg`, `_load_system_prompt`, `_build_file_scopes_text`, `_inject_docstore`). No FastAPI dependency. |
| `api.py` | FastAPI app, route handlers, Pydantic request/response models, `get_security_context` auth dependency, `_bearer_scheme`, `lifespan`. Imports `ConversationService` from `service.py`. |
| `__init__.py` | Re-exports public API surface. `ConversationService` and `AgentContext` are re-exported from `service.py`; `app`, `main`, request models from `api.py`. |

Dependency direction: `api.py` → `service.py` (one-way). No reverse import from `service.py` into `api.py`.

## Context

The Conversation API is the standard interface for external clients to interact
with Worldtree agents. It provides session-based multi-turn conversations with
streaming responses via Server-Sent Events (SSE). All agent capabilities —
tool execution, model switching, persona expression — are available through
this single interface.

This is the protocol boundary between the agent runtime (Worldtree) and any
external consumer (web app, TUI, bridge). Clients only need to understand
HTTP + SSE; all agent complexity is encapsulated server-side.

## Data flow

**Input:**
- HTTP requests to REST endpoints (session CRUD, agent listing)
- HTTP POST with SSE response for message sending (streaming turns)

**Output:**
- JSON responses for REST endpoints
- SSE event streams for conversation turns (thinking, text, tool_start, tool_result, done, error). The `done` event carries a `usage: {prompt_tokens, completion_tokens, total_tokens, cached_input_tokens}` object sourced from `TurnResult` (Midgaard #4) — fields are present even when the provider reports zero.
- Probe responses: `/healthz` returns minimal liveness body `{"status": "ok"}`; `/readyz` returns `{"status": "ok", "version": ..., "git_sha": ...}` (200) when ready, or `{"status": "not_ready", "reason": ...}` (503) when not (Midgaard #2).

**Persistence:**
- `ConversationStore` (SQLite) — session metadata, message history
- In-memory session cache — active sessions for fast access

## Invariants

- **INV-001**: Every session belongs to exactly one agent; the agent_id is immutable after creation
- **INV-002 (REVISED #123)**: Session message history contains only user/assistant pairs. Tool-call intermediates (`tool_start` / `tool_result` SSE events) are by default NOT persisted; opt-in per-session persistence is allowed via the `record_tool_intermediates` flag on session create, storing **metadata only** (no args, no results) into the `tool_events` table — see INV-081..INV-087. Cancelled turns drop their partial assistant output by default; `?persist_partial=true` opts in to persisting one assistant message with `partial: true` metadata.
- **INV-003**: If a turn errors mid-stream, the user message that triggered it is rolled back from history. Cancellation is NOT an error: the user message stays in history; only the in-flight assistant response is discarded.
- **INV-004**: SSE streams always terminate with one of `done`, `error`, or `cancelled` events
- **INV-005**: Model switches during a turn update both the provider and the tool context atomically
- **INV-006**: Persona affect context is injected fresh before each turn; post-turn appraisal is **scheduled** (not awaited) after each successful turn. Appraisal is fire-and-forget — the SSE `done` event does NOT block on appraisal completion (Issue #177 Phase A; see `persona_registry.contract.md` invariants 6-10). Appraisal does NOT run after cancellation — appraisal requires a meaningful turn outcome. Mood state MAY lag by one turn relative to the conversation, which is acceptable per Issue #151's `async-off-Finishing-path` Vor decision.
- **INV-007**: When auth is enabled, sessions are scoped to the user who created them; users cannot access other users' sessions
- **INV-008**: When auth is disabled (no API keys configured), all endpoints are open — dev mode preserves current behavior
- **INV-009**: Session access denials return 404 (not 403) to avoid leaking session existence
- **INV-010 (cancel-cooperative)**: Cancellation propagates through agent_bus_v2's `TaskHandle.cancel()` and a per-turn `asyncio.Event` checked between agent_turn events. Tools currently executing run to completion before cancel takes effect; cancel applies before the NEXT LLM call, not mid-tool. This preserves filesystem and shell-state integrity at the cost of an unbounded (but bounded by tool timeout) tail latency.
- **INV-011 (cancel-deafen)**: After a turn is cancelled, the streaming generator MUST NOT yield further events from any source — late sub-pipeline tool results, skald-routed messages from subordinate agents, or completing LLM HTTP calls. The turn is "deaf" from cancel-time forward; subsequent events for the cancelled turn_id are discarded silently.
- **INV-012 (cancel-idempotent)**: Cancel is idempotent. A second cancel for the same turn_id while cancellation is in flight returns `200 {cancelled: false, reason: "already_cancelling"}`. Cancel after the turn has already completed (success OR error) returns `200 {cancelled: false, reason: "already_complete"}`. Both no-op cancels are still audited.
- **INV-013 (cancel-audited)**: Every cancel call writes an entry to the Heimdall audit store with metadata `{user_id, session_id, turn_id, was_in_flight, tool_calls_completed, partial_chars_emitted, cancelled_at}`. Partial assistant content is NEVER logged — only the metadata about it.
- **INV-014 (turn-id-public)**: Every SSE event from `stream_turn` carries a composite `{turn_id}:{seq}` id. `turn_id` is the integer from SQLite `turns.id`; `seq` is a per-turn monotonic integer starting at 1 (resets per turn). The composite id is the SSE `id:` wire field and the event dict's `"id"` key. Clients parse `turn_id` from the prefix for cancellation. The `turn_id` integer also appears in the `done`/`cancelled` event's JSON body.
- **INV-015 (readiness-flag)**: `ConversationService._started` is `False` at `__init__`, set to `True` at the very end of `startup()` after all initialisation succeeds, and set to `False` at the very start of `shutdown()` before the store closes. `/readyz` returns 200 only when this flag is True (and the store and agent contexts are populated as defence-in-depth). Liveness (`/healthz`) is independent of this flag — it answers process responsiveness, not service readiness.
- **INV-022 (rate-limit-pre-stream)**: Rate checks fire in the `send_message` handler after `_check_session_access` and BEFORE `stream_turn` is entered. Denied requests do NOT enter the cancel registry, do NOT increment turn_id, and do NOT touch session.messages. The deny response is a `JSONResponse(429)` returned before the SSE generator is constructed. Probe endpoints (`/healthz`, `/readyz`), admin endpoints, and GET /me are exempt — they are separate route handlers that never call `check_rate_limits`. GET /me's exemption is structural (no call in its handler), not via an explicit allowlist.
- **INV-023 (token-rate-post-charge)**: Per-user token rate is post-charged after the TurnComplete event in `stream_turn`. Debt is allowed — a single bad turn can over-spend by orders of magnitude, leaving the bucket in negative state. The NEXT turn's pre-check (`peek`) sees the new TAT and returns 429. Charging only fires from the TurnComplete branch; cancel and exception paths do NOT charge. Operators wanting hard per-turn ceilings should pair with the cost-cap (#114).
- **INV-016 (replay-buffer-bounded)**: `_replay_buffer[turn_id]` is a `collections.deque(maxlen=10)`. Lifetime equals turn lifetime (created in `stream_turn` at turn start, popped in its `finally` block). No time-based expiry. Cleanup belongs exclusively to `stream_turn`'s `finally` block.
- **INV-017 (queue-source-of-truth)**: Every event yielded by `stream_turn` MUST also be published to `_event_queues[turn_id]` via `put_nowait`, and publication MUST precede the `yield`. The `_END` sentinel MUST be pushed after all terminal events (`done`/`cancelled`/`error`) unconditionally on all paths. The queue is the source of truth for the HTTP reconnect consumer (`attach_turn`).
- **INV-018 (turn-session-binding)**: `_turn_session[turn_id]` MUST equal the `session_id` that initiated the turn. `attach_turn` MUST verify this binding (before accessing the replay buffer) and raise `PermissionError` on mismatch. `PermissionError` routes to HTTP 404 per INV-009. Cleanup belongs exclusively to `stream_turn`'s `finally` block.
- **INV-019 (admin-key-cleartext-once)**: The cleartext API key appears in exactly one HTTP response body — the 201 body of `POST /admin/keys`. It is NEVER persisted to the database, NEVER logged, and NEVER re-derivable from any stored field. `key_suffix` (last 8 chars) is stored for leak detection only. Operators MUST configure proxy-level redaction for `POST /admin/keys`.
- **INV-020 (rotate-atomic)**: The rotation of an API key — INSERT of the new key row + UPDATE of the old key row with `superseded_at`, `superseded_by_key_id`, and `grace_seconds` — MUST execute as a single SQLite transaction. Partial state (new key visible without old key marked superseded, or vice versa) MUST NOT exist. On any failure, both writes roll back.
- **INV-021 (rotate-grace-captured)**: `grace_seconds` is captured per-record at rotation time and stored on the old key row. The lazy-disable path in `Authenticator._authenticate_api_key` reads `key_record.grace_seconds` from the stored value, NOT from the live `conversation_api.key_rotation_grace_seconds` config. Subsequent config changes do NOT retroactively shorten or extend in-progress grace windows.
- **INV-029 (agent-metadata-additive)**: GET /agents response is ADDITIVE. New optional fields are omitted when null or empty; they are never emitted as null. The `agent.ui_hints` config slot is additive — once a subfield is documented, it is never removed. Empty list for `capabilities` or `supported_models` is equivalent to absent: both produce field omission in the response (normalize to `None` before serialization; do not rely on `exclude_none` alone for lists).
- **INV-026 (me-key-resolution-best-effort)**: GET /me's `key_id`/`key_label`/`key_created_at` fields identify the most-recently-used active key for the authenticated user, resolved by sorting `get_api_keys_for_user()` results by `last_used_at DESC NULLS LAST`. Heimdall stamps `last_used_at` synchronously on every successful auth, so the first row is overwhelmingly likely to be the authenticating key. The invariant contracts the resolution rule, not absolute identity. Clients MUST treat `key_id` as best-effort. Optional fields (`display_name`, `user_created_at`, `key_*`, `superseded_*`, `grace_seconds`) are OMITTED (not null) when not applicable; `tier` is always present, including `"unknown"` on Heimdall failure.
- **INV-024 (session-mutation-ownership-only)**: `PATCH /sessions/{id}` requires session ownership only (same `_check_session_access` check as GET). No new Heimdall scope is required. Anonymous-mode PATCH is allowed (mirrors today's anonymous-mode GET behavior). Ownership denials return 404 (not 403) per INV-009.
- **INV-025 (metadata-internal-namespace)**: Metadata keys prefixed with `_` are reserved for Worldtree-internal use. Client writes to such keys are rejected with 422 at the Pydantic validator in `api.py`. Internal callers that bypass the API layer (e.g., `stream_turn` writing `{"model": used_model}`) are exempt — the service layer does NOT re-apply the namespace check. The `model` key (no underscore) is the current internal key and is NOT renamed in this commit.
- **INV-030 (error-code-stable)**: Every error response (REST 4xx/5xx + SSE `error` events) carries a flat snake_case `error_code` from the `ErrorCode` enum defined in `core/conversation_api/errors.py`. Codes are additive — an existing code value is never repurposed with different semantics. New codes may be added without a contract amend. Both `api.py` and `service.py` import `ErrorCode` and `_err` from `errors.py`; there is no back-reference from `service.py` to `api.py`.
- **INV-031 (error-code-shape)**: REST error responses use `{"detail": {"error_code": "<code>", "message": "<human>", **extra}}`. SSE error events use `{"type": "error", "error_code": "<code>", "message": "<human>"}` — no `detail` wrapper. Existing structured fields (e.g., `scope`/`retry_after_s` from #117, `buffered_from_seq`/`turn_id` from #113) are preserved verbatim as siblings of `error_code` and `message`. Success responses (2xx) are unchanged.
- **INV-032 (httpexception-handler-status-mapping)**: The custom `HTTPException` handler MUST inspect `exc.status_code` when `detail` is not already a coded dict and apply the mapping table: `405 → method_not_allowed`, `400 → malformed_request`. All other uncoded status codes fall back to `internal_error` with a `logger.warning(...)` call. The handler MUST NOT produce `error_code='internal_error'` for 405 or 400 responses.
- **INV-033 (phase-column-canonical-vocab)**: The `turns.phase` TEXT column holds exactly one of `succeeded`, `failed`, `cancelled`, `stalled`, or NULL (for in-flight turns; pre-migration rows are populated per INV-034). The value is written exactly once per turn, at terminal transition, in the same `complete_turn(...)` call that sets `completed_at`. NULL on a row whose `completed_at` is non-null indicates a contract violation — every terminal write path MUST pass `phase`.
- **INV-034 (phase-backfill-derivation)**: The one-shot `_migrate_phase_column()` step in `ConversationStore._migrate()` adds the `phase` column when absent and populates pre-existing rows by deriving the value with the following precedence (first match wins): `cancelled = 1 → 'cancelled'`; ELSE `error IS NOT NULL → 'failed'`; ELSE `completed_at IS NOT NULL → 'succeeded'`; ELSE → `'stalled'` (the row was orphaned by a prior server restart). Migration is idempotent — re-running on a database that already has the column is a no-op; values written by a prior run are NOT re-derived.
- **INV-035 (stall-resets-on-event)**: Inside `stream_turn`'s event loop, every event yielded by the `agent_turn` engine — `ThinkingDeltaEvent`, `ContentDeltaEvent`, `ToolCallStartEvent`, `ToolCallResultEvent`, `TurnCompleteEvent` — MUST reset the stall watchdog timer for the active turn BEFORE the event is yielded to the client. Reset is via `_reset_stall_timer(turn_id, stall_timeout_s)`. Missing the reset on any event type creates a false-stall hole; in particular, extended-thinking models (zai/glm) emit `ThinkingDeltaEvent` only for long stretches and would trigger spurious stalls without this invariant.
- **INV-036 (stall-uses-cancel-path)**: When the stall watchdog fires for `turn_id`, it MUST reuse the cancel-registry mechanism — `_CancelRequest.event.set()` on the same registry entry — so the streaming generator's loop-exit code is one branch. The two paths are distinguished by `_CancelRequest.reason`: `"user_cancel"` set by `cancel_turn`, `"stall"` set by `_on_stall`. Whichever sets `reason` first wins (later setter is a no-op). Terminal phase persisted to `turns.phase` differs accordingly: `'cancelled'` for user-cancel, `'stalled'` for stall. The SSE `cancelled` event carries `reason` reflecting the winning path. Partial assistant output is dropped UNCONDITIONALLY on stall — `persist_partial` is only honoured on the user-cancel path; stall is system-initiated and the user never opted in.
- **INV-037 (stall-timer-bound-to-turn)**: The stall watchdog `asyncio.TimerHandle` is owned by the turn's `_CancelRequest` instance via a new field, `stall_timer`. At most ONE live `TimerHandle` exists per `turn_id` at any time. The handle MUST be cancelled (`handle.cancel()`) in `stream_turn`'s `finally` block on every path — success, error, user cancel, stall, exception. New turns get a fresh handle; reusing a handle across turns is forbidden. Failure to cancel leaks one timer per orphaned turn — bounded only by process restart.
- **INV-038 (log-correlation-prefix)**: Log lines emitted from `stream_turn`, `cancel_turn`, `attach_turn`, and `_on_stall` carry the contextual prefix `[{session_prefix}/{turn_id}/{seq}]`, where `session_prefix` is the first 8 hex chars of the session UUID with dashes stripped, `turn_id` is the integer turn id, and `seq` is the most recently emitted event's per-turn seq (or `0` if no events yet). The prefix is logs-only — it does NOT appear in SSE events, HTTP response bodies, or audit records. Implementation mechanism (LoggerAdapter, helper class, f-string composition) is unconstrained; the contract is on the format. When `session_id` is shorter than 8 hex chars (defensive — should not occur for UUIDs), the full `session_id` is used.
- **INV-039 (pagination-cursor-opacity)**: Pagination cursors are opaque `v1.<base64url(json)>` tokens. The `v1.` prefix is a literal version handshake. The JSON body is server-internal and MUST NOT be exposed, parsed, or constructed by clients. Future format changes bump to `v2.`; old `v1.` cursors are invalid after version upgrade. Implementation lives in `core/conversation_api/pagination.py` (`Cursor.encode`/`Cursor.decode`).
- **INV-040 (pagination-cursor-exclusive)**: Pagination cursors are forward-only and exclusive of the anchor row. The anchor is the last visible row of the current page; the next page returns rows strictly _after_ the anchor. `GET /sessions` uses `(created_at, session_id) < (anchor_c, anchor_i)` (lexicographic, DESC order). `GET /sessions/{id}/messages` uses `seq > anchor_s`. Re-fetching the same cursor always yields the same result set relative to surviving data (deleted anchor → fewer items, no error).
- **INV-041 (pagination-page-bounds)**: `?limit` must satisfy `1 ≤ N ≤ 200` inclusive. Values of 0, negative, non-integer, or > 200 yield 422 with `error_code=validation_failed`. Default is 50. Enforced declaratively via FastAPI `Query(ge=1, le=200, default=50)`.
- **INV-042 (pagination-end-of-list)**: End-of-list detection uses the LIMIT n+1 trick: the server fetches `limit+1` rows. If `len(rows) > limit`, it truncates to the first `limit` rows and emits `next_cursor` for the last visible row. If `len(rows) ≤ limit`, `next_cursor` is null. Cost: one extra row per page, no separate `COUNT(*)`.
- **INV-043 (pagination-shape-divergence)**: `GET /sessions` emits `{items, next_cursor}`; `GET /sessions/{id}/messages` emits `{session_id, items, next_cursor}`. The `session_id` key is preserved in the messages response for client-side response indexing. `message_count` is absent from both paginated responses.
- **INV-044 (pagination-filter-agnostic)**: Cursors encode only sort-key fields — `(created_at, session_id)` for sessions, `seq` for messages — never filter state. `?include_archived` and future per-request filters are re-applied on every request independently of the cursor. A filter change mid-iteration is well-defined: the new filter applies from the anchor position forward.
- **INV-045 (pagination-predictable-schema)**: `next_cursor` is ALWAYS present in paginated list responses (null when last page, string otherwise). This diverges from INV-029's omit-when-null convention: cursor presence is a structural signal, not a cosmetic field, so explicit null is required.
- **INV-046 (admin-events-envelope-stable)**: The `AdminEvent` envelope is exactly `{id: int, type: str, timestamp: iso8601_z, data: dict}`. Additive field-additions are allowed, but the four core fields must never be reordered or renamed. This is a public contract for the `/admin/events` stream; changes require a major version bump in the SSE spec.
- **INV-047 (admin-events-id-monotonic-per-process)**: Event ids auto-increment starting at 1 per process start, reset on restart. The first event emitted by `EventBus.start()` has id=1 (the `system.startup` event). Ids are strictly monotonically increasing within a process run; no gaps within a run are possible except across buffer eviction.
- **INV-048 (admin-events-after-commit)**: Events are emitted AFTER the canonical SQL commit — never before. `session.created` fires after `INSERT INTO sessions` returns; `turn.completed` fires after `UPDATE turns SET phase='succeeded'` returns; `key.rotated` fires after `supersede_api_key` returns. Consumers can always query the corresponding row immediately upon receiving the event.
- **INV-049 (admin-events-pii-discipline)**: Events carry IDs (`session_id`, `turn_id`, `key_id`, `user_id`, `agent_id`) and small metadata (`phase`, `error` string ≤ 200 chars, `duration_ms`, `changed_fields` list) only. Events NEVER carry message content, tool-call arguments, tool-call results, cleartext API keys, partial assistant output, persona affect state, LLM tokens, or any other sensitive content. For error strings the PII rule is: emit `ExceptionType: truncated_message[:200]` — the full traceback stays in the logger, not in the event.
- **INV-050 (admin-events-buffer-bounded)**: The in-memory ring buffer has `maxlen=1000` events (`collections.deque(maxlen=1000)`). No persistence; buffer resets on process restart. When the buffer is full, the oldest event is silently evicted; `buffer_floor` advances accordingly.
- **INV-051 (admin-events-per-consumer-queue)**: Each subscriber has a bounded `asyncio.Queue(maxsize=1000)`. On overflow: the oldest item is popped from the subscriber queue, a `system.events_dropped` event describing the gap (`count`, `gap_first_id`, `gap_last_id`) is enqueued in its place, and the incoming event is dropped. Drop notifications are best-effort — a persistently-slow consumer can lose them.
- **INV-052 (admin-events-scope)**: `GET /admin/events` requires `admin.events.read` scope; the connection itself is audited as `conversation_api:admin:events:connect` with the actor's `user_id`; per-event emission is NOT audited. The endpoint is rate-limit exempt (structural exemption, same as `/healthz`). Missing scope returns 403 `auth_scope_denied`; missing bearer returns 401.

## Constraints

- **[security]** API key auth via `Authorization: Bearer <key>` header; keys defined in defaults.yaml with user_id mapping. Dev mode (no keys) = open access. CORS origins configurable, defaults to `["*"]` in dev mode only. Key comparison uses `hmac.compare_digest` for timing safety.
- **[performance]** Agent contexts are built once at startup; per-turn overhead is system prompt assembly + LLM call
- **[compatibility]** SSE event format is the public contract — field names and types must not change without versioning
- **[style]** All endpoints return JSON; SSE data fields are JSON-serialised dicts

## Resume semantics

Sessions are persistent via SQLite. On server restart, existing sessions are
loadable from the store (lazy-loaded on first access). In-memory cache is
rebuilt on demand, not at startup.

```contract
FN ConversationService.startup() -> None
BRIEF: Discover agents, build per-agent contexts, initialise shared infrastructure
PRE: [PRE-001 hard] project root contains agents/ directory -- assert (root / "agents").is_dir()
PRE: [PRE-002 hard] config/providers.yaml exists and is valid YAML -- assert providers_path.exists()
POST: [POST-001 state_change] _agent_contexts populated for all valid agents -- assert len(self._agent_contexts) > 0
POST: [POST-002 state_change] _persona_registry has configs for all loaded agents -- all agent_ids registered
POST: [POST-003 state_change] _store is initialised (SQLite connection open) -- assert self._store is not None
POST: [POST-004 side_effect] all BaseAgent instances started and registered on bus -- bus.agent_ids matches contexts
ERRORS:
  ImportError -> log warning, skip that agent, continue with others
  YAML parse error -> log warning, skip that agent
  Provider creation failure -> log warning, skip that agent
STEPS:
  1. [setup] Load providers.yaml, expand env vars
  2. [setup] Create AgentRegistry, run discover(), log findings
  3. [setup] Create AgentBus
  4. [loop] FOR EACH agent directory with agent.py + config.yaml:
     - Import agent module, find BaseAgent subclass
     - Instantiate and start agent, register on bus
     ON ImportError or startup failure:
       - LOG warning, skip agent
  5. [loop] FOR EACH started agent:
     - Resolve LLM config (honour llm_profiles.default)
     - Load tool schemas, modules, context via role_loader
     - Inject shared state into tool context (bus, registry, llm_config, memory)
     - Build system prompt (base + registry + file scopes + model profiles)
     - Create LLM provider
     - Store as _AgentContext
  6. [setup] Create PersonaRegistry, register all agent configs
  7. [setup] Wire persona registry to bus for deep inter-agent calls
  8. [setup] Create ConversationStore (SQLite)
  9. [cleanup] LOG ready message with agent count
TESTS:
  all_agents_loaded [happy]: valid agents dir → all agents in _agent_contexts
  bad_agent_skipped [error]: one agent has broken config → others still load
  no_agents [boundary]: empty agents dir → service starts with 0 agents
  missing_providers [error]: no providers.yaml → startup fails
```

```contract
FN ConversationService.available_agents() -> list[dict]
BRIEF: Return list of agent summaries with richer metadata for client discovery
PRE: [PRE-001 soft] startup() has been called -- returns empty list if not
POST: [POST-001 return_value] each dict always has agent_id, name, description -- guaranteed present
POST: [POST-002 return_value] optional fields (version, capabilities, supported_models, persona_traits, ui_hints) present only when populated -- never null in response
POST: [POST-003 return_value] empty list for capabilities or supported_models is normalized to None -- field omitted, not emitted as []
STEPS:
  1. [sequential] Iterate _agent_contexts
  2. [sequential] For each AgentContext, build dict with agent_id, name, description always present
  3. [sequential] Append version, capabilities, supported_models, persona_traits, ui_hints when non-null and non-empty
  4. [sequential] RETURN list of dicts
TESTS:
  two_agents [happy]: two agents loaded → returns both with all fields
  empty [boundary]: no agents → returns []
  full_shape [happy]: agent with all metadata → all 8 fields present in response
  minimum_shape [edge]: agent with only required fields → exactly 3 fields in response
  empty_list_omitted [edge]: capabilities=[] → field absent from response
```

```contract
FN ConversationService.create_session(agent_id: str) -> ConversationSession
BRIEF: Create a new conversation session for the given agent
PRE: [PRE-001 hard] agent_id exists in _agent_contexts -- raise ValueError if not
POST: [POST-001 return_value] returned session has matching agent_id and a UUID session_id -- assert session.agent_id == agent_id
POST: [POST-002 state_change] session stored in _sessions cache -- assert session_id in self._sessions
POST: [POST-003 state_change] session persisted to SQLite store -- store.get(session_id) is not None
STEPS:
  1. [setup] Validate agent_id exists in _agent_contexts
     IF not found:
       RAISE ValueError("Unknown agent: {agent_id}")
  2. [sequential] Create ConversationSession.new(agent_id) — generates UUID
  3. [sequential] Store in _sessions dict
  4. [sequential] Persist to _store if available
  5. [cleanup] RETURN session
TESTS:
  valid_agent [happy]: known agent_id → session with UUID and correct agent_id
  unknown_agent [error]: bad agent_id → raises ValueError
  multiple_sessions [happy]: create two for same agent → different session_ids
```

```contract
FN ConversationService.get_session(session_id: str) -> ConversationSession | None
BRIEF: Retrieve a session by ID, checking cache then persistent store
PRE: [PRE-001 soft] session_id is a non-empty string -- return None for empty
POST: [POST-001 return_value] returns session if found, None if not -- type is ConversationSession or None
POST: [POST-002 state_change] if loaded from store, session is cached in _sessions -- assert in _sessions after load
STEPS:
  1. [branch] IF session_id in _sessions cache:
     - RETURN cached session
  2. [branch] IF _store is available:
     - Query store for session info
     IF found:
       - Load messages from store
       - Reconstruct ConversationSession
       - Cache in _sessions
       - RETURN session
  3. [sequential] RETURN None
TESTS:
  cached [happy]: session in cache → returns it directly
  from_store [happy]: session not cached but in store → loads and caches
  not_found [boundary]: no such session → returns None
```

```contract
FN ConversationService.delete_session(session_id: str) -> bool
BRIEF: Remove a session from cache and persistent store
POST: [POST-001 return_value] returns True if session existed, False if not -- bool
POST: [POST-002 state_change] session removed from _sessions cache -- assert session_id not in _sessions
POST: [POST-003 state_change] session removed from store -- store.get(session_id) is None
STEPS:
  1. [sequential] Pop from _sessions cache, note if existed
  2. [branch] IF _store available:
     - Delete from store, note if existed
  3. [sequential] RETURN True if existed in either location
TESTS:
  exists [happy]: existing session → True, removed from both
  not_found [boundary]: no such session → False
```

```contract
FN ConversationService.stream_turn(
    session_id: str,
    content: str,
    sender_id: str | None = None,
) -> AsyncIterator[dict[str, Any]]
BRIEF: Execute one conversation turn, yielding SSE events as they occur
PRE: [PRE-001 hard] session_id exists in _sessions -- yield error event if not
PRE: [PRE-002 hard] agent context exists for session's agent_id -- yield error event if not
PRE: [PRE-003 soft] content is non-empty -- empty content may produce empty response
POST: [POST-001 side_effect] user message appended to session history -- session.messages[-2] is user msg
POST: [POST-002 side_effect] assistant response appended to session history -- session.messages[-1] is assistant msg (omitted when cancelled and persist_partial is false)
POST: [POST-003 side_effect] messages persisted to store -- store has both messages (subject to POST-002)
POST: [POST-004 side_effect] persona appraisal is SCHEDULED post-turn -- update_after_turn called and returns immediately; appraisal runs as a background task (NOT awaited; NOT called when turn cancelled, per INV-006 and Issue #177 Phase A)
POST: [POST-005 return_value] stream ends with done, error, or cancelled event -- last yielded event type is "done", "error", or "cancelled"
POST: [POST-006 side_effect] on error, user message is rolled back -- session.messages unchanged from pre-call. Cancellation does NOT roll back per INV-003.
POST: [POST-007 side_effect] every yielded event carries the integer turn_id as SSE id field -- per INV-014
POST: [POST-008 state_change] when cancelled, _store.complete_turn(turn_id, error="cancelled") records the run -- audit + run-tracking
ERRORS:
  Session not found -> yield {"type": "error"} and return
  Agent not available -> yield {"type": "error"} and return
  LLM/tool exception -> yield {"type": "error"}, roll back user message
  CancelledError raised mid-loop -> yield {"type": "cancelled", "turn_id": N, "reason": "user_cancel"}, do NOT roll back user message
STATE: idle -> streaming -> idle | streaming -> cancelling -> idle
STEPS:
  1. [setup] Look up session in _sessions cache
     IF not found:
       YIELD error event, RETURN
  2. [setup] Look up agent context for session.agent_id
     IF not found:
       YIELD error event, RETURN
  3. [sequential] Append user message to session.messages and persist to store
  4. [setup] Call _store.start_turn() to obtain turn_id (existing behavior — see api.py:623)
  5. [setup] Register turn in cancel registry: cancel_registry[turn_id] = asyncio.Event()
  6. [setup] Set conversation_session_id in tool context
  7. [sequential] Build working message list (copy of session.messages)
  8. [sequential] Inject persona affect context into system prompt
  9. [sequential] Create AgentTurnEngine with provider, tools, on_model_switch callback
  10. [loop] FOR EACH event from engine.run_turn_streaming(working):
      - Tag event with turn_id as SSE id field per INV-014
      - ThinkingDeltaEvent -> YIELD {"type": "thinking", "content": ...}
      - ContentDeltaEvent -> YIELD {"type": "text", "content": ...}; track partial_chars_emitted
      - ToolCallStartEvent -> YIELD {"type": "tool_start", "name": ..., "arguments": ...}
      - ToolCallResultEvent -> YIELD {"type": "tool_result", "name": ..., "result": ..., "duration_ms": ...}; tool_calls_completed += 1
      - TurnCompleteEvent -> append assistant message, persist, run appraisal, YIELD done event with turn_id
      AFTER each event yielded, check cancel_registry[turn_id].is_set():
        IF set:
          - YIELD {"type": "cancelled", "turn_id": N, "reason": "user_cancel"}
          - Call _store.complete_turn(turn_id, error="cancelled", duration_ms=elapsed)
          - Write Heimdall audit entry per INV-013
          - IF persist_partial flag set: append assistant message with partial=true metadata
          - BREAK out of loop (per INV-011 deafen)
  11. [error_handler] ON CancelledError:
      - YIELD cancelled event (same as above)
      - Per INV-003: do NOT roll back user message
  12. [error_handler] ON other exception:
      - LOG exception
      - Roll back user message from session.messages and store
      - Call _store.complete_turn(turn_id, error=str(exc))
      - YIELD {"type": "error", "message": str(exc)}
  13. [cleanup] Always: cancel_registry.pop(turn_id, None) — per INV-011, defensive
TESTS:
  simple_response [happy]: user says "hello" -> text events + done event; 2 messages in history; SSE id field present
  tool_use [happy]: agent uses a tool -> tool_start + tool_result + text + done events
  thinking [happy]: thinking model -> thinking events before text events
  unknown_session [error]: bad session_id -> single error event
  unknown_agent [error]: session references missing agent -> error event
  llm_error [error]: provider raises -> error event, user message rolled back
  model_switch [edge]: agent switches model mid-turn -> provider updated, response completes
  cancel_mid_stream [happy]: cancel between events -> cancelled event yielded, loop exits, user message NOT rolled back
  cancel_drops_partial [happy]: cancel after partial text -> no assistant message persisted (default persist_partial=false)
  cancel_persists_partial [edge]: cancel with persist_partial=true -> one assistant message with partial=true metadata
  cancel_appraisal_skipped [edge]: cancelled turn -> update_after_turn NOT called per INV-006
  late_event_after_cancel [edge]: sub-pipeline emits event after cancel -> event discarded silently per INV-011
  turn_id_in_sse [happy]: every yielded event has integer SSE id matching the turn_id
```

```contract
FN ConversationService.update_session(
    session_id: str,
    patch: dict[str, Any],
    *,
    user_id: str | None = None,
) -> dict[str, Any]
BRIEF: Apply a partial update to a session; returns the full updated to_info() shape
PRE: [PRE-001 hard] session_id exists -- raises ValueError("Session ... not found") if not
PRE: [PRE-002 hard] if "metadata" in patch, post-merge serialized JSON <= 16 KiB -- raises ValueError("metadata size cap exceeded")
POST: [POST-001 return_value] returns updated session's to_info() -- dict with session_id, name, archived, tags, metadata, ...
POST: [POST-002 state_change] only fields present in patch are mutated -- fields absent from patch are unchanged
POST: [POST-003 state_change] metadata merge: keys with None value deleted, others added/replaced
POST: [POST-004 state_change] tags replace: patch["tags"] is the new complete list
POST: [POST-005 state_change] in-memory cache evicted after store update; reloaded via get_session()
ERRORS:
  session not found -> raise ValueError("Session ... not found")
  metadata size cap -> raise ValueError("metadata size cap exceeded (16 KiB post-merge)")
NOTES:
  - Namespace check (_-prefix) NOT applied here; that lives only in UpdateSessionRequest validator at api.py
  - Internal callers (e.g. stream_turn via update_metadata) bypass api.py and are exempt from namespace check
  - GET /sessions default-hides archived (include_archived=False); direct GET /sessions/{id} is unaffected
TESTS:
  name_only [happy]: patch={"name": "x"} -> name updated, archived/tags/metadata unchanged
  tags_replace [happy]: existing tags ["a"]; patch {"tags": ["b"]} -> tags == ["b"]
  metadata_merge_null_delete [edge]: existing {"a":1,"b":2}; patch {"b":null,"c":3} -> {"a":1,"c":3}
  metadata_size_cap_exceeded [error]: post-merge > 16 KiB -> ValueError
  metadata_size_reduced_ok [edge]: post-merge that deletes keys brings total under cap -> 200
  empty_patch_noop [edge]: patch={} -> to_info() returned unchanged
  not_found [error]: unknown session_id -> ValueError
```

```contract
FN list_agents() -> list[dict]
BRIEF: GET /agents — return available agents with richer metadata
POST: [POST-001 return_value] HTTP 200 with list of agent dicts -- each always has agent_id, name, description; optional fields present when populated, omitted when null/empty (INV-029)
POST: [POST-002 return_value] response serialized via AgentInfoResponse.model_dump(exclude_none=True) -- no null values emitted
STEPS:
  1. [sequential] Delegate to _service.available_agents()
  2. [sequential] Map each dict to AgentInfoResponse, serialize via model_dump(exclude_none=True)
  3. [sequential] RETURN list of serialized dicts
TESTS:
  ok [happy]: agents loaded → 200 with agent list containing richer metadata
  minimum_shape [edge]: agent with no extras → 200 with exactly {agent_id, name, description}
  full_shape [happy]: agent with all metadata → 200 with all 8 fields
```

```contract
FN create_session(body: CreateSessionRequest) -> dict
BRIEF: POST /sessions — create a new conversation session
PRE: [PRE-001 hard] body.agent_id is a known agent -- return 404 if not
POST: [POST-001 return_value] HTTP 201 with session info dict -- has session_id, agent_id
ERRORS:
  ValueError (unknown agent) -> HTTP 404
STEPS:
  1. [sequential] Call _service.create_session(body.agent_id)
     ON ValueError:
       RAISE HTTPException(404)
  2. [sequential] RETURN session.to_info() with 201 status
TESTS:
  ok [happy]: known agent → 201 with session_id
  unknown [error]: bad agent → 404
```

```contract
FN send_message(session_id: str, body: SendMessageRequest) -> EventSourceResponse
BRIEF: POST /sessions/{id}/messages — send message and stream SSE response
PRE: [PRE-001 hard] session exists -- return 404 if not
POST: [POST-001 return_value] returns SSE stream that yields JSON events -- content-type is text/event-stream
POST: [POST-002 side_effect] stream delegates to service.stream_turn -- all turn logic applies
ERRORS:
  Session not found -> HTTP 404
STEPS:
  1. [setup] Look up session via _service.get_session
     IF not found:
       RAISE HTTPException(404)
  2. [sequential] Create async generator wrapping _service.stream_turn
  3. [sequential] RETURN EventSourceResponse
TESTS:
  ok [happy]: valid session + message → SSE stream with done event
  not_found [error]: bad session_id → 404
```

```contract
FN get_current_user(
    credentials: HTTPAuthorizationCredentials | None,
) -> str | None
BRIEF: FastAPI dependency — extract and validate API key from Authorization header
PRE: [PRE-001 soft] credentials extracted by HTTPBearer(auto_error=False) -- None if no header
POST: [POST-001 return_value] returns user_id if auth enabled and valid key -- string
POST: [POST-002 return_value] returns None if auth disabled (dev mode) -- None
POST: [POST-003 exception] raises 401 if auth enabled but key missing or invalid -- HTTPException
STEPS:
  1. [branch] IF no API keys configured (dev mode):
     - RETURN None
  2. [branch] IF credentials is None:
     - RAISE HTTPException(401, "Missing Authorization header")
  3. [loop] FOR EACH key entry in config:
     IF hmac.compare_digest(entry.key, token):
       - RETURN entry.user_id
  4. [error_handler] No match found:
     - RAISE HTTPException(401, "Invalid API key")
TESTS:
  dev_mode [happy]: no keys configured → returns None, no auth required
  valid_key [happy]: valid Bearer token → returns user_id
  invalid_key [error]: wrong token → 401
  missing_header [error]: no Authorization header → 401
  timing_safe [security]: key comparison uses hmac.compare_digest
```

```contract
FN ConversationService.cancel_turn(
    session_id: str,
    turn_id: int,
    user_id: str | None = None,
    persist_partial: bool = False,
) -> dict
BRIEF: Server-side cancel of an in-flight turn; idempotent; audit-logged.
PRE: [PRE-001 hard] session_id exists in _sessions -- raise ValueError if not (HTTP layer translates to 404)
PRE: [PRE-002 hard] turn_id exists in turns table AND belongs to session_id -- raise ValueError if not (translates to 404)
PRE: [PRE-003 soft] when auth is enabled, user_id matches the session owner per INV-007 -- raise PermissionError if not (translates to 404 per INV-009)
POST: [POST-001 return_value] returns {turn_id, cancelled: bool, reason: str | None, partial_message_id: str | None}
POST: [POST-002 state_change] on first cancel of an in-flight turn: cancel_registry[turn_id].set() and _store.complete_turn(turn_id, error="cancelled") called
POST: [POST-003 side_effect] Heimdall audit entry written per INV-013 on every call (including no-op cancels)
POST: [POST-004 return_value] second call returns {cancelled: false, reason: "already_cancelling"} without raising -- idempotent per INV-012
POST: [POST-005 return_value] call after turn completes returns {cancelled: false, reason: "already_complete"} -- idempotent per INV-012
ERRORS:
  Session not found -> ValueError (HTTP 404)
  Turn not found, or wrong session -> ValueError (HTTP 404)
  Other-user turn (auth on, ownership mismatch) -> PermissionError (HTTP 404 per INV-009)
STATE: turn=in_flight -> turn=cancelling -> turn=complete(error="cancelled")
STEPS:
  1. [setup] Look up session in _sessions
     IF not found:
       RAISE ValueError("Unknown session")
  2. [setup] Validate session ownership (when auth enabled)
     IF user_id provided AND session.user_id != user_id:
       RAISE PermissionError (translates to 404)
  3. [setup] Look up turn record from _store
     IF not found OR turn.session_id != session_id:
       RAISE ValueError("Unknown turn")
  4. [branch] IF turn already complete (turn.completed_at is not None):
     - Write audit entry: was_in_flight=False, reason="already_complete"
     - RETURN {turn_id, cancelled: false, reason: "already_complete", partial_message_id: null}
  5. [branch] IF turn_id not in cancel_registry (turn finished + cleaned up between checks):
     - Treat same as already_complete branch
  6. [branch] IF cancel_registry[turn_id] already set (concurrent cancel in flight):
     - Write audit entry: was_in_flight=True, reason="already_cancelling"
     - RETURN {turn_id, cancelled: false, reason: "already_cancelling", partial_message_id: null}
  7. [sequential] Set the cancel event: cancel_registry[turn_id].set()
  8. [sequential] Write Heimdall audit entry: was_in_flight=True, reason="user_cancel", with current tool_calls_completed and partial_chars_emitted
  9. [sequential] partial_message_id <- if persist_partial requested AND a partial message will be persisted: derive id; else null
  10. [cleanup] RETURN {turn_id, cancelled: true, reason: null, partial_message_id}
TESTS:
  inflight_cancel [happy]: in-flight turn cancelled -> cancelled=true, registry event set
  already_complete [boundary]: turn finished -> cancelled=false, reason="already_complete", audit written
  already_cancelling [boundary]: second cancel -> cancelled=false, reason="already_cancelling", audit written
  unknown_session [error]: bad session_id -> ValueError -> HTTP 404
  unknown_turn [error]: bad turn_id -> ValueError -> HTTP 404
  other_user [security]: turn owned by different user_id -> PermissionError -> HTTP 404 (NOT 403, per INV-009)
  audit_metadata [security]: audit entry includes turn_id, was_in_flight, tool_calls_completed, partial_chars_emitted; never partial content
  matrix_bridge [edge]: bridge calls service.cancel_turn directly (no HTTP) -> same code path, same outcomes
```

```contract
FN cancel_turn_endpoint(
    session_id: str,
    turn_id: int,
    persist_partial: bool = False,
    user_id: str | None = Depends(get_current_user),
) -> dict
BRIEF: POST /sessions/{session_id}/turns/{turn_id}/cancel — HTTP cancel endpoint
PRE: [PRE-001 hard] session exists AND user owns it -- 404 otherwise
PRE: [PRE-002 hard] turn_id is an integer parsed from the path -- FastAPI 422 on parse error
POST: [POST-001 return_value] HTTP 200 with cancel result dict on success and idempotent no-ops -- returned as JSON
POST: [POST-002 side_effect] returns within 100ms p99 -- cancel is administrative; actual stream-shutdown is asynchronous (constraint enforced by the latency test below, not at type level)
ERRORS:
  Session/turn not found, or other-user -> HTTP 404 (translates from ValueError/PermissionError)
STEPS:
  1. [setup] Auth scope check: user has conversation_api:message:send -- per the locked decision to reuse this scope
  2. [sequential] Call _service.cancel_turn(session_id, turn_id, user_id=user_id, persist_partial=persist_partial)
     ON ValueError or PermissionError:
       RAISE HTTPException(404)
  3. [sequential] RETURN result dict (HTTP 200 by default)
TESTS:
  cancel_inflight [happy]: in-flight turn -> 200 {cancelled: true}
  cancel_idempotent [happy]: second cancel -> 200 {cancelled: false, reason: "already_cancelling"}
  cancel_after_complete [happy]: completed turn -> 200 {cancelled: false, reason: "already_complete"}
  bad_session [error]: unknown session_id -> 404
  bad_turn [error]: unknown turn_id -> 404
  other_user [security]: other-user turn -> 404 not 403 per INV-009
  no_auth [error]: missing/invalid bearer token when auth on -> 401
  latency [edge]: p99 < 100ms across happy + idempotent cases (synthetic timing harness)
```

```contract
FN ConversationService.check_rate_limits(
    user_id: str | None,
    provider_name: str,
    session_id: str | None = None,
) -> RateLimitOutcome
BRIEF: Pre-flight rate checks before a streaming turn; short-circuits on first denial
PRE: [PRE-001 soft] _rate_limiting_enabled is True -- returns RateLimitOutcome(denied=False) if not
POST: [POST-001 return_value] denied=False + semaphore_release_fn if all checks pass -- caller MUST pass fn to stream_turn
POST: [POST-002 return_value] denied=True + scope + retry_after_s on first denial -- semaphore_release_fn is None
POST: [POST-003 side_effect] on denial, any acquired semaphore slot is released before returning -- no slot leak
POST: [POST-004 side_effect] denial fires _heimdall_audit_store.record(...) best-effort -- try/except logger.exception
POST: [POST-005 invariant] fails open on SqliteRateStateStore exception -- log + allow request through
ERRORS:
  RateStateStore raises -> log via logger.exception, allow request through (fail-open)
  asyncio.TimeoutError on semaphore -> denied=True, scope="per_provider_concurrency", retry_after_s=0
STATE: stateless check; semaphore state changes only on deny (released) or pass (slot held until release_fn called)
STEPS:
  1. [branch] IF not _rate_limiting_enabled OR _rate_limiter is None:
     - RETURN RateLimitOutcome(denied=False)
  2. [setup] Resolve effective_uid: user_id or "anonymous"; select req_cfg/tok_cfg based on is_anonymous
  3. [sequential] Lazy-create provider semaphore: self._provider_semaphores.setdefault(provider_name, Semaphore(limit))
     -- single setdefault call ensures concurrent first-requests share the same semaphore
  4. [branch] TRY asyncio.wait_for(sem.acquire(), timeout=0)
     ON TimeoutError:
       - _audit_denial(scope="per_provider_concurrency", ...)
       - RETURN RateLimitOutcome(denied=True, scope="per_provider_concurrency", retry_after_s=0)
  5. [branch] IF req_cfg is not None:
     TRY await _rate_limiter.consume("conv:req:{uid}", req_cfg, cost=1)
       IF not allowed:
         - sem.release()
         - _audit_denial(scope="per_user_req_per_min", ...)
         - RETURN denied outcome
     ON Exception:
       - logger.exception("...failing open")
  6. [branch] IF tok_cfg is not None:
     TRY await _rate_limiter.peek("conv:tok:{uid}", tok_cfg)
       IF not allowed:
         - sem.release()
         - _audit_denial(scope="per_user_tokens_per_hour", ...)
         - RETURN denied outcome
     ON Exception:
       - logger.exception("...failing open")
  7. [cleanup] RETURN RateLimitOutcome(denied=False, semaphore_release_fn=lambda: sem.release(), ...)
TESTS:
  req_burst_denied [happy]: burst exhausted → 429 scope=per_user_req_per_min
  token_debt_denied [edge]: post-charge debt → next-turn peek denied; scope=per_user_tokens_per_hour
  concurrency_denied [edge]: semaphore at 0 → 429 scope=per_provider_concurrency, retry_after_s=0
  fail_open [failure]: store raises → request allowed, logger.exception in log
  anonymous_shared_bucket [edge]: two anon requests share conv:req:anonymous
  first_call_full_capacity [happy]: new user → GCRA bucket starts full, allowed
```

```contract
FN healthz() -> dict
BRIEF: GET /healthz — liveness probe; unconditional 200 if the request reaches the handler
PRE: [PRE-001 soft] none — endpoint is unauthenticated and stateless
POST: [POST-001 return_value] HTTP 200 with body {"status": "ok"} -- always
STEPS:
  1. [sequential] RETURN {"status": "ok"}
TESTS:
  ok [happy]: GET /healthz → 200 with {"status": "ok"}, regardless of service state
```

```contract
FN readyz() -> dict
BRIEF: GET /readyz — readiness probe; 200 when the service is serving traffic, 503 otherwise
PRE: [PRE-001 soft] none — endpoint is unauthenticated; reads _service in-process state
POST: [POST-001 return_value] HTTP 200 with {status, version, git_sha} when ready -- per INV-015
POST: [POST-002 return_value] HTTP 503 with {status: "not_ready", reason} when not ready -- detail dict
ERRORS:
  Not ready -> HTTPException(503, detail={"status": "not_ready", "reason": ...})
STEPS:
  1. [sequential] reason = _readiness_reason(_service)
  2. [branch] IF reason is None:
     - RETURN {"status": "ok", "version": _VERSION, "git_sha": _GIT_SHA}
  3. [error_handler] ELSE:
     - RAISE HTTPException(503, detail={"status": "not_ready", "reason": reason})
TESTS:
  ready [happy]: _started=True + agents populated → 200 with status/version/git_sha
  not_started [error]: _started=False → 503 with reason="startup_incomplete"
  env_vars [edge]: WORLDTREE_VERSION/WORLDTREE_GIT_SHA present → values surface in body; absent → "unknown"
```

```contract
FN ConversationService.attach_turn(
    session_id: str,
    turn_id: int,
    last_event_id: str,
    *,
    user_id: str | None = None,
) -> AsyncIterator[dict[str, Any]]
BRIEF: Attach to an in-flight turn; replay buffered events then drain live queue
PRE: [PRE-001 hard] last_event_id is composite "{int}:{int}" with parsed turn_id == turn_id param -- raise ValueError("invalid_last_event_id")
PRE: [PRE-002 hard] _turn_session[turn_id] == session_id if turn_id is bound -- raise PermissionError (INV-018)
PRE: [PRE-003 hard] session exists and user_id matches session owner -- raise PermissionError (INV-009)
PRE: [PRE-004 hard] turn_id present in _replay_buffer (turn still in flight) -- raise LookupError("turn_finished")
PRE: [PRE-005 hard] buffer's earliest seq <= parsed_seq + 1 (no eviction gap) -- raise ValueError("buffer_expired:{earliest_seq}")
POST: [POST-001 return_value] yields all events with seq > parsed_seq from buffer then queue -- monotonic, no duplicates
POST: [POST-002 side_effect] does NOT pop _replay_buffer, _event_queues, or _turn_session -- cleanup is stream_turn's responsibility
ERRORS:
  Missing colon / non-integer / negative seq / turn_id mismatch -> ValueError("invalid_last_event_id") -> HTTP 400
  _turn_session binding mismatch -> PermissionError -> HTTP 404 per INV-009
  Session not found / ownership mismatch -> PermissionError -> HTTP 404
  turn_id not in _replay_buffer -> LookupError("turn_finished") -> HTTP 410
  earliest_seq > parsed_seq + 1 -> ValueError("buffer_expired:X") -> HTTP 412
STEPS:
  1. [setup] Parse last_event_id; reject on invalid format or turn_id mismatch
  2. [branch] Check _turn_session binding (INV-018); raise PermissionError on mismatch
  3. [setup] Validate session existence and ownership; raise PermissionError on mismatch
  4. [branch] Look up _replay_buffer; raise LookupError("turn_finished") if absent
  5. [branch] Check buffer_expired (earliest_seq > parsed_seq + 1); raise ValueError if gap exists
  6. [loop] Replay buffered events with seq > parsed_seq; check cancel between each yield
  7. [loop] Drain queue: skip events with seq <= parsed_seq; yield live events; stop on _END or cancel
TESTS:
  replay_happy [happy]: attach mid-turn; gets events with seq > Last-Event-ID's seq
  malformed [error]: non-composite header → ValueError(invalid_last_event_id)
  turn_id_mismatch [error]: Last-Event-ID turn_id != param → ValueError(invalid_last_event_id)
  finished [error]: turn cleaned up → LookupError(turn_finished)
  buffer_expired [edge]: seq before buffer → ValueError(buffer_expired:X)
  cross_session [security]: turn from S1 via S2 path → PermissionError
  cancel_between_events [edge]: cancel detected between replay events, not before first
```

```contract
FN send_message(session_id: str, body: SendMessageRequest, request: Request, ctx: SecurityContext) -> EventSourceResponse
BRIEF: POST /sessions/{session_id}/messages — start new turn (no header) or resume via attach_turn (Last-Event-ID header)
PRE: [PRE-001 hard] session exists -- HTTPException(404)
PRE: [PRE-002 hard] ctx owns the session -- HTTPException(404) per INV-009
POST: [POST-001 return_value] SSE stream terminating with done/cancelled/error -- per INV-004
POST: [POST-002 return_value] resume path: HTTP 400/410/412 on invalid/expired/finished Last-Event-ID
ERRORS:
  Session not found -> HTTPException(404)
  Session ownership mismatch -> HTTPException(404) per INV-009
  Last-Event-ID malformed or turn_id mismatch -> JSONResponse(400, {"error": "invalid_last_event_id"})
  Turn finished + cleaned up -> JSONResponse(410, {"error": "turn_finished", "turn_id": N})
  Buffer gap -> JSONResponse(412, {"error": "buffer_expired", "buffered_from_seq": X, "turn_id": N})
  Rate limited (new-turn path only) -> JSONResponse(429) per INV-022
STEPS:
  1. [setup] Get and validate session; check session access
  2. [branch] IF Last-Event-ID header present:
     - Parse turn_id from composite header
     - Call attach_turn(session_id, turn_id, last_event_id, user_id=...)
     - Map ValueError/LookupError/PermissionError to 400/410/412/404
     - Return EventSourceResponse wrapping attach_turn iterator
  3. [branch] ELSE (no header — new turn):
     - Rate-limit check (INV-022); return JSONResponse(429) if denied
     - Call stream_turn; return EventSourceResponse
TESTS:
  no_header [happy]: header absent → stream_turn path, new turn
  with_header [happy]: valid Last-Event-ID → attach_turn path, events replayed
  malformed_header [error]: bad format → 400
  finished_turn_header [error]: turn cleaned up → 410
  buffer_expired_header [error]: seq too old → 412
  cross_user_header [security]: Bob attaches Alice's session → 404
```

```contract
FN rotate_key(key_id: str, body: RotateKeyRequest, ctx: SecurityContext) -> dict
BRIEF: Atomically rotate an API key — insert a new key and mark the old one superseded
       within a single transaction, with a configurable grace window.
PRE: [PRE-001 hard] ctx has admin.keys.write scope -- 403 otherwise
PRE: [PRE-002 hard] key_id exists in the user store -- 404 otherwise
PRE: [PRE-003 hard] old key disabled_at IS NULL -- 410 "Key is revoked; cannot rotate"
PRE: [PRE-004 hard] old key superseded_at IS NULL -- 409 "Key is already superseded"
PRE: [PRE-005 validate] grace_seconds in [0, 86400] inclusive if provided -- 422 otherwise
POST: [POST-001 side_effect] new key row inserted in api_keys with fresh key_hash, key_id, key_suffix
POST: [POST-002 side_effect] old key row has superseded_at, superseded_by_key_id, grace_seconds set
POST: [POST-003 state_change] both writes atomically visible or both rolled back (INV-020)
POST: [POST-004 side_effect] audit entry written: action=conversation_api:admin:key:rotate, outcome=success
POST: [POST-005 return] response body == POST /admin/keys body + {supersedes, supersedes_until}
ERRORS:
  PermissionError -> 403
  key_id not found -> audit outcome=not_found, 404
  disabled_at IS NOT NULL -> audit outcome=gone, 410
  superseded_at IS NOT NULL (pre-check) -> audit outcome=already_superseded, 409
  KeyAlreadySupersededError (0 rowcount in supersede_api_key) -> audit outcome=already_superseded, 409 (identical body to pre-check path)
  grace_seconds < 0 or > 86400 -> 422 (Pydantic validation, no audit)
STEPS:
  1. [setup] Authorize ctx for admin.keys.write scope
  2. [branch] Fetch record by key_id; if None → audit not_found, raise 404
  3. [branch] If record.disabled_at IS NOT NULL → audit gone, raise 410
  4. [branch] If record.superseded_at IS NOT NULL → audit already_superseded, raise 409
  5. [setup] Resolve grace: body.grace_seconds if provided, else config.key_rotation_grace_seconds (default 300)
  6. [setup] Generate cleartext = "wt_live_" + token_hex(16); hash it; take suffix
  7. [setup] Generate new_kid with collision retry (3 attempts)
  8. [sequential] Call user_store.supersede_api_key(old_key_hash, ..., grace_seconds=grace)
  9. [error_handler] ON KeyAlreadySupersededError: audit already_superseded, raise 409
  10. [sequential] Audit success with {old_key_id, new_key_id, grace_seconds}
  11. [sequential] Compute supersedes_until = old_record.superseded_at + timedelta(seconds=grace)
  12. [cleanup] Return _record_to_dict(new_record, cleartext=cleartext) + {supersedes, supersedes_until}
TESTS:
  rotate_happy [happy]: valid key_id → 200, key=wt_live_*, supersedes=old_kid, supersedes_until present
  rotate_old_key_during_grace [happy]: auth with old key during grace → 200
  rotate_old_key_after_grace [edge]: grace=0, auth with old key → 401, disabled_at populated
  rotate_already_superseded [error]: second rotate → 409, detail="Key is already superseded"
  rotate_revoked [error]: rotate revoked key → 410
  rotate_unknown [error]: rotate unknown key_id → 404
  rotate_no_admin_scope [security]: non-admin → 403
  rotate_grace_bounds [error]: -1 → 422; 86401 → 422; 0 → 200; 86400 → 200
  rotate_concurrent [edge]: two concurrent rotates → one 200, one 409; 409 body identical
  rotate_list_fields [happy]: after rotate, list shows superseded_at + effectively_disabled=true
  rotate_grace_captured [edge]: stored grace_seconds governs lazy-disable, not live config
```

```contract
FN me(ctx: SecurityContext) -> dict
BRIEF: Return the authenticated principal's identity and resolved key metadata.
       Cheap read-only introspection; no side effects; fails open on storage error.
PRE: [PRE-001 soft] ctx has valid auth or is anonymous -- 401 handled by get_security_context before this handler
POST: [POST-001 return] response always contains {user_id, scopes (sorted list), tier}
POST: [POST-002 return] anonymous short-circuit: ctx.user_id == "anonymous" → {user_id, scopes, tier="anonymous"}, no DB call
POST: [POST-003 return] authenticated: tier/display_name/user_created_at from get_user; key fields from get_api_keys_for_user
POST: [POST-004 return] key fields (key_id, key_label, key_created_at) present only when active key resolved (INV-026)
POST: [POST-005 return] superseded_at/superseded_by_key_id/grace_seconds present only when resolved key is in rotation grace
POST: [POST-006 return] all optional fields omitted (not null) via model_dump(exclude_none=True) -- INV-026
POST: [POST-007 return] tier="unknown" on Heimdall lookup failure (fail-open, non-5xx) -- INV-026
POST: [POST-008 side_effect] NO audit entry written
POST: [POST-009 side_effect] NO rate-limit check or counter update (INV-022)
ERRORS:
  Heimdall lookup raises → logger.exception; return degraded {user_id, scopes, tier="unknown"} with 200
STEPS:
  1. [branch] IF ctx.user_id == _ANONYMOUS_USER_ID: return {user_id, scopes=sorted(ctx.scopes), tier="anonymous"}
  2. [error_handler] try: get_user(ctx.user_id) + get_api_keys_for_user(ctx.user_id)
     except Exception: logger.exception; return {user_id, scopes=sorted(ctx.scopes), tier="unknown"}
  3. [sequential] Build base response: {user_id, scopes=sorted(ctx.scopes), tier, display_name?, user_created_at?}
  4. [sequential] Filter keys to active (disabled_at IS NULL AND (superseded_at IS NULL OR superseded_at+grace>now))
  5. [sequential] Sort active keys by last_used_at DESC NULLS LAST; take first as best_key
  6. [branch] IF best_key: add key_id, key_label, key_created_at; IF superseded_at: add rotation fields
  7. [cleanup] Return MeResponse(**data).model_dump(exclude_none=True)
TESTS:
  authenticated_full_shape [happy]: valid key → 200, user_id+scopes+tier+display_name+user_created_at+key_id+key_label+key_created_at present
  anonymous_minimal_shape [happy]: dev mode → 200, user_id="anonymous", tier="anonymous", key_* absent
  invalid_key_401 [error]: bad token → 401
  scopes_sorted_list [edge]: scopes is list, alphabetically sorted
  most_recently_used_key [edge]: B has most recent last_used_at → key_id=B even when C authenticated (best-effort)
  rotation_grace_fields [edge]: old key in grace → key_id=old, superseded_at+superseded_by_key_id+grace_seconds present
  no_active_keys_omits_key_fields [edge]: get_api_keys_for_user returns [] → key_* absent, user fields present
  lookup_failure_fallback [failure]: get_user raises → 200, tier="unknown", exception logged
  no_rate_limit_budget [edge]: 10 /me calls + 1 message → all 200 (budget unconsumed)
  no_audit_emission [security]: write_detail NOT called for /me
```

```contract
FN http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse
BRIEF: Custom HTTPException handler; ensures every HTTPException response carries an error_code (INV-031/INV-032)
POST: [POST-001 return_value] if exc.detail is already a coded dict (has "error_code" key), return it unchanged at original status_code
POST: [POST-002 return_value] if exc.detail is uncoded and status_code is in mapping table, return mapped code
POST: [POST-003 return_value] if exc.detail is uncoded and status_code has no mapping, return internal_error + logger.warning
ERRORS:
  No errors — handler must not raise
STEPS:
  1. [branch] IF isinstance(exc.detail, dict) AND "error_code" in exc.detail:
     - RETURN JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
  2. [branch] Inspect exc.status_code: 405 → METHOD_NOT_ALLOWED, 400 → MALFORMED_REQUEST
     IF mapped:
       - RETURN JSONResponse(status_code=exc.status_code, content={"detail": _err(code, str(exc.detail))})
  3. [error_handler] No mapping:
     - LOG logger.warning("HTTPException raised without error_code: status=%d detail=%r", exc.status_code, exc.detail)
     - RETURN JSONResponse(status_code=exc.status_code, content={"detail": _err(ErrorCode.INTERNAL_ERROR, str(exc.detail))})
TESTS:
  coded_passthrough [happy]: detail already has error_code → returned unchanged with original status
  mapped_405 [happy]: 405 plain string → method_not_allowed code (INV-032)
  mapped_400 [happy]: 400 plain string → malformed_request code (INV-032)
  unmapped_warning [failure]: status 418 string detail → internal_error + warning logged
```

```contract
FN validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse
BRIEF: Custom RequestValidationError handler; wraps Pydantic's per-field errors with error_code (INV-031)
POST: [POST-001 return_value] HTTP 422 with {error_code: "validation_failed", message: "Request validation failed", errors: [...]}
POST: [POST-002 return_value] errors array is Pydantic's per-field error list verbatim
STEPS:
  1. [sequential] RETURN JSONResponse(status_code=422, content={"detail": _err(ErrorCode.VALIDATION_FAILED, "Request validation failed", errors=exc.errors())})
TESTS:
  missing_field [happy]: body missing required field → 422 validation_failed + errors array present
  errors_array_preserved [happy]: per-field detail in errors → forwarded verbatim to client
```

```contract
FN generic_exception_handler(request: Request, exc: Exception) -> JSONResponse
BRIEF: Last-resort handler for unhandled exceptions; logs traceback, never leaks internals (INV-031)
POST: [POST-001 return_value] HTTP 500 with {error_code: "internal_error", message: "An internal error occurred"}
POST: [POST-002 side_effect] logger.exception(...) called to capture full traceback server-side
POST: [POST-003 invariant] exception type name, repr, or traceback NEVER leaked in client response body
STEPS:
  1. [sequential] LOG via logger.exception("Unhandled exception", exc_info=exc)
  2. [sequential] RETURN JSONResponse(status_code=500, content={"detail": _err(ErrorCode.INTERNAL_ERROR, "An internal error occurred")})
TESTS:
  uncaught_exception [security]: Exception raised in handler → 500, error_code=internal_error, type name absent from body, logger.exception called
```

## Amendment — turn lifecycle infrastructure (INV-033..INV-038)

This amendment adds three primitives to `core/conversation_api/`:

1. A terminal `phase` column on the `turns` table (one of `succeeded` / `failed` / `cancelled` / `stalled`) with a one-shot backfill for pre-existing rows.
2. A per-turn stall watchdog (`asyncio.TimerHandle` via `loop.call_later`) that auto-terminates a turn when no engine event has been seen for `stall_timeout_s` (default 300, matching the bus converse timeout), reusing the cancel registry mechanism so the streaming generator has one loop-exit branch.
3. A log-correlation prefix `[{session_prefix}/{turn_id}/{seq}]` on stream-related log lines for cross-system grep.

SSE phase events (the `BuildingPrompt → CallingLLM → ProcessingTools → Streaming → Finishing` progress vocabulary) are explicitly **out of scope** for this amendment. They will land in a follow-up amendment after the column and watchdog have been live long enough to clarify the SSE schema design ambiguities.

### Storage schema delta

`core/conversation_api/store.py`, `turns` table:

```sql
ALTER TABLE turns ADD COLUMN phase TEXT;
-- NULL until terminal write; one of {'succeeded','failed','cancelled','stalled'} thereafter
```

Backfill on first migration (single transaction, applied in INV-034 precedence order).

### Cancel-registry shape delta

`_CancelRequest` in `service.py` gains two fields:

```python
@dataclass
class _CancelRequest:
    event: asyncio.Event
    persist_partial: bool = False
    requested_at: float | None = None
    reason: str | None = None              # NEW: 'user_cancel' | 'stall' | None
    stall_timer: asyncio.TimerHandle | None = None  # NEW
```

### Configuration

`stall_timeout_s` resolution order (first match wins, value in seconds):

1. `agents/{agent_id}/config.yaml` → `conversation.stall_timeout_s`
2. `config/defaults.yaml` → `conversation_api.stall_timeout_s`
3. Built-in fallback: `300.0`

```contract
FN ConversationStore._migrate_phase_column() -> None
BRIEF: One-shot migration; adds `phase` column and backfills pre-existing rows
PRE: [PRE-001 soft] turns table exists -- skip silently otherwise
POST: [POST-001 state_change] `phase` column present after first run -- subsequent runs are no-op
POST: [POST-002 state_change] every pre-existing row has `phase` derived per INV-034
POST: [POST-003 invariant] values written by a prior run are NOT re-derived on subsequent runs (idempotent)
ERRORS:
  ALTER TABLE fails because column already present -> log debug and continue (safety net for racing migrators)
STEPS:
  1. [setup] PRAGMA table_info(turns) -- detect 'phase' presence
  2. [branch] IF 'phase' present: RETURN (idempotent no-op)
  3. [setup] BEGIN immediate transaction
  4. [sequential] ALTER TABLE turns ADD COLUMN phase TEXT
  5. [sequential] UPDATE turns SET phase = 'cancelled' WHERE cancelled = 1
  6. [sequential] UPDATE turns SET phase = 'failed'    WHERE phase IS NULL AND error IS NOT NULL
  7. [sequential] UPDATE turns SET phase = 'succeeded' WHERE phase IS NULL AND completed_at IS NOT NULL
  8. [sequential] UPDATE turns SET phase = 'stalled'   WHERE phase IS NULL
  9. [cleanup] COMMIT
TESTS:
  fresh_db [happy]: empty turns table → column added, no UPDATE rows touched
  cancelled_row [happy]: cancelled=1 → phase='cancelled' after migrate
  errored_row [happy]: error='boom', cancelled=0 → phase='failed'
  succeeded_row [happy]: completed_at set, no error, cancelled=0 → phase='succeeded'
  orphaned_row [edge]: started_at set, completed_at NULL, error NULL, cancelled=0 → phase='stalled'
  rerun_idempotent [edge]: migrate twice → second is no-op
  rerun_preserves_existing_values [edge]: first run writes phase='succeeded'; manually NULL out completed_at then re-run → phase still 'succeeded'
  precedence_cancelled_over_error [edge]: cancelled=1 AND error='something' → phase='cancelled' (cancelled wins)
```

```contract
FN ConversationStore.complete_turn(...amended)
BRIEF: Persist turn completion with terminal phase (amends the prior signature)
SIGNATURE_DELTA:
  + phase: str | None = None       # one of 'succeeded'|'failed'|'cancelled'|'stalled' or None
PRE: [PRE-001 hard] phase ∈ {'succeeded','failed','cancelled','stalled', None} -- raise ValueError otherwise
POST: [POST-001 state_change] turns.phase column = phase argument (NULL when None)
POST: [POST-002 invariant] when phase is non-None, the (cancelled, error, completed_at) tuple agrees with phase per INV-034 -- contract obligation on the caller, not enforced at the boundary
NOTES:
  - All other args (duration_ms, tool_call_count, output_chars, error, cancelled, had_estimate, ...) unchanged.
  - Passing phase=None preserves backwards compatibility for any caller not yet updated; rollout-window only — every terminal path in stream_turn MUST pass phase explicitly per INV-033.
  - The store is the only writer of the phase column. Direct SQL UPDATEs from elsewhere are forbidden.
TESTS:
  succeeded [happy]: complete_turn(..., phase='succeeded') → row has phase='succeeded'
  failed [happy]: complete_turn(..., error='boom', phase='failed') → row has phase='failed'
  cancelled [happy]: complete_turn(..., cancelled=True, error='cancelled', phase='cancelled') → phase='cancelled'
  stalled [happy]: complete_turn(..., error='stalled', phase='stalled') → phase='stalled'
  invalid_phase [error]: complete_turn(..., phase='weird') → ValueError
  none_phase_legacy [edge]: complete_turn(..., phase=None) → row has phase NULL (rollout-window legacy path)
```

```contract
FN ConversationService._start_stall_timer(turn_id: int, timeout_s: float) -> None
       ConversationService._reset_stall_timer(turn_id: int, timeout_s: float) -> None
       ConversationService._clear_stall_timer(turn_id: int) -> None
BRIEF: Per-turn watchdog timer lifecycle. Created at turn start, reset on each engine event, cleared in stream_turn's finally block.
PRE: [PRE-001 soft] turn_id may or may not be in _cancel_registry -- functions are best-effort no-ops on missing entries (defensive)
POST: [POST-001 state_change] after _start: _cancel_registry[turn_id].stall_timer holds a live TimerHandle
POST: [POST-002 state_change] after _reset: prior handle cancelled, new handle stored
POST: [POST-003 state_change] after _clear: handle cancelled, slot is None
POST: [POST-004 invariant] at most ONE live TimerHandle exists per turn_id at any moment -- INV-037
ERRORS:
  None. Missing registry entry → silent no-op. Cancelling an already-fired handle is a stdlib no-op.
STEPS:
  _start_stall_timer:
    1. [setup] entry = _cancel_registry.get(turn_id); IF None: RETURN
    2. [sequential] handle = loop.call_later(timeout_s, _on_stall, turn_id)
    3. [sequential] entry.stall_timer = handle
  _reset_stall_timer:
    1. [setup] entry = _cancel_registry.get(turn_id); IF None: RETURN
    2. [branch] IF entry.stall_timer is not None: entry.stall_timer.cancel()
    3. [sequential] handle = loop.call_later(timeout_s, _on_stall, turn_id)
    4. [sequential] entry.stall_timer = handle
  _clear_stall_timer:
    1. [setup] entry = _cancel_registry.get(turn_id); IF None: RETURN
    2. [branch] IF entry.stall_timer is not None:
       - entry.stall_timer.cancel()
       - entry.stall_timer = None
TESTS:
  start_then_clear [happy]: start, then clear → handle cancelled, slot is None
  reset_replaces [happy]: start, reset → first handle cancelled, new handle stored, only one live timer
  clear_idempotent [edge]: clear twice → no error
  no_registry_entry [edge]: clear before start → no error (defensive no-op)
  fired_handle_clear [edge]: timer fires first, then clear is called → no error (cancelling fired handle is stdlib-safe)
```

```contract
FN ConversationService._on_stall(turn_id: int) -> None
BRIEF: Watchdog callback. Signals stall via the cancel registry; same loop-exit as user cancel; different reason and terminal phase.
PRE: [PRE-001 soft] turn_id may have been cleaned up between scheduling and firing -- no-op if not in registry
POST: [POST-001 state_change] _cancel_registry[turn_id].reason = 'stall' (only when not already set)
POST: [POST-002 state_change] _cancel_registry[turn_id].event.set() called
POST: [POST-003 side_effect] stream_turn's loop exits at the next iteration via the existing cancel-detection branch (per INV-036)
POST: [POST-004 invariant] when user cancel won the race (reason already 'user_cancel'), this callback is a no-op -- reason stays 'user_cancel', phase will be 'cancelled' not 'stalled'
ERRORS:
  None. Watchdog firing on a missing turn is a silent no-op.
STEPS:
  1. [setup] entry = _cancel_registry.get(turn_id); IF None: RETURN
  2. [branch] IF entry.event.is_set():
     - RETURN -- someone else won the race; their reason wins
  3. [sequential] entry.reason = 'stall'
  4. [sequential] entry.requested_at = time.monotonic()
  5. [sequential] entry.event.set()
TESTS:
  fires_on_active_turn [happy]: live registry entry, event not set → reason='stall', event set
  noop_on_missing_turn [edge]: turn already cleaned up → no error, no exception
  loses_race_to_user_cancel [edge]: cancel_turn ran first (reason='user_cancel', event set) → _on_stall is no-op; reason stays 'user_cancel'
```

### `stream_turn` — STEPS amendment

The existing `stream_turn` contract block (above) is amended as follows. New numbering uses `5a/5b/5c` etc. to insert without renumbering existing steps.

After step 5 (existing — register turn in cancel registry):

  5a. [setup] Resolve `stall_timeout_s` per the configuration order above.
  5b. [setup] `_start_stall_timer(turn_id, stall_timeout_s)`
  5c. [setup] Construct a `TurnLogContext(session_id=session_id, turn_id=turn_id)` (or equivalent helper) that emits log lines via `logger` with the prefix `[{session_prefix}/{turn_id}/{seq}]` per INV-038. All `logger.*(...)` calls within this generator from this point use the context's `.log(...)` interface. The context's internal `seq` field is updated immediately after each `yield` so the next log line carries the just-emitted seq.

Modify step 10 (existing — FOR EACH event from engine):

  - At the START of every iteration (before the event-type dispatch), call `_reset_stall_timer(turn_id, stall_timeout_s)` per INV-035.
  - At the cancel-detection branch (existing — `IF _cancel_registry[turn_id].event.is_set()`), branch on `entry.reason`:
    - **IF entry.reason == 'stall'**:
      - YIELD `{"type": "cancelled", "turn_id": N, "reason": "stall", "partial_message_id": null}` (the existing event shape; the `reason` field is set to `"stall"`)
      - Call `_store.complete_turn(turn_id, error="stalled", duration_ms=elapsed, cancelled=True, phase="stalled")`
      - Drop partial assistant output UNCONDITIONALLY — `persist_partial` is NOT honoured on stall per INV-036
      - Write Heimdall audit entry with `reason='stall'` and `was_in_flight=True`
      - BREAK out of the event loop
    - **ELSE** (entry.reason == 'user_cancel' or None):
      - Existing user-cancel branch unchanged (yield `cancelled` with `reason='user_cancel'`, honour `persist_partial`, audit, etc.)
      - Call `_store.complete_turn(turn_id, ..., phase='cancelled')`

Modify step 12 (existing — TurnCompleteEvent / success path):

  - Replace `_store.complete_turn(turn_id, ...)` with `_store.complete_turn(turn_id, ..., phase='succeeded')`.

Modify step 13 (existing — other-exception error_handler):

  - Replace `_store.complete_turn(turn_id, error=str(exc))` with `_store.complete_turn(turn_id, error=str(exc), phase='failed')`.

Insert step 14a in the cleanup `finally` block (existing step 14: `cancel_registry.pop(turn_id, None)`):

  14a. [cleanup] `_clear_stall_timer(turn_id)` — MUST run on every exit path per INV-037. Order: `_clear_stall_timer` BEFORE `cancel_registry.pop(...)`, since `_clear_stall_timer` reads from the registry.

### `cancel_turn` — STEPS amendment

The existing `cancel_turn` contract block is amended as follows.

Modify step 7 (existing — Set the cancel event):

  - Replace `cancel_registry[turn_id].set()` with the equivalent block that also stamps `reason` and `requested_at`:
    ```
    entry = cancel_registry[turn_id]
    entry.reason = 'user_cancel'
    entry.requested_at = time.monotonic()
    entry.event.set()
    ```

All other existing steps and POSTs unchanged.

### Acceptance tests for the amendment

These tests are in addition to the per-block `TESTS` listed above:

- `stall_after_silent_300s [edge]`: in-flight turn produces no engine events for 300 s → cancelled SSE event yielded with `reason="stall"`; `turns.phase = 'stalled'`; `duration_ms ≈ 300000`; client-visible loop-exit identical to user cancel except for the `reason` and `phase` strings.
- `thinking_resets_stall [edge]`: zai/glm thinking mode emits `ThinkingDeltaEvent` at t=200s → no stall fires before t=500s (200+300). Verifies INV-035.
- `tool_running_resets_stall [edge]`: `ToolCallStartEvent` at t=200s, `ToolCallResultEvent` at t=400s → no stall before t=700s. Verifies tool-call boundaries also reset.
- `user_cancel_wins_race [edge]`: user calls `cancel_turn` at t=299.9s; stall timer scheduled to fire at t=300s → terminal phase is `'cancelled'`, not `'stalled'`; second `event.set()` from `_on_stall` is a no-op; SSE event has `reason='user_cancel'`.
- `stall_persist_partial_dropped [security]`: hypothetical caller registered `persist_partial=True` then stall fires (today this combination cannot happen via the API, but defended) → partial output is still dropped; no `partial=true` row inserted into `messages`.
- `timer_cleanup_on_success [edge]`: normal completion → `_clear_stall_timer` runs in `finally`; `_cancel_registry[turn_id].stall_timer` is None at the moment the registry entry is popped; no leaked handle.
- `timer_cleanup_on_provider_exception [edge]`: provider raises mid-stream → finally clears the handle; INV-037 holds.
- `timer_cleanup_on_user_cancel [edge]`: user cancel exits the loop → finally clears the handle (the handle that didn't fire); no leaked timer.
- `log_prefix_format [happy]`: log line from inside `stream_turn` → matches regex `\[[a-f0-9]{8}/\d+/\d+\] `.
- `log_prefix_session_no_dashes [edge]`: session UUID `550e8400-e29b-41d4-...` → log prefix uses `550e8400` (the leading 8 hex chars after dashes are stripped).
- `log_prefix_short_session [edge]`: a synthetic session_id of 4 chars → log prefix uses the full 4-char id (defensive; production UUIDs always exceed 8 hex).
- `log_prefix_seq_advances [happy]`: across N events emitted by a turn, log lines emitted between event K and event K+1 carry seq=K (the most recently yielded seq).
- `phase_succeeded_on_normal_complete [happy]`: turn completes via `TurnCompleteEvent` → `turns.phase = 'succeeded'`.
- `phase_failed_on_provider_exception [happy]`: provider raises → `turns.phase = 'failed'`.
- `phase_cancelled_on_user_cancel [happy]`: user calls `cancel_turn` → `turns.phase = 'cancelled'`.
- `phase_stalled_on_watchdog [happy]`: watchdog fires → `turns.phase = 'stalled'`.
- `stall_timeout_per_agent_override [edge]`: agent config sets `conversation.stall_timeout_s: 60`; turn with no events at t=60.5s → stall fires at ~t=60s.
- `stall_timeout_service_default [edge]`: agent config has no override; service config sets `conversation_api.stall_timeout_s: 120`; turn stalls at ~t=120s.
- `stall_timeout_builtin_fallback [edge]`: neither agent nor service config sets the value → 300.0 s default applies.

---

## Amendment — SSE phase events (issue #151, INV-053..INV-061)

This amendment adds in-flight worker-phase events to the SSE stream emitted by
`POST /sessions/{id}/messages`. Commit 1 of the Symphony lift (`62a5ab7`) landed
the `turns.phase` DB column, the stall watchdog, the log-correlation prefix, and
the `phase` field on terminal SSE events (`done`/`error`/`cancelled`). This amendment
covers the IN-FLIGHT half: events telling the client what the worker is doing right
now, emitted at phase-entry boundaries.

The five phases form a closed-set Literal vocabulary: `BuildingPrompt | CallingLLM |
ProcessingTools | Streaming | Finishing`. Phase events flow through the same
`_publish → _replay_buffer (deque maxlen=10) + queue` pipeline as content events,
consume seq numbers in the composite `{turn_id}:{seq}` SSE id, and replay on
`Last-Event-ID` reconnect identically to content events.

Note on numbering: the locked issue-scoped contract at
`docs/contracts/issues/151.contract.md` originally enumerated these invariants
as INV-039..INV-047. Those numbers collide with the existing pagination +
admin-events invariants on this module-scoped contract. The amendment uses
INV-053..INV-061 (continuing the existing sequence after INV-052) and keeps
the same one-to-one mapping in order.

### New invariants

- **INV-053 (phase-event-shape)** (issue-contract INV-039): The wire shape of an in-flight phase event is exactly `{"type": "worker_phase", "phase": "<name>", "turn_id": N}` — three fields, no extras. Adding fields is a breaking change requiring a contract amendment.

- **INV-054 (phase-vocabulary-closed)** (issue-contract INV-040): The `phase` field on a `worker_phase` event takes exactly one of five values: `BuildingPrompt`, `CallingLLM`, `ProcessingTools`, `Streaming`, `Finishing`. Implementation MUST guard via `assert phase_name in _WORKER_PHASE_VOCAB` (or equivalent); a value outside this set is a contract violation.

- **INV-055 (emit-on-entry)** (issue-contract INV-041): Phase events emit at phase ENTRY, not exit. Clients infer "phase X ended" from the arrival of the next phase event or the terminal event.

- **INV-056 (every-turn-min-sequence)** (issue-contract INV-042): Every successful turn emits at minimum `BuildingPrompt → CallingLLM → Streaming → Finishing → done(phase="succeeded")`. Zero phase events on a successful turn is a contract violation. `Streaming` MUST emit before `Finishing` even when no `ContentDeltaEvent` arrived during the turn (e.g. when the LLM returns text only in the terminal `TurnResult`).

- **INV-057 (ProcessingTools-conditional)** (issue-contract INV-043): A `ProcessingTools` phase event emits if and only if at least one tool call happens during the turn. Turns without tool calls never emit `ProcessingTools`. Multiple tool calls within a single LLM round-trip emit one `ProcessingTools` event for the whole batch (not one per tool). The wire ordering of a tool-using round-trip is `tool_start × N → ProcessingTools → tool_result × N`: all `tool_start` events emit BEFORE the single `ProcessingTools`; that emits BEFORE the first `tool_result`.

- **INV-058 (tool-roundtrip-cycle)** (issue-contract INV-044): Each LLM round-trip that includes tool calls produces the cycle `CallingLLM → ProcessingTools → CallingLLM`. Implementation MUST emit `CallingLLM` on re-entry after the last `tool_result` of a round-trip (not omit it, not collapse the cycle to a single span).

- **INV-059 (cancel-error-skip-Finishing)** (issue-contract INV-045): Cancel, error, and stall paths emit the terminal event (`cancelled` / `error`) directly after the most recent in-flight phase event. NO `Finishing` event emits on these paths.

- **INV-060 (replay-includes-phase)** (issue-contract INV-046): Phase events flow through the same `_publish → _replay_buffer + queue` path as content events. They consume seq numbers in the composite `{turn_id}:{seq}` SSE id and replay on `Last-Event-ID` reconnect identically to content events. No separate ring buffer; no skip-on-replay carve-out.

- **INV-061 (BuildingPrompt-first)** (issue-contract INV-047): `BuildingPrompt` is the FIRST event of every successful turn, emitted before any agent code runs (auth check, agent-context resolution, persona load, tool resolution, prompt-build, LLM call). In practice the emission happens before the `async for event in engine.run_turn_streaming(...)` loop is entered — the engine's generator-body where setup occurs does not execute until the first `__anext__()`. The next phase event (`CallingLLM`) MUST emit only after that first engine event is received (proof that the engine has completed setup and the LLM call is in flight), not back-to-back with `BuildingPrompt`. Clients can reliably treat the arrival of the first `worker_phase` event as the canonical "the server received my request and started work" signal.

### Integration notes

- `_WORKER_PHASE_VOCAB: frozenset[str]` is a module-level constant in `service.py`.
- `_publish_phase(phase_name: str) -> dict` is a nested function inside `stream_turn`, defined immediately after `_publish`. It asserts the vocab guard and delegates to `_publish`.
- Terminal events gain `phase` on the wire: `done` carries `"phase": "succeeded"`, `error` carries `"phase": "failed"`, `cancelled` carries `"phase": cancel_phase` (one of `"cancelled"` / `"stalled"`).
- No per-phase configuration knob. Clients that don't want phase events filter on `event["type"] != "worker_phase"` client-side.

---

## Amendment: Pending-Task Visibility (issue #119)

### New invariants

**INV-065 (pending-bus-only-v0):** The pending list surfaces only active Bus-v2 tasks in v0. `'huginn_job'`, `'muninn_ingestion'`, and `'scheduled_task'` kinds are RESERVED but not emitted. Their persistence layers exist (`core/huginn/`, `agents/muninn/jobs/`) but neither has a session-id linkage today; surfacing them requires per-agent registration hooks. Defer until the first agent that needs it.

**INV-066 (pending-active-only):** Tasks appear in the pending list from Bus-v2 `submit()` time (TaskHandle created, originator contextvars set) until TaskHandle reaches a terminal state (`completed` / `failed` / `canceled`). No grace period; no completed-but-unread state. The result has already been delivered via the existing turn-completion path (skald → message); pending was a forward signal only.

**INV-067 (pending-in-memory-only):** The pending list reflects in-memory Bus-v2 state and resets on process restart. Bus-v2 active TaskHandles are in-memory only; persisting the pending list while the underlying tasks are gone would lie to the gateway. Gateways building long-lived pending UIs MUST handle empty-after-restart as a normal path, not as an error.

**INV-068 (pending-rate-limit-exempt):** `GET /sessions/{id}/pending` and `GET /pending` are structurally exempt from `check_rate_limits` per the polling-cadence rationale: a 5s default polling cadence on a per-user endpoint applied across N sessions would burn the user's GCRA budget at N/5 req/s, pre-empting all message-send capacity. Structural exemption (handlers never call `check_rate_limits`) is the only viable design.

**INV-069 (pending-cross-user-404):** Cross-user pending references return 404 with no body distinguishing missing-vs-cross-user. Mirrors INV-009 single-error-path no-leak pattern. The per-user endpoint (`GET /pending`) automatically scopes to `ctx.user_id` via the bus filter; cross-user data simply does not appear in the result.

**INV-070 (pending-envelope-stable):** The PendingTask envelope is `{task_id, kind, target_agent_id, started_at, eta_seconds, status, session_id, turn_id}` — additive field-additions are allowed but the envelope may never be reordered, renamed, or have fields removed without a version bump. All 8 fields are always present; none are omitted when null (explicit-over-implicit per project feedback).

**INV-071 (pending-kind-additive):** The `kind` enum is additive. `'bus_call'` is the only emitted v0 value. Future kinds documented as RESERVED: `'huginn_job'`, `'muninn_ingestion'`, `'scheduled_task'`. Clients MUST tolerate unknown kinds (treat as opaque, not as errors).

### New function blocks

```contract
FN core.agent_bus_v2.bus.AgentBus.list_active_tasks(originator_session_id: str | None = None, originator_user_id: str | None = None) -> list[TaskHandle]
BRIEF: Linear-scan snapshot of active (non-terminal) TaskHandles with optional filters.
PRE: [PRE-001 soft] both filters optional; no filter → return all active tasks
POST: [POST-001 return_value] returned handles have originator fields matching supplied filters
POST: [POST-002 invariant] terminal-state handles (completed / failed / canceled) NEVER appear
POST: [POST-003 invariant] returned list is a snapshot; concurrent mutations do not affect it
STEPS:
  1. [setup] active = [h for h in self._outstanding.values() if not h.is_terminal()]
  2. [branch] IF originator_session_id is not None: filter to matching session
  3. [branch] IF originator_user_id is not None: filter to matching user
  4. [sequential] RETURN active (new list)
```

```contract
FN core.conversation_api.service.ConversationService.list_pending_for_session(session_id: str, limit: int, after: tuple[str, str] | None) -> tuple[list[dict], str | None]
BRIEF: Per-session pending tasks, paginated. Caller must have verified ownership first.
PRE: [PRE-001 hard] session ownership verified by HTTP handler before this call
POST: [POST-001 return_value] (items, next_cursor) — items is PendingTask dicts; next_cursor null on last page
POST: [POST-002 invariant] items sorted started_at DESC, task_id DESC
POST: [POST-003 invariant] all 8 PendingTask fields always present (INV-070)
```

```contract
FN core.conversation_api.service.ConversationService.list_pending_for_user(user_id: str, limit: int, after: tuple[str, str] | None) -> tuple[list[dict], str | None]
BRIEF: Cross-session per-user pending tasks. user_id is auth-context user_id; cross-user data filtered at bus layer.
PRE: [PRE-001 hard] user_id non-empty; "anonymous" used in dev mode
POST: [POST-001 return_value] same shape as list_pending_for_session
POST: [POST-002 invariant] tasks with null originator_user_id excluded (INV-069)
```

```contract
FN GET /sessions/{session_id}/pending (api.py)
BRIEF: Per-session pending list. Session-ownership-gated. Rate-limit exempt (INV-068).
PRE: [PRE-001 hard] pending.read scope → 403 on miss
PRE: [PRE-002 hard] session exists AND ctx owns it → 404 otherwise (INV-069)
PRE: [PRE-003 soft] limit ∈ [1, 200] AND cursor well-formed → 422 otherwise
POST: [POST-001 return_value] HTTP 200 {items, next_cursor}
POST: [POST-002 invariant] structurally exempt from check_rate_limits (INV-068)
ERRORS:
  Missing scope → 403 AUTH_SCOPE_DENIED
  Session not found / cross-user → 404 SESSION_NOT_FOUND (INV-069)
  Bad cursor → 422 CURSOR_INVALID
  Bad limit → 422 VALIDATION_FAILED
```

```contract
FN GET /pending (api.py)
BRIEF: Cross-session per-user pending list. Rate-limit exempt (INV-068).
PRE: [PRE-001 hard] pending.read scope → 403 on miss
PRE: [PRE-002 soft] limit ∈ [1, 200] AND cursor well-formed → 422 otherwise
POST: [POST-001 return_value] HTTP 200 {items, next_cursor} scoped to ctx.user_id
POST: [POST-002 invariant] structurally exempt from check_rate_limits (INV-068)
ERRORS:
  Missing scope → 403 AUTH_SCOPE_DENIED
  Bad cursor → 422 CURSOR_INVALID
  Bad limit → 422 VALIDATION_FAILED
```

## Amendment — Admin Event Stream (issue #127)

```contract
FN core.conversation_api.events.EventBus.emit(event_type: str, data: dict) -> None
BRIEF: Append an event to the ring buffer and broadcast to all live subscribers.
PRE: [PRE-001 hard] event_type is non-empty string -- raise ValueError
PRE: [PRE-002 hard] data is a dict -- raise TypeError
POST: [POST-001 state_change] AdminEvent appended to ring buffer with id=self._next_id, _next_id incremented
POST: [POST-002 invariant] event id strictly greater than all prior ids (INV-047)
POST: [POST-003 side_effect] each subscriber: put_nowait(event); on QueueFull oldest popped and system.events_dropped enqueued
POST: [POST-004 invariant] caller responsible for emit AFTER canonical commit (INV-048)
ERRORS:
  Empty event_type -> ValueError
  Non-dict data -> TypeError
```

```contract
FN core.conversation_api.events.EventBus.subscribe(last_event_id: int | None) -> AsyncIterator[AdminEvent]
BRIEF: Subscribe a new consumer; replay buffered events from last_event_id then yield live events until disconnect.
PRE: [PRE-001 soft] last_event_id is None or non-negative int
POST: [POST-001 return_value] async iterator: first replay events (if any), then live events
POST: [POST-002 invariant] events yielded in id order; skips signalled via system.replay_gap or system.events_dropped
POST: [POST-003 side_effect] subscriber registered for duration of iteration
POST: [POST-004 side_effect] heartbeat task emits system.heartbeat every 30s while connected
POST: [POST-005 side_effect] subscriber unregistered when iteration cancelled or client disconnects
ERRORS:
  None at EventBus boundary; auth errors handled at HTTP layer
```

```contract
FN core.conversation_api.events.EventBus.start() -> None
BRIEF: Initialise the bus and emit system.startup as event id=1.
PRE: [PRE-001 hard] start() not called before -- raise RuntimeError on second call
POST: [POST-001 state_change] _next_id = 2 after system.startup emit
POST: [POST-002 state_change] buffer contains exactly one event: system.startup
POST: [POST-003 invariant] startup data has {version, git_sha, first_event_id: 1}
ERRORS:
  Called twice -> RuntimeError('EventBus already started')
```

```contract
async FN core.conversation_api.events.EventBus.shutdown() -> None
BRIEF: Emit system.shutdown, drain subscriber queues with 1s budget, close all subscriber iterators.
PRE: [PRE-001 soft] start() may or may not have been called -- shutdown is safe unconditionally
POST: [POST-001 side_effect] system.shutdown emitted if start() was called
POST: [POST-002 side_effect] all subscriber queues receive _CLOSE_SENTINEL within 1.0s
POST: [POST-003 state_change] self._subscribers is empty after this returns
ERRORS:
  None -- shutdown is best-effort, never raises
```

```contract
FN GET /admin/events handler (api.py)
BRIEF: SSE endpoint broadcasting all conv-api events to admin-tier consumers (INV-052).
PRE: [PRE-001 hard] caller has valid bearer token -- 401 otherwise
PRE: [PRE-002 hard] caller has admin.events.read scope -- 403 AUTH_SCOPE_DENIED otherwise
POST: [POST-001 return_value] EventSourceResponse iterating AdminEvent envelopes; SSE id=int event_id; SSE data=JSON envelope
POST: [POST-002 side_effect] connection audited as conversation_api:admin:events:connect with actor_user_id
POST: [POST-003 invariant] rate-limit exempt -- no check_rate_limits call (INV-022/INV-052)
ERRORS:
  Missing/invalid bearer -> 401 AUTH_MISSING or AUTH_INVALID
  Missing scope -> 403 AUTH_SCOPE_DENIED
  Malformed Last-Event-ID -> 400 MALFORMED_REQUEST
```

## Amendment — Admin Session Inspection (issue #176)

Two read-only admin-tier endpoints for diagnosing Bifrost-bound sessions:
`GET /admin/sessions/<session_id>/bifrost` exposes a live `BifrostClient`'s
connection metadata plus the post-filter/post-namespace tool list (sourced
from `session.bifrost_tools`, NOT `bfclient._tools`); `GET
/admin/sessions/<session_id>/tools` exposes the full turn-time-merged view
(builtin + bifrost) mirroring `service.py:1743`'s actual predicate.

### Invariants added

**INV-176-1 (read-only):** Neither endpoint mutates session state, re-handshakes, invokes a tool, or modifies `_bifrost_clients[session_id]` / `session.bifrost_tools`.

**INV-176-2 (admin-only):** Both endpoints require the `admin.sessions.read` scope; non-admin tiers receive 403 `auth_scope_denied`. The auth gate fires before any session lookup or `_bifrost_clients` access.

**INV-176-3 (distinct 404 codes):** `session_not_found` (no such session) and `session_not_bifrost_bound` (session exists, no live client) are different `error_code` values so clients can disambiguate without parsing the message. Both codes are members of `ErrorCode`; `SESSION_NOT_BIFROST_BOUND = "session_not_bifrost_bound"` is new in this amendment.

**INV-176-4 (graceful-on-stale, live-client only):** When a live `BifrostClient` is registered but `_is_connected == False`, `/bifrost` returns 200 with `connected: false` and the cached `session.bifrost_tools`. No re-handshake. This does NOT extend to the no-live-client case — INV-176-3 governs there.

**INV-176-5 (override tools excluded):** `/tools` `bifrost_tools` reflects only session-bound binding state; per-message overrides (#166) are turn-scoped and never appear.

**INV-176-6 (audit emission on success):** Each 200 triggers exactly one `_audit_admin_action` write with action `conversation_api:admin:session:read.bifrost` or `conversation_api:admin:session:read.tools` and `extra={"target_session_id": session_id}`. Failures (403/404) are not audited at the resource layer.

**INV-176-7 (Pydantic schema stability):** `BifrostInspectionResponse` and `SessionToolsResponse` use `extra="forbid"` (#158). Field renames/removals are breaking changes.

**INV-176-8 (no no-client fallback for /bifrost):** `/bifrost` requires a live `_bifrost_clients[session_id]` entry. Restored sessions without a live client return 404 `session_not_bifrost_bound` — connection metadata cannot be reconstructed from persisted session state.

**INV-176-9 (/tools mirrors turn-time merge predicate):** `bifrost_tools` is populated only when BOTH `session.bifrost_tools` is non-empty AND a live `_bifrost_clients[session_id]` entry exists (matching `service.py:1743`). Otherwise `[]`.

**INV-176-10 (tool-list source is session.bifrost_tools):** Both endpoints' `tools` / `bifrost_tools` fields are sourced from `session.bifrost_tools` (post-tier-filter / post-namespace), NOT from `bfclient._tools`. Surfacing the raw handshake list would leak tier-filtered tools.

### Function blocks

```contract
async FN GET /admin/sessions/{session_id}/bifrost handler (api.py)
BRIEF: Admin-tier read of a Bifrost-bound session's connection state + post-filter tool list. Read-only; graceful-on-stale (live-client only).
PRE: [PRE-176-1 hard] caller has admin.sessions.read scope -- 403 AUTH_SCOPE_DENIED otherwise
PRE: [PRE-176-3 hard] ErrorCode.SESSION_NOT_BIFROST_BOUND exists -- import-time assertion
POST: [POST-176-1 return_value] BifrostInspectionResponse{endpoint_url, consumer_id, connected, capabilities_granted, tools}; tools == list(session.bifrost_tools or [])
POST: [POST-176-2 exception] no live _bifrost_clients entry -> 404 SESSION_NOT_BIFROST_BOUND
POST: [POST-176-3 exception] no such session -> 404 SESSION_NOT_FOUND
POST: [POST-176-5 side_effect] success emits one _audit_admin_action with action="conversation_api:admin:session:read.bifrost"
POST: [POST-176-6 state_change] no state mutation
ERRORS:
  Missing bearer -> 401 AUTH_MISSING/AUTH_INVALID
  Missing scope -> 403 AUTH_SCOPE_DENIED
  No session -> 404 SESSION_NOT_FOUND
  No live client -> 404 SESSION_NOT_BIFROST_BOUND
```

```contract
async FN GET /admin/sessions/{session_id}/tools handler (api.py)
BRIEF: Admin-tier read of the full post-merge tool list (builtin + bifrost), mirroring the turn-time merge predicate.
PRE: [PRE-176-2 hard] caller has admin.sessions.read scope -- 403 AUTH_SCOPE_DENIED otherwise
POST: [POST-176-7 return_value] SessionToolsResponse{agent_id, builtin_tools, bifrost_tools}
POST: [POST-176-8 return_value] bifrost_tools == [] when no live _bifrost_clients entry (covers non-bound AND restored-no-live-client)
POST: [POST-176-9 return_value] bifrost_tools == list(session.bifrost_tools) when both non-empty AND live client present
POST: [POST-176-10 return_value] per-message override tools (#166) never appear
POST: [POST-176-12 side_effect] success emits one _audit_admin_action with action="conversation_api:admin:session:read.tools"
ERRORS:
  Missing scope -> 403 AUTH_SCOPE_DENIED
  No session -> 404 SESSION_NOT_FOUND
```

## Invariants — Upload subsystem (issue #118)

**INV-053 (uploads-user-scoped):** All uploads addressed by `upload_id` are scoped to the creating `user_id`; cross-user reference returns 404 per INV-009.

**INV-054 (uploads-mime-allowlist):** MIME type validated against config-driven allowlist via magic-byte signature, NOT `Content-Type` header. Mismatch deletes the tmp file and returns 415.

**INV-055 (uploads-quota-enforced):** Per-user quota (`max_total_bytes` + `max_count`) checked incrementally during stream. Over-quota uploads return 413 `quota_exceeded`.

**INV-056 (uploads-ttl-bounded):** Uploads expire at `expires_at`. References after expiry return 410 `upload_expired`. Background sweep every 5 minutes plus lazy fallback at access.

**INV-057 (uploads-magic-bytes):** MIME type authority is the magic-byte signature, not the `Content-Type` header.

**INV-058 (uploads-atomic-write):** Writes use tmp + fsync + rename. Partial uploads never visible at canonical path.

**INV-059 (uploads-id-format):** `upload_id` matches `^upl_[a-f0-9]{32}$`. Server-generated only, never user-supplied.

**INV-060 (uploads-path-containment):** Upload paths must resolve under `${WORLDTREE_DATA_PATH}/uploads/`. Defense in depth against traversal.

**INV-061 (uploads-channels-dual):** `agent_turn` populates two channels — LLM content blocks for `image/*` MIMEs only when `supports_vision` is true; `ctx.tool_context['uploads']` always populated with Upload objects regardless of MIME.

**INV-062 (uploads-llm-image-only):** LLM content blocks NEVER receive non-image MIMEs. PDFs and text always go through the tool channel.

**INV-063 (uploads-validation-gate):** Messages with `upload_ids` require `agent.capabilities` to include `accepts_uploads` or return 422 `agent_lacks_upload_support`.

**INV-064 (uploads-not-in-message-content):** `messages.content` stays text-only. Upload references persist as a separate `messages.upload_ids` JSON column.

```contract
FN core.conversation_api.uploads.UploadManager.create_from_stream(file: UploadFile, user_id: str) -> Upload
BRIEF: Stream multipart-uploaded bytes to disk with atomic write, magic-byte validation, sha256 streaming, and incremental quota enforcement.
PRE: [PRE-001 hard] file is a valid FastAPI UploadFile; user_id is a non-empty string matching [a-zA-Z0-9_-]+
POST: [POST-001 state_change] canonical file written at <root>/<user_id>/<upload_id>; sha256 + size + mime_type + expires_at recorded in uploads table
POST: [POST-002 invariant] partial uploads NEVER visible at canonical path (atomic write per INV-058)
POST: [POST-003 invariant] returned Upload's user_id matches the user_id arg; upload_id matches ^upl_[a-f0-9]{32}$
POST: [POST-004 invariant] sha256 in returned Upload matches a recomputed sha256 of the canonical file
ERRORS:
  Stream exceeds max_upload_bytes mid-read → raise UploadTooLarge; tmp file deleted
  User over total_bytes quota → raise QuotaExceeded(limit_type='total_bytes'); tmp file deleted
  User over max_count quota → raise QuotaExceeded(limit_type='max_count'); tmp file deleted
  Magic-byte signature not in allowlist → raise MimeTypeDisallowed(detected, allowed); tmp file deleted
```

```contract
FN core.conversation_api.uploads.UploadManager.get_metadata(upload_id: str, user_id: str | None) -> Upload | None
BRIEF: Fetch upload metadata with ownership filter. Returns None for missing/cross-user/soft-deleted. Raises UploadExpired when the row exists but has passed its TTL.
POST: [POST-001 return_value] returns Upload object when upload exists AND user_id matches AND deleted_at IS NULL AND expires_at >= now()
POST: [POST-002 return_value] returns None when upload is missing, cross-user, or soft-deleted (deleted_at IS NOT NULL)
POST: [POST-003 side_effect] raises UploadExpired(upload_id, expires_at) when the row exists, user_id matches, deleted_at IS NULL, AND expires_at < now()
```

```contract
FN core.conversation_api.uploads.UploadManager.delete(upload_id: str, user_id: str) -> DeleteResult
BRIEF: Soft-delete an upload owned by the given user. Idempotent.
POST: [POST-001 return_value] DeleteResult(found=False, already_deleted=False, deleted_at=None) when upload is missing or cross-user
POST: [POST-002 return_value] DeleteResult(found=True, already_deleted=False, deleted_at=<now>) on first delete — file unlinked, row deleted_at set
POST: [POST-003 return_value] DeleteResult(found=True, already_deleted=True, deleted_at=<original>) on subsequent calls
POST: [POST-004 invariant] cross-user call returns found=False (no leak per INV-009)
```

```contract
FN core.conversation_api.uploads.UploadManager.sweep_expired() -> int
BRIEF: Hard-delete all expired rows (active and soft-deleted alike) + their files; return count deleted.
POST: [POST-001 state_change] all rows with expires_at < now() are hard-deleted regardless of deleted_at value; files unlinked from disk
POST: [POST-002 return_value] count of rows hard-deleted
```

```contract
FN core.conversation_api.uploads.UploadManager.list_for_user(user_id: str, limit: int, after_cursor: str | None) -> tuple[list[Upload], str | None]
BRIEF: Paginated listing of user's uploads via #121's Cursor convention. Sorted by created_at DESC.
POST: [POST-001 return_value] (items, next_cursor) where len(items) ≤ limit; next_cursor is null on last page
POST: [POST-002 invariant] items sorted by created_at DESC, upload_id DESC
POST: [POST-003 invariant] only the calling user's non-deleted, non-expired uploads appear
ERRORS:
  CursorInvalid → propagate (HTTP layer maps to 422 cursor_invalid)
```

```contract
FN POST /uploads handler (api.py)
BRIEF: Multipart upload endpoint. Returns the upload metadata envelope; per-user quota'd; magic-byte validated.
PRE: [PRE-001 hard] caller has uploads.write scope — 403 otherwise
PRE: [PRE-002 hard] request is multipart/form-data with field 'file' — 422 otherwise
POST: [POST-001 return_value] HTTP 200 with the Upload envelope dict — all 8 fields present
ERRORS:
  Missing scope → 403 AUTH_SCOPE_DENIED
  Stream exceeds max_upload_bytes → 413 upload_too_large
  Quota exceeded → 413 quota_exceeded with limit_type
  MIME disallowed → 415 mime_type_disallowed with detected + allowed list
```

```contract
FN DELETE /uploads/{upload_id} handler (api.py)
BRIEF: Idempotent revoke endpoint. Branches on DeleteResult.
POST: [POST-001 return_value] 404 when upload missing or cross-user (DeleteResult.found=False)
POST: [POST-002 return_value] 200 {upload_id, deleted: true, deleted_at} on first call
POST: [POST-003 return_value] 200 {upload_id, deleted: false, reason: 'already_deleted', deleted_at} on subsequent calls
POST: [POST-004 side_effect] audit entry conversation_api:uploads:delete written on first delete only
```

```contract
FN agent_turn._inject_uploads(working_messages: list[dict], uploads: list[Upload], llm_supports_vision: bool, tool_context: dict) -> None
BRIEF: Two-channel routing per INV-061/INV-062. Pure mutation.
POST: [POST-001 state_change] tool_context['uploads'] = list(uploads) — ALWAYS, even when uploads is empty
POST: [POST-002 state_change] when llm_supports_vision is True AND any upload's mime_type in IMAGE_MIME_ALLOWLIST: last user message content rewritten to list of content blocks
POST: [POST-003 invariant] non-image MIMEs NEVER added to the LLM channel (INV-062)
POST: [POST-004 invariant] when llm_supports_vision is False or no image MIMEs: working_messages.content stays as a string
```

---

## Invariants: Search (issue #122)

INV-072 (search-fts5-backend): cross-session search uses SQLite FTS5 in v0 — client code MUST NOT depend on FTS5-specific behavior; the wire shape is backend-agnostic so future Postgres-FTS or external-search-engine migration does not break clients.

INV-073 (search-cross-user-scoped): all search results are filtered by `sessions.user_id` matching the calling SecurityContext's `user_id`; cross-user data CANNOT leak even via crafted FTS queries. Anonymous / dev mode (user_id=None) sees all sessions (same as `GET /sessions` dev-mode behavior).

INV-074 (search-bm25-ranking): results sorted by BM25 score ASC then message_id DESC; emitted score is the raw BM25 value (typically negative; lower = better).

INV-075 (search-fts-sync-via-triggers): `messages_fts` is kept in sync with `messages` via three SQLite triggers AFTER INSERT/UPDATE/DELETE; no Python-side maintenance is reliable.

INV-076 (search-tool-content-excluded): only `messages.content` is searched; tool-call intermediates are excluded by INV-002 storage and consequently never appear in search results.

INV-077 (search-archived-default-include): `GET /search` includes hits from archived sessions by default; `?include_archived=false` opt-out narrows to non-archived; this DIFFERS from `GET /sessions` where archived are excluded by default — the divergence is search-consistency-by-design.

INV-078 (search-cursor-shape): cursor body is `{v: 1, r: <bm25_score_float>, i: <message_id_int>}`; sort key is `(score ASC, message_id DESC)`; anchor predicate is `(score > anchor_r) OR (score = anchor_r AND message_id < anchor_i)`.

INV-079 (search-query-syntax-exposed): the `?q=` parameter is passed to FTS5 MATCH directly; phrase / boolean / prefix syntaxes are supported; on FTS5 syntax errors, a 422 with `error_code='search_query_invalid'` is returned (additive to the ErrorCode enum).

INV-080 (search-no-audit): individual searches are NOT audited — search is read-only and high-volume; differs from upload create/delete which IS audited (small-volume, mutation, integrity-relevant).

---

## Function-level contracts: Search (issue #122)

```contract
FN ConversationStore.search(user_id: str | None, query: str, limit: int, after: tuple[float, int] | None = None, session_id: str | None = None, role: str | None = None, after_ts: str | None = None, before_ts: str | None = None, include_archived: bool = True) -> list[dict]
BRIEF: Run the BM25-ranked, user-scoped FTS query with optional filters; return raw rows for the service layer to paginate.
PRE: [PRE-001 hard] query is non-empty; limit >= 1
PRE: [PRE-002 hard] role (if provided) ∈ {'user', 'assistant'}
PRE: [PRE-003 soft] after_ts / before_ts are valid ISO 8601 if provided (caller validates)
POST: [POST-001 return_value] list[dict] each with {session_id, message_id, agent_id, role, content, snippet, score, created_at} — at most limit rows
POST: [POST-002 invariant] when user_id is not None, every returned row's session satisfies (sessions.user_id = user_id OR sessions.user_id IS NULL) per INV-073
POST: [POST-003 invariant] rows sorted by bm25 score ASC, message_id DESC per INV-074
POST: [POST-004 invariant] when after is provided, results satisfy (score > after_r) OR (score = after_r AND message_id < after_i)
POST: [POST-005 invariant] when include_archived=False, sessions.archived = 0 filter applied per INV-077
POST: [POST-006 invariant] when after_ts/before_ts provided, messages.created_at IS NOT NULL filter applied
ERRORS:
  Invalid FTS5 query syntax → sqlite3.OperationalError propagates (handler maps to 422 search_query_invalid per INV-079)
```

```contract
FN core.conversation_api.pagination.build_search_cursor(score: float, message_id: int) -> str
BRIEF: Build a search-result cursor from a (score, message_id) anchor.
PRE: [PRE-001 hard] score is a float; message_id is a non-negative int
POST: [POST-001 return_value] decodes back to {'v': 1, 'r': score, 'i': message_id}
STEPS:
  1. [sequential] RETURN Cursor.encode({'v': 1, 'r': score, 'i': message_id})
```

```contract
FN core.conversation_api.pagination.search_cursor_anchor(token: str) -> tuple[float, int]
BRIEF: Decode a search-result cursor to its (score, message_id) tuple.
PRE: [PRE-001 hard] token is a well-formed search cursor — raise CursorInvalid otherwise
POST: [POST-001 return_value] returns (score_float, message_id_int)
ERRORS:
  Cursor.decode raises → propagate CursorInvalid
  Body missing 'r' or 'i' key → raise CursorInvalid('search cursor body missing required keys')
  Body 'r' is not numeric or 'i' is not int (or is bool) → raise CursorInvalid('search cursor body has wrong field types')
  int 'r' value → coerced to float (valid)
```

```contract
FN ConversationService.search(user_id: str | None, query: str, limit: int, cursor: str | None, session_id: str | None, role: str | None, after_ts: str | None, before_ts: str | None, include_archived: bool) -> tuple[list[dict], str | None]
BRIEF: User-facing search: decode cursor, run store.search, paginate, build next_cursor.
PRE: [PRE-001 hard] limit ∈ [1, 200] — caller (HTTP handler) enforces
PRE: [PRE-002 soft] cursor is None or a valid search cursor; CursorInvalid raised on malformed
POST: [POST-001 return_value] (items, next_cursor) where len(items) ≤ limit; next_cursor is null on last page
POST: [POST-002 invariant] items projected into 8-field SearchHit shape per INV-073/INV-074
POST: [POST-003 invariant] sort and pagination semantics inherited from store.search per INV-074 and INV-078
ERRORS:
  CursorInvalid → propagate (HTTP layer maps to 422 cursor_invalid)
  sqlite3.OperationalError (FTS syntax) → propagate (HTTP layer maps to 422 search_query_invalid)
```

```contract
FN GET /search handler (api.py)
BRIEF: Cross-session text-search endpoint with FTS5 backend.
PRE: [PRE-001 hard] caller has search.read scope — 403 otherwise
PRE: [PRE-002 hard] q is 1–200 chars (FastAPI Query bounds enforce) — 422 otherwise
PRE: [PRE-003 soft] limit ∈ [1, 200] AND cursor (if present) is well-formed — 422 otherwise
PRE: [PRE-004 soft] role (if present) is 'user' or 'assistant' — 422 otherwise
PRE: [PRE-005 soft] after / before (if present) are valid ISO 8601 with Z — 422 otherwise
POST: [POST-001 return_value] HTTP 200 with {items: list[SearchHit], next_cursor: str | null}
POST: [POST-002 invariant] rate-limit consumed (per-user request budget) — NOT exempt per INV-079
POST: [POST-003 invariant] no audit entry written per INV-080
ERRORS:
  Missing scope → 403 AUTH_SCOPE_DENIED
  Bad q length → 422 validation_failed (FastAPI default)
  Bad cursor → 422 cursor_invalid
  Bad date format → 422 validation_failed
  FTS syntax error → 422 search_query_invalid (with the FTS error message in detail.message)
  Rate-limited → 429 rate_limited
```

## Invariants — Tool-Call Persistence (issue #123)

INV-081 (tool-events-opt-in): tool-call intermediates are persisted only when the session was created with `record_tool_intermediates=true`. The flag is sticky for the session's lifetime; no mid-session toggle. Default is false. INV-002 is REVISED to permit this opt-in path; the default-off behavior of INV-002 is preserved for sessions created without the flag.

INV-082 (tool-events-metadata-only): the `tool_events` table stores `tool_call_id`, `session_id`, `turn_id`, `tool_name`, `started_at`, `ended_at`, `status`, `error_type`, `error_msg` (truncated to ≤200 chars per INV-049 PII discipline), `duration_ms`. It NEVER stores tool arguments and NEVER stores tool results. The PII attack surface of tool payloads is the reason; this mirrors INV-049 (admin events MUST NOT carry tool payloads).

INV-083 (tool-events-best-effort): persist failures (SQLite locked, disk full, etc.) are logged at WARNING and metric-counted but NEVER fail the turn. The SSE stream continues uninterrupted. Strict-mode (turn-fatal on persist failure) is a future flag if compliance demands it.

INV-084 (tool-events-after-emit): the persist write happens AFTER the SSE event is published to the live stream, never before. The live stream is the source of truth for the agent loop; the persistent table is observability. A consumer that reads `GET /tool-events` after seeing a `tool_result` SSE event MUST eventually see the row, but the persist write is non-blocking on the stream.

INV-085 (tool-events-cursor-shape): the `list_tool_events` cursor body is `{v: 1, s: <started_at_iso8601_z>, i: <tool_call_id>}`. Sort key is `(started_at DESC, tool_call_id DESC)`. Anchor predicate is `(started_at, tool_call_id) < (anchor_s, anchor_i)` strict-less-than. Cursors are opaque base64 envelopes via the same `pagination.Cursor` primitive as #121 messages and #122 search.

INV-086 (tool-events-cancellation): when a turn is cancelled before the corresponding `tool_result` fires, the `tool_events` row persists with `status='cancelled'` and `ended_at=NULL`. Captures the audit-trail truth ("this tool started; the turn was cancelled before it finished") without forcing a misleading `'completed'` status. `duration_ms` carries the elapsed wall time from start to cancel for debugging context.

INV-087 (tool-events-admin-discipline): emitting a `session.tool_recording_enabled` admin event at session create (when the flag is set) is allowed and contains `{session_id, agent_id, user_id}` only. Per-write `tool.recorded` admin events are PROHIBITED — they would dwarf the admin event stream and create the same payload-in-events PII risk that INV-049 forbids. Persistence failures bump a counter; they don't emit admin events.

## Function-level contracts: Tool-Call Persistence (issue #123)

```contract
FN core.conversation_api.store.ConversationStore._migrate_tool_events() -> None
BRIEF: Idempotent migration — create tool_events table + index, add sessions.record_tool_intermediates column.
PRE: [PRE-001 hard] self._conn is an open SQLite connection
POST: [POST-001 state_change] tool_events table exists with columns (tool_call_id PK, session_id, turn_id, tool_name, started_at, ended_at, status DEFAULT 'started', error_type, error_msg, duration_ms)
POST: [POST-002 state_change] index idx_tool_events_session_started exists on (session_id, started_at DESC, tool_call_id DESC)
POST: [POST-003 state_change] sessions.record_tool_intermediates INTEGER NOT NULL DEFAULT 0 column exists
POST: [POST-004 invariant] running this method twice is a no-op (CREATE TABLE IF NOT EXISTS + colcheck-gated ALTER)
ERRORS:
  None — all DDL is idempotent
```

```contract
FN core.conversation_api.store.ConversationStore.record_tool_event_started(session_id: str, turn_id: int, tool_call_id: str, tool_name: str) -> None
BRIEF: Insert a tool_events row at start time with status='started' and ended_at=NULL.
PRE: [PRE-001 hard] all four arguments non-empty
POST: [POST-001 state_change] tool_events row exists with status='started', ended_at=NULL, started_at=_now()
POST: [POST-002 invariant] duplicate calls with the same tool_call_id are no-ops (INSERT OR IGNORE)
ERRORS:
  sqlite3.OperationalError on locked DB / disk full — RAISED for the caller to swallow per INV-083
```

```contract
FN core.conversation_api.store.ConversationStore.record_tool_event_completed(tool_call_id: str, status: str, duration_ms: float, error_type: str | None = None, error_msg: str | None = None) -> None
BRIEF: Update an existing tool_events row at completion time, setting status, duration_ms, error fields, and (conditionally) ended_at.
PRE: [PRE-001 hard] tool_call_id non-empty; status one of 'completed', 'failed', 'cancelled'
PRE: [PRE-002 soft] error_msg ≤200 chars (caller responsibility per INV-082)
POST: [POST-001 state_change] tool_events row updated: status, duration_ms, error_type, error_msg
POST: [POST-002 invariant] when status='cancelled', ended_at remains NULL per INV-086; otherwise ended_at=_now()
POST: [POST-003 invariant] missing tool_call_id row → silent no-op (UPDATE WHERE matches zero rows)
ERRORS:
  ValueError on invalid status (caller bug)
  sqlite3.OperationalError on locked DB / disk full — RAISED for caller to swallow per INV-083
```

```contract
FN core.conversation_api.store.ConversationStore.list_tool_events(session_id: str, limit: int, after: tuple[str, str] | None = None) -> list[dict]
BRIEF: Return tool_events rows for a session, ordered by (started_at DESC, tool_call_id DESC), optionally cursor-anchored.
PRE: [PRE-001 hard] session_id non-empty; limit ≥ 1
POST: [POST-001 return_value] list of dicts with the documented 10-field shape (tool_call_id, session_id, turn_id, tool_name, started_at, ended_at, status, error_type, error_msg, duration_ms)
POST: [POST-002 invariant] when after is None, returns up to `limit` rows from the most recent
POST: [POST-003 invariant] when after=(s, i) is provided, returns rows where (started_at, tool_call_id) < (s, i) strict per INV-085
```

```contract
FN core.conversation_api.pagination.build_tool_events_cursor(started_at: str, tool_call_id: str) -> str
BRIEF: Pack a tool-events cursor.
PRE: [PRE-001 hard] both args are non-empty strings
POST: [POST-001 return_value] Cursor.encode({'v': 1, 's': started_at, 'i': tool_call_id}) — base64url-safe opaque string
```

```contract
FN core.conversation_api.pagination.tool_events_cursor_anchor(token: str) -> tuple[str, str]
BRIEF: Decode a tool-events cursor and validate structure.
PRE: [PRE-001 hard] token is a non-empty string
POST: [POST-001 return_value] (started_at, tool_call_id) tuple
ERRORS:
  CursorInvalid on missing keys, wrong types, empty strings, or unsupported version
```

```contract
FN core.conversation_api.service.ConversationService.list_tool_events(session_id: str, user_id: str, limit: int, cursor: str | None) -> tuple[list[dict], str | None]
BRIEF: Cursor-paginated wrapper around store.list_tool_events.
PRE: [PRE-001 hard] session_id, user_id non-empty; limit in [1, 200]
POST: [POST-001 return_value] (items, next_cursor) where items is list of dict envelopes and next_cursor is str or None
ERRORS:
  CursorInvalid on bad cursor — raises through to handler for 422 mapping
```

```contract
FN GET /sessions/{session_id}/tool-events handler (api.py)
BRIEF: Returns persisted tool-call metadata for a session, paginated.
PRE: [PRE-001 hard] caller has valid bearer token (existing get_security_context dependency)
PRE: [PRE-002 hard] caller has tool_events.read scope — 403 AUTH_SCOPE_DENIED otherwise
PRE: [PRE-003 hard] session belongs to caller's user_id (or caller is admin) — 404 otherwise (INV-009 — never leak existence)
POST: [POST-001 return_value] {session_id, items: list[dict], next_cursor: str | None}
POST: [POST-002 invariant] sessions opted-out (default) return items=[] regardless of how many turns ran
ERRORS:
  Missing/invalid bearer → 401
  Missing scope → 403 AUTH_SCOPE_DENIED
  Cross-user / unknown session → 404 SESSION_NOT_FOUND
  Bad cursor → 422 CURSOR_INVALID
  Bad limit → 422 VALIDATION_FAILED (FastAPI Query validation)
```

## Invariants — Transient Characters (issue #153)

INV-088 (transient-characters-ephemeral): transient characters live in-memory only; process restart drops them. Worldtree NEVER persists character JSON or state JSON to disk.

INV-089 (transient-characters-user-scoped): each character belongs to exactly one user_id (the creating user); cross-user reference returns 404 (mirrors INV-009 leak discipline).

INV-090 (transient-characters-sliding-ttl): TTL is sliding — refreshed by every turn within a binding session and by GET /characters/{id}/state. Default 4h, max 24h, configurable per-create within bounds via the ttl_seconds field on CharacterSchema.

INV-091 (transient-characters-sweep-respects-in-flight): the 60s sweep task refuses to evict a character whose binding sessions have any in-flight turn. Refused characters are reconsidered next cycle.

INV-092 (transient-characters-detach-on-delete): DELETE /characters/{id} does NOT cascade-delete binding sessions; the character is simply removed and binding sessions' next turn returns 410 character_not_found.

INV-093 (transient-characters-state-schema-versioned): CharacterSchema and CharacterStateSchema both carry schema_version: "1"; servers refuse unsupported versions with 422 state_schema_outdated + accepted_versions: ["1"] in detail. Forward-compat with future "2" handled via accepted_versions list expansion in a v2 amendment.

INV-094 (transient-characters-pii-discipline): admin events (character.created / character.deleted / character.expired) carry {character_id, user_id} ONLY. Never the consumer-supplied name, description, narrative, OCEAN values, or voice_profile_block.

INV-095 (transient-characters-displaces-persona-only): when character_id is bound to a session, the character displaces the persona+model layer ONLY. The system prompt, tool set, and capabilities still come from agent_id (typically agents/actor/). The character does NOT swap LLM providers — the model field selects from the available_for_characters allowlist; provider is a property of the model profile.

INV-096 (transient-characters-quota): per-user cap (default 100) enforced at create time; over-cap returns 429 quota_exceeded. Configurable via conversation_api.transient_characters.max_characters_per_user.

INV-097 (transient-characters-no-valence-write): build_persona_from_character produces a Persona with NO SQLite valence DB write. Transient characters do not accumulate relational valence (the consumer's own state model owns cross-character relationships if needed).

INV-098 (text-boundary-additive): the `text_boundary` SSE event is purely additive. Clients ignoring it see Worldtree's existing behaviour bit-identically. No mode-switch query parameter, no flag-on-existing-text-event approach. One consistent wire shape across all agents.

INV-099 (boundary-after-text): each `text_boundary` event MUST be yielded AFTER its parent `text` event in the stream. The BoundaryDetector runs after each ContentDeltaEvent yield; `stream_turn` emits the boundary as a separate SSE event immediately following the text.

INV-100 (boundary-checkpoint-not-commit): a `text_boundary` is a checkpoint, not a commitment. Cancellation between two boundaries discards unbuffered text; no synthesised final boundary on cancel. Boundaries do NOT change cancel semantics.

INV-101 (text-boundary-no-admin-event): boundaries do NOT emit admin events. Per-boundary admin events would dwarf the admin stream and provide no operational signal beyond the existing turn-summary aggregate.

INV-102 (implicit-narration): tool-call narration is implicit. The presence-or-absence of a `text` event before `tool_start` IS the suppression signal. NO new field on `tool_start`. Agent prompt-engineering teaches the model when to narrate.

INV-103 (classifier-markers-static-per-agent): `voice.classifier_markers` is static for the agent's lifetime — cannot be flipped mid-session. Re-rendering the system prompt mid-turn is not supported.

INV-104 (classifier-markers-pii-discipline): the classifier-marker prompt block contains NO per-session content. Admin events do NOT carry classifier-marker payloads. The `<system>` / `<character>` tags appear inline in `text` event content as authored by the LLM; gateway parses client-side.

## Ephemeral Template Surface (issue #161)

Ephemeral templates are a new agent kind that bypass persona, memory, tools, and motivational injection. Session config (system_prompt + model) is supplied by the consumer at session-create time and frozen as a snapshot; subsequent turns use that snapshot verbatim.

**Invariants added by issue #161:**

- **INV-161-1 (ephemeral-template-bypass)**: For sessions where `session.ephemeral_config is not None`, `PersonaRegistry.inject_context` is NOT called pre-turn; `PersonaRegistry.update_after_turn` is NOT called post-turn; valence side-channel is NOT called; tool list passed to provider is `[]`.
- **INV-161-2 (frozen-session-config)**: Once a session is created with an `ephemeral_config` snapshot, subsequent mutations to `agents/saga/config.yaml`, `config/providers.yaml → saga_allowed_models`, or `config/defaults.yaml → saga.default_model` do NOT affect that session's per-turn `system_prompt` or `model`.
- **INV-161-3 (no-tools-for-ephemeral)**: Tool list passed to the provider for an ephemeral session is `[]` regardless of any `tools:` block in the template's config.yaml.
- **INV-161-4 (foundational-flow-unchanged)**: For sessions where `session.ephemeral_config is None`, the per-turn path is bit-identical to pre-#161 — same system_prompt loading, same persona injection, same tool list, same audit-log shape.
- **INV-161-5 (config-required-for-ephemeral-create)**: `POST /sessions` against an ephemeral template MUST reject the request with 422 if `config` is missing or fails any validation step.
- **INV-161-6 (model-allowlist-enforcement)**: `config.model`, when supplied, MUST be in `saga_allowed_models` at session-create time. When omitted, server resolves to `saga.default_model` (startup-validated to be in the allowlist).
- **INV-161-7 (full-prompt-in-audit)**: Session-create audit entries for ephemeral sessions include `tier: 2` and `ephemeral_config` (full JSON).
- **INV-161-8 (cross-user-isolation)**: A Saga session created by user A is invisible to user B — `GET /sessions/{id}` returns 404.
- **INV-161-9 (foundational-rejects-config)**: `POST /sessions { agent_id: "<foundational>", config: {...} }` returns 422 with `error_code: "foundational_does_not_accept_config"`.
- **INV-161-10 (capabilities-public-shape)**: `GET /capabilities` is callable by any authenticated key. The response has `ephemeral_templates` at top-level.
- **INV-161-11 (template-kind-immutable-at-runtime)**: The `kind` field on a loaded `AgentContext` is set once at startup and never mutated.

**New error codes (issue #161):**

| code | HTTP | trigger |
|---|---|---|
| `ephemeral_requires_config` | 422 | saga session without `config:` |
| `foundational_does_not_accept_config` | 422 | foundational agent with `config:` |
| `system_prompt_required` | 422 | `config.system_prompt` missing or null |
| `system_prompt_empty` | 422 | `config.system_prompt` whitespace-only |
| `system_prompt_too_large` | 422 | > 32768 bytes UTF-8 |
| `model_not_allowed` | 422 | model not in `saga_allowed_models` |

**New `AgentContext` fields:** `kind: str = "foundational"`, `saga_allowed_models: list | None`, `saga_default_model: str | None` — populated for ephemeral templates, `None` for foundational agents.

**Startup failfast:** server refuses to start if `agents/saga/config.yaml` is missing/malformed OR `saga.default_model` is not in `saga_allowed_models`. Raises `ConfigurationError` before binding any port.

**Function-level contracts for issue #161** are documented in `docs/contracts/issues/161.contract.md`.

## Function-level contracts: Transient Characters (issue #153)

Implemented across three commits per `docs/contracts/issues/153.contract.md`. Function blocks for CharacterManager (create / get_state / delete / refresh_ttl / get_persona / _sweep_loop), build_persona_from_character, GET /models/available-for-characters, POST /characters, GET /characters/{id}/state, DELETE /characters/{id}, and the POST /sessions character_id extension are documented in the issue-scoped contract; the module-level invariants above govern their behaviour going forward.

## Bifrost MCP-in-Reverse Binding (issue #160)

Bifrost allows consumers to expose tools to Worldtree agents. `POST /sessions` accepts an optional `bifrost: {endpoint_url, scope?}` field. When present, a synchronous handshake runs before the 201 response. On failure, 502 is returned and no session row is created.

**New session field:** `ConversationSession.bifrost_tools: list[dict] | None` — populated at session-create from the handshake response (filtered by tier, namespaced as `bifrost.<consumer-id>.<tool-name>`). Persisted as `bifrost_tools_json TEXT` in the sessions table (forward-only migration).

**New `ErrorCode` value:** `bifrost_handshake_failed` — returned in 502 responses when the Bifrost handshake fails. The spec-level Bifrost error code rides in `detail.bifrost_error`.

**New Heimdall scope:** `bifrost:invoke` — added to the `user` tier in `config/policies.yaml`. Gates the ability to create a Bifrost-bound session.

**Invariants added by issue #160:**

- **INV-160-1 (handshake-at-create)**: When `POST /sessions` carries `bifrost: {endpoint_url, ...}`, the handshake completes BEFORE the 201 response. No "create session, handshake later" path in v0.1. Verifiable via test: handshake-failing endpoint → 502; session not in store.
- **INV-160-2 (one-connection-per-session)**: Each Bifrost-bound session owns exactly one MCP connection. Two sessions binding to the same `endpoint_url` open two independent connections. No pooling, no sharing.
- **INV-160-3 (saga-incompatible)**: A session cannot be both ephemeral (Saga, `kind: "ephemeral"`) AND Bifrost-bound. Session-create rejects with 422 `ephemeral_does_not_accept_bifrost`. Verifiable: `POST /sessions { agent_id: "saga", config: {...}, bifrost: {...} }` → 422.
- **INV-160-4 (jwt-bound-to-session-expiry)**: JWT TTL is bound to session expiry — far-future `expires_at` for sessions without a fixed TTL. Re-mint happens only when a re-handshake fires (connection-loss recovery). No standalone JWT-staleness check.
- **INV-160-5 (reentrancy-25-per-turn)**: At most 25 successful Bifrost tool invocations per agent turn. The 26th returns `bifrost.reentrancy_cap_exceeded` without contacting the consumer. Counter resets per turn via `BifrostClient.reset_turn_counter()`. Enforced inside `BifrostClient.invoke_tool`.
- **INV-160-6 (tool-list-cached-per-session)**: Bifrost tools are fetched once at handshake and cached on `ConversationSession.bifrost_tools`. Per-turn dispatch reads from the cache; never re-fetches mid-session except on connection-loss recovery.
- **INV-160-7 (cross-consumer-isolation)**: An agent cannot invoke another consumer's Bifrost tools. Session bound to consumer A sees only `bifrost.A.*` tools; consumer B's tools are absent from the agent's tool list.
- **INV-160-8 (tier-filter-at-merge-time)**: Tools whose `bifrost_required_tier` exceeds the session's caller tier are filtered OUT at tool-list-merge time. Tier change mid-session does NOT re-filter (cached state wins until session restart).
- **INV-160-9 (cancellation-fire-and-forget)**: When a turn is cancelled mid-Bifrost-call, the client sends cancel and closes the agent-side wait without waiting for ack. Late `tool_result` for the cancelled `tool_call_id` is dropped.
- **INV-160-10 (no-bifrost-no-overhead)**: Sessions without `bifrost: {...}` at create-time pay zero Bifrost-related overhead. No JWT minting, no connection state, no per-turn tool-list extension.
- **INV-160-11 (audit-redaction-by-default)**: `turn.completed` Bifrost-call entries redact arguments and result content unless the consumer's tool schema sets `bifrost_log_arguments: true`. Default-secure.
- **INV-160-12 (heimdall-scope-gate)**: A session-create request with `bifrost: {...}` MUST come from a caller whose Heimdall scopes include `bifrost:invoke`. Missing scope → 403 (existing Heimdall flow). User tier includes it by default.

## Per-Message Bifrost Endpoint Override (issue #166)

Extends the Bifrost surface to support stateless one-off consumer-MCP calls without binding a Bifrost endpoint to the whole session (bifrost v0.2 spec § 9). The primary driver is Skaldsong's YAML-linting use case: lint one YAML file, get one structured result — no session lifetime worth binding a Bifrost endpoint to.

### New Request Model

`BifrostEndpointOverride` is a Pydantic model added to `core/conversation_api/api.py`:

```python
class BifrostEndpointOverride(BaseModel):
    endpoint_url: str   # HTTPS by default; http://localhost / http://127.0.0.1 always allowed; LAN-plaintext under BIFROST_CLIENT_ALLOWED_HOSTS allowlist (#170)
    consumer_id: str    # Heimdall user_id, non-empty
    scope: str | None = None

    @field_validator("endpoint_url")
    @classmethod
    def _check_endpoint_url(cls, v: str) -> str:
        # Delegates to core.transports._url_guard.check_bifrost_endpoint_url
        # so all three sites (this, BifrostBindingRequest, BifrostClient) stay
        # in lockstep including the env-gated allowlist (#170).
        from core.transports._url_guard import check_bifrost_endpoint_url
        return check_bifrost_endpoint_url(v)
```

**URL allowlist (#170):** Default policy is HTTPS for any routable address; `http://localhost` and `http://127.0.0.1` are always allowed for local development. The env var `BIFROST_CLIENT_ALLOWED_HOSTS` (comma-separated `host:port`) widens the policy for LAN-internal consumers behind a shared trust boundary. The allowlist is **per-deployment, not per-consumer** — any consumer's endpoint at a listed `host:port` passes. Future multi-tenant deployments where one Worldtree fronts multiple LAN-internal consumers will need richer scoping (explicit non-goal for v0). Symmetric with mead-hall's `docs/contracts/bifrost-tracer.contract.md` deferral: both halves explicitly defer production TLS to the same future slice.

`SendMessageRequest` gains `bifrost: BifrostEndpointOverride | None = None` (backward-compatible; absent field = no override).

### New ErrorCode

`BIFROST_CONSUMER_NOT_FOUND = "bifrost_consumer_not_found"` — returned as 502 when `body.bifrost.consumer_id` doesn't match any Heimdall user or matches a user whose `bifrost_jwt_algorithm IS NULL` (not Bifrost-registered).

### Validation and handshake flow (in `send_message` handler)

1. HTTPS URL check — Pydantic field validator; 422 on miss.
2. `bifrost:invoke` scope check — same as session-bound path; 403 on miss.
3. Ephemeral session rejection — 422 `ephemeral_does_not_accept_bifrost` when session is Saga (extends INV-160-3).
4. Heimdall consumer lookup — 502 `bifrost_consumer_not_found` on miss or unregistered.
5. Instantiate a new `BifrostClient` with the override consumer's algorithm + key; set `_jwt_ttl_seconds = 60`.
6. `await override_client.connect()` — 502 `bifrost_handshake_failed` on failure.
7. Capability check: `'per-message-endpoint-override'` MUST be in `capabilities_granted` — 502 `bifrost_handshake_failed` with `bifrost_error: 'bifrost.capability_unavailable'` on miss.
8. Build `override_tools` with namespace `bifrost.<override-consumer-id>.<tool-name>`.
9. Pass `override_client`, `override_tools`, and `bifrost_override_applied=True` to `stream_turn`.

### Changes to `stream_turn`

`ConversationService.stream_turn` gains three new keyword parameters:
- `override_client: Any | None = None` — pre-connected BifrostClient for the override session.
- `override_tools: list[dict] | None = None` — already-namespaced tool list.
- `bifrost_override_applied: bool = False` — flag for `turn.started` event.

When `override_tools` is non-empty, they are merged alongside platform and session-bound tools. When `override_client` is also provided, tool handlers are registered and the per-turn counter is reset (fresh 25-call budget independent of the session-bound client's counter).

The `turn.started` event always carries `bifrost_override_applied: bool` (True/False, never null or absent).

In the `finally` block, `await override_client.disconnect()` is called unconditionally when `override_client is not None`. Disconnect failures are logged but not re-raised.

### Invariants added by issue #166

- **INV-166-1 (override-is-separate-client)**: A per-message override always instantiates a NEW `BifrostClient`, distinct from any session-bound client — even when the override `endpoint_url` matches the session's bound endpoint. Enforced structurally: `stream_turn` receives `override_client` as a local parameter, never consulting `_bifrost_clients[session_id]`.
- **INV-166-2 (ad-hoc-jwt-60s)**: Ad-hoc JWTs minted for override sessions carry `expires_at = issued_at + 60`. Hard-coded via `override_client._jwt_ttl_seconds = 60`; not operator-configurable. Per bifrost v0.2 spec § 9.
- **INV-166-3 (capability-required)**: Override target's handshake response MUST include `'per-message-endpoint-override'` in `capabilities_granted`. Absent → 502 `bifrost.capability_unavailable`. No fallback to session-bound or platform-only mode.
- **INV-166-4 (reentrancy-cap-per-client)**: Each BifrostClient (session-bound or ad-hoc override) tracks its own per-turn 25-call counter. Override calls do NOT increment the session-bound counter; session-bound calls do NOT increment the override counter.
- **INV-166-5 (teardown-on-message-end)**: When a message with `bifrost: {...}` ends (success, error, or cancellation), `override_client.disconnect()` is called in `stream_turn`'s `finally` block. No state survives the message.
- **INV-166-6 (ephemeral-rejects-override)**: Messages with `bifrost: {...}` against a session whose `agent_id` resolves to an ephemeral template are REJECTED at validation (422 `ephemeral_does_not_accept_bifrost`). Extends INV-160-3 to the override surface.
- **INV-166-7 (turn-started-flag-present)**: `turn.started` event always carries `bifrost_override_applied: bool` — `True` when override is in the request, `False` otherwise. Never null, never absent.
- **INV-166-8 (override-tools-namespace-isolation)**: Override tools always namespace as `bifrost.<override-consumer-id>.<tool-name>`. Cross-consumer namespace collision is impossible by construction.
- **INV-166-9 (no-tool-list-cache-across-messages)**: Override tool list is fetched on every message's ad-hoc handshake. Not cached, not reused across messages. Each message re-handshakes, re-fetches tools, dispatches, tears down.


## Amendment — Tier 3 consumer-defined agents (issue #181, Phase 2.0)

The conversation API grows a three-tier agent model. Tier 1 is the
foundational set (Mimir, Bragi, Leif, ...) wired at startup. Tier 2 is
the ephemeral template surface (Saga). Tier 3 is the consumer-defined
class addressed by `<user_id>:<agent_name>` and stored in Heimdall's
SQLite `consumer_agents` table.

### Invariants (Phase 2.0 scope)

- **INV-181-1 (tier3-agent-id-format, Phase 2.0 scope)**: Tier 3
  agent_ids match `^[a-z][a-z0-9-]{2,63}:[a-z][a-z0-9-]{2,63}$`. The
  presence of `:` is the unambiguous Tier 3 discriminator.
- **INV-181-2 (tier3-routing, Phase 2.0 scope)**: `POST /sessions`
  dispatches on `agent_id` content: contains `:` → Tier 3 lookup in
  `consumer_agents`; no `:` → existing Tier 1/2 lookup in
  `_agent_contexts`.
- **INV-181-3 (layer-deferred, Phase 2.0 scope)**: `POST
  /agents/define` returns 422 `layer_deferred` for any non-null
  `persona` / `motivational` / `valence` / `memory`. The fields are
  schema-reserved with `null` default.
- **INV-181-4 (per-key-quota, Phase 2.0 scope)**: Quota counts only
  rows where `owner_key_hash = ctx.api_key_hash AND deleted_at IS
  NULL`. A user holding two keys gets two independent 50-agent
  quotas. Soft-deleted rows within the grace do not count.
- **INV-181-5 (agent-name-immutable, Phase 2.0 scope)**: PATCH rejects
  any payload that includes `agent_name`, returning 422
  `field_not_mutable` BEFORE the DB lookup.
- **INV-181-6 (layer-immutable-in-patch, Phase 2.0 scope)**: PATCH
  rejects payloads carrying any of `persona`, `motivational`,
  `valence`, `memory` even when set to `null`.
- **INV-181-7 (owner-delete-hard, Phase 2.0 scope)**: `DELETE
  /agents/<id>` is a hard-delete; bypasses the 24h grace.
- **INV-181-8 (cascade-key-scoped, Phase 2.0 scope)**: Key revocation
  cascades only to rows with `owner_key_hash = <revoked-key.key_hash>`.
  Other keys for the same user are untouched. The cascade runs in the
  same SQLite database as the api_keys update.
- **INV-181-9 (session-invalidation-key-scoped, Phase 2.0 scope)**:
  Cascade flags only sessions whose stored `owner_key_hash` equals
  the revoked key's hash. Sessions owned by other keys are unaffected.
- **INV-181-10 (end-user-id-passthrough, Phase 2.0 scope)**: Tier 3
  session-create requires non-empty `end_user_id`. Stored on the
  session record but not enforced for cross-namespace isolation in
  Phase 2.0 (that's Phase 2.3 valence-layer territory).
- **INV-181-11 (sweep-grace-honored, Phase 2.0 scope)**: Soft-deleted
  rows live ≥ 24h before the sweeper hard-deletes them. Sweep also
  revokes the orphaned `agents.call:<user>:<name>` scope grant.
- **INV-181-12 (no-tier3-in-get-agents, Phase 2.0 scope)**: `GET
  /agents` does NOT include Tier 3 agents. Owner discovery (`GET
  /me/agents`) is a follow-up.
- **INV-181-13 (audit-emission, Phase 2.0 scope)**: Every successful
  `POST /agents/define`, `DELETE /agents/<id>`, `PATCH /agents/<id>`,
  and each cascade-delete emits exactly one audit event. Action
  codes: `agents.define`, `agents.delete`, `agents.patch`,
  `agents.cascade_delete`. Cascade events carry the revocation
  initiator as `actor`.
- **INV-181-14 (partial-unique-index, Phase 2.0 scope)**: The
  `consumer_agents` schema declares `CREATE UNIQUE INDEX
  consumer_agents_active_name_idx ON consumer_agents(user_id,
  agent_name) WHERE deleted_at IS NULL`. Soft-deleted rows coexist
  with fresh definitions of the same name.
- **INV-181-15 (user-id-slug-gate, Phase 2.0 scope)**: Any Tier 3
  operation verifies that `ctx.user_id` matches `[a-z][a-z0-9-]{2,63}`
  before any other processing; non-slug user_ids return 403
  `tier3_user_id_unsupported`.

### Persona-state observability (issue #204)

- **INV-204-1 (affect_update event type)**: `affect_update` is a
  top-level SSE event `type` discriminator, sibling to `worker_phase`
  / `tool_*` / `text` / `thinking` / `done`. Not a `worker_phase` sub-
  phase. INV-061's "BuildingPrompt is the FIRST event" property is
  scoped to `worker_phase` events only — `affect_update status="current"`
  may precede BuildingPrompt for persona-enabled agents.
- **INV-204-2 (per-turn emission)**: For agents with persona enabled
  on non-ephemeral sessions, `stream_turn` emits `status="current"`
  before any other SSE event on a successful or failed turn, and
  `status="scheduled"` after `update_after_turn` schedules the
  appraisal task (success path only — skipped on cancel / error
  before update_after_turn was reached). See contract
  `docs/contracts/issues/204.contract.md`.
- **INV-204-3 (emission suppression)**: Persona-disabled agents and
  ephemeral sessions emit ZERO `affect_update` events.
- **INV-204-6 / INV-204-7 (persona_state endpoint)**: New
  `GET /agents/{agent_id}/persona_state` gated on Heimdall scope
  `persona.read`. Route ordering: auth → Tier 3 short-circuit (404
  `persona_not_configured`) → Tier 1/2 existence (404
  `agent_not_available`) → persona-enabled check (404
  `persona_not_configured`) → snapshot (200).
- **INV-204-9 (read-only registry primitive)**: `PersonaRegistry.get_state`
  is mutex-free and never mutates `persona.emotions`. Eventual
  consistency under concurrent `_appraisal_wrapper` mutations.
- **INV-204-14 (replay participation)**: `affect_update` events flow
  through `_publish`, so SSE resume / replay handles them with no
  special case.

## Amendment — AwaitingLLMFirstToken heartbeat (issue #201, INV-201-1..7)

Adds a periodic SSE heartbeat event during the gap between
`BuildingPrompt` and `CallingLLM` so consumers can distinguish
"engine is thinking" from "engine is wedged" without out-of-band
server inspection. Filed by ratatoskr-dev; ships in v0.29.0.

- **INV-201-1 (new top-level event type)**: `awaiting_llm_first_token`
  is a new top-level SSE event type, sibling to `worker_phase` /
  `tool_*` / `text` / `thinking` / `debug` / `done` / `affect_update`.
  `_WORKER_PHASE_VOCAB` is NOT extended; INV-053 / INV-054 unchanged.
  Same precedent as #204's `affect_update`.

- **INV-201-2 (config-gated emission)**: Heartbeat emission requires
  `awaiting_llm_first_token_heartbeat_s > 0.0`. When the resolved
  value is `0.0`, the heartbeat task is never started and zero
  `awaiting_llm_first_token` events emit for the turn. When > 0.0,
  the task starts immediately after `_publish_phase("BuildingPrompt")`
  and emits an event every `interval` seconds until cancelled.

- **INV-201-3 (defense-in-depth cancellation)**: The heartbeat task
  is cancelled at three sites (idempotent via the `_cancel_heartbeat`
  helper): (a) immediately before `_publish_phase("CallingLLM")` on
  the engine-first-event path; (b) inside the `cancelled`/`error`
  handling that wraps `_handle_cancel` (covers stall + user-cancel
  paths); (c) in the outer `finally` block alongside
  `_clear_stall_timer`. After cancellation, no further
  `awaiting_llm_first_token` events emit.

- **INV-201-4 (wire shape)**: Payload is exactly `{type:
  "awaiting_llm_first_token", turn_id: <int>,
  elapsed_ms_since_building_prompt: <float>}` plus the composite `id:
  "<turn_id>:<seq>"` stamped by `_publish`. No additional fields.
  `elapsed_ms_since_building_prompt` is `(time.monotonic() -
  building_prompt_t) * 1000.0` where `building_prompt_t` is captured
  immediately before `BuildingPrompt` is published.

- **INV-201-5 (first-gap-only scope)**: Heartbeat is scoped to the
  FIRST `BuildingPrompt → CallingLLM` gap of the turn. Tool round-trip
  `CallingLLM` re-entries (INV-058) emit ZERO
  `awaiting_llm_first_token` events. Out-of-scope sub-phases
  (`AwaitingToolResult`, `AwaitingNextLLMCall`) would be separate
  follow-up features.

- **INV-201-6 (replay participation)**: Heartbeat events flow through
  `_publish → _replay_buffer + queue` per INV-060 — same replay
  semantics as worker_phase events. On `Last-Event-ID` reconnect,
  prior heartbeats replay identically.

- **INV-201-7 (config resolution precedence)**: Per-agent
  `agent.conversation.awaiting_llm_first_token_heartbeat_s` →
  `api_cfg.awaiting_llm_first_token_heartbeat_s` → built-in `5.0`.
  Negative values raise `ConfigurationError` at agent load; `0.0`
  is valid and means "disabled." Mirrors the `_resolve_stall_timeout_s`
  precedence pattern (INV-038).

### Mechanism note

The heartbeat task is a separate `asyncio.Task` (NOT `loop.call_later`,
because heartbeats repeat at an interval rather than fire once at a
timeout). An `asyncio.Queue` shared between the heartbeat task and the
generator carries events; the generator uses
`asyncio.wait(return_when=FIRST_COMPLETED)` to race the engine's
`__anext__` against the heartbeat queue's `get` ONLY during the first
iteration. After `CallingLLM` fires, the heartbeat task is cancelled
and subsequent iterations use the original non-race pattern.

### Storage extension

The `consumer_agents` table lives in `core/heimdall/storage/sqlite.py`
alongside `users` / `api_keys`. New protocol: `ConsumerAgentStore` in
`core/heimdall/storage/base.py`. Implementation:
`SqliteConsumerAgentStore`. Cascade and sweep hooks:
`core/heimdall/lifecycle.py`. Audit emission for the new action codes:
`core/heimdall/audit.emit_consumer_agent_event`.

### `AgentContext` discriminator

`core/conversation_api/service.py:AgentContext.kind` accepts
`"consumer_defined"` (in addition to `"foundational"` / `"ephemeral"`)
and carries `owner_user_id`, `agent_name`, `owner_key_hash` for Tier 3
rows.

## Amendment — Default agent (Lofn) routing (issue #182)

Adds Lofn as the default-resolved agent for `POST /sessions` when the
caller omits `agent_id`. Companion to ADR-0003 (no preference store).

### Request-model change

`CreateSessionRequest.agent_id` becomes `str | None = None` (was
required `str`). Absent / null / empty resolves to `"lofn"` via
`core/conversation_api/api.py:route_default_agent_id` (single guard
at the head of `create_session`). Non-empty strings pass through
unchanged.

### Invariants

- **INV-182-1**: `POST /sessions` with `body.agent_id` absent, null,
  or empty string resolves to `agent_id == "lofn"`. Whitespace-only
  strings (e.g., `" "`) pass through and surface 404 unknown-agent
  rather than rerouting.
- **INV-182-2**: Sessions with `agent_id == "lofn"` REQUIRE
  `end_user_id` as a non-empty string. Missing / null / empty returns
  422 with `error_code: end_user_id_required`. Same envelope shape
  as Tier 3's identical requirement (INV-181-17).
- **INV-182-3**: Default routing does NOT shadow explicit non-Lofn
  agent_ids. `agent_id: "mimir"` still resolves Mimir; the routing
  guard only fires when `body.agent_id` is None / empty.
- **INV-182-18**: Default routing is opt-in at session-create only.
  Existing non-Lofn sessions are never retroactively re-routed; turns
  on a Mimir session continue to dispatch to Mimir regardless of
  whether the per-message payload re-supplies `agent_id`.

### Function block

```contract
FN route_default_agent_id(agent_id: str | None) -> str
BRIEF: Single guard at the head of create_session. Resolves Lofn when caller omits the field.
PRE: [PRE-182-19 hard] agent_id is None or a string -- Pydantic-typed at caller
POST: [POST-182-23 return_value] returns "lofn" when agent_id is None or "" -- assert
POST: [POST-182-24 return_value] returns agent_id unchanged otherwise (including whitespace-only) -- assert
```

### Handoff: no surface added

Lofn ships with NO structured handoff signal — no marker convention
in her output, no post-turn parser, no `lofn.handoff_suggested`
audit event, no server-side session-switch, no Lofn-specific tool.
Specialist suggestions are prose-only ("Mimir would have better
visibility on that than I do"); the user (or, in multi-agent
contexts, the suggested agent via the existing agent_bus) acts on
the suggestion. The act of saying it IS the handoff — Worldtree
adds nothing on top.

The marker-text approach was specifically considered and ruled out as
a weak version of a tool the contract didn't want in the first place;
any future structured handoff API will be a real schema'd endpoint,
not text-parsed.

### No new storage, no new audit events

Lofn introduces zero net-new persistence surface. No table, no
column, no Mimir KB collection. No new audit-event types. Existing
session-create / session-revoke audit covers Lofn the same way it
covers Mimir / Forseti.