---
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.

## Memory-partition scope (#245 / ADR-0011)

`end_user_id` is the per-end-user memory partition key (distinct from `user_id`,
the API-key owner). It is REQUIRED at session-create for Lofn (Tier-1) and Tier-3
agents and must survive a store reload, because "remember me next session" is by
definition a reload. Memory partition resolution flows through ONE resolver that
cannot hand an authenticated session the shared `local_dev` partition.

- **INV-245-1 (end-user-id-durable)**: `end_user_id` is persisted as a `sessions`
  table column at create and rehydrated onto the `ConversationSession` on every
  cache-miss load (`get_session`). A session loaded from the store carries the
  same `end_user_id` it was created with. Pre-migration rows read as `None`.
- **INV-245-2 (end-user-id-threaded-all-tiers)**: the `POST /sessions` handler
  forwards `body.end_user_id` to `create_session` for EVERY agent, not only
  Tier-3. (The pre-fix `if tier3_agent_context is not None else None` conditional
  dropped it for Lofn despite the create gate requiring it.)
- **INV-245-3 (no-authenticated-local-dev)**: the two MEMORY partition sites —
  auto-recall (read) and the ContextPromotion producer (write) — resolve via
  `memory_scope_for_session`. An authenticated, memory-bearing session (one not
  carrying the explicit `local_dev` sentinel) NEVER resolves to `local_dev`; a
  missing `end_user_id` raises `MemoryScopeError`, and because both sites are
  best-effort (recall is fire-and-forget; the producer is `_run_promotion_safe`),
  the caller skips memory — it never silently writes to the shared partition.
- **INV-245-5 (persona-plane-corrected-by-persistence)**: the three PERSONA-plane
  sites (`inject_context`, `get_state`, `update_after_turn` — ADR-0008 mood/PAD/
  valence) keep their `session.end_user_id or "local_dev"` form but are on the
  main turn path where a raise would break the turn. They are corrected by
  INV-245-1/2: once `end_user_id` is persisted + threaded, the fallback yields a
  real partition for authenticated sessions and `local_dev` only for the explicit
  terminal path. Unifying the persona plane under the resolver (with main-path
  error semantics) is follow-up, tracked with the #246-adjacent hardening.
- **INV-245-4 (terminal-explicit-local-dev)**: the internal terminal transport
  creates its sessions with `end_user_id="local_dev"` explicitly. `local_dev` is
  reached only by this positive assertion, never by omission. (External API
  callers passing `local_dev` are still rejected per #216.)

```contract
FN memory_scope_for_session(session) -> MemoryScope
BRIEF: The single authority resolving a session to its memory partition scope.
       Returns a typed MemoryScope(scope_type, scope_id); scope_type ∈
       {local_dev, end_user, room, tenant} (only local_dev + end_user active in
       v1; room/tenant reserved for ADR-0010). Cannot yield local_dev for an
       authenticated session.
PRE: [PRE-001 soft] callers have already gated ephemeral / consumer_defined
     sessions out (those skip memory before resolution)
POST: [POST-001 return_value] end_user_id == "local_dev" -> MemoryScope("local_dev", "local_dev")
POST: [POST-002 return_value] end_user_id truthy and != "local_dev" -> MemoryScope("end_user", end_user_id)
POST: [POST-003 exception] end_user_id is None/empty -> raise MemoryScopeError (NEVER local_dev)
ERRORS:
  MemoryScopeError -> caller skips memory (best-effort) + emits an audit/log line; turn proceeds
STEPS:
  1. [setup] read euid = session.end_user_id
  2. [branch] euid == "local_dev" -> RETURN MemoryScope("local_dev", "local_dev") (terminal sentinel)
  3. [branch] euid truthy -> RETURN MemoryScope("end_user", euid)
  4. [error_handler] else (None/empty) -> RAISE MemoryScopeError (never silently local_dev)
TESTS:
  end_user_partition [happy,tracer]: session end_user_id="alice" -> MemoryScope("end_user","alice")
  terminal_local_dev [boundary]: session end_user_id="local_dev" -> MemoryScope("local_dev","local_dev")
  authenticated_none_raises [boundary]: foundational session end_user_id=None -> raises MemoryScopeError, NOT local_dev
  isolation_roundtrip [happy]: create_session(end_user_id="alice") write + clear cache + reload + recall isolates from a "bob" session; negative-assert no local_dev write
```

```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/echo/config.yaml`, `config/providers.yaml → echo_allowed_models`, or `config/defaults.yaml → echo.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 `echo_allowed_models` at session-create time. When omitted, server resolves to `echo.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)**: An Echo 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 | echo 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 `echo_allowed_models` |

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

**Startup failfast:** server refuses to start if `agents/echo/config.yaml` is missing/malformed OR `echo.default_model` is not in `echo_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 (echo-incompatible)**: A session cannot be both ephemeral (Echo, `kind: "ephemeral"`) AND Bifrost-bound. Session-create rejects with 422 `ephemeral_does_not_accept_bifrost`. Verifiable: `POST /sessions { agent_id: "echo", 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 Echo (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 (Echo). 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; AMENDED #188)**:
  PATCH rejects payloads carrying any of `persona`, `motivational`,
  `memory` even when set to `null`, returning `field_not_mutable`.
  **Amended by #188 (Phase 2.3):** `valence` was moved out of this
  `field_not_mutable` set — it now returns `layer_deferred` (see
  INV-188-1), because valence is a not-yet-shipped layer, not a frozen
  trait. `memory` is rejected wholesale (see INV-188-2).
- **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 — Suspended-tier license-state gate (issue #174, INV-174-1..9)

Adds a `suspended` tier with empty scope set to drive license-expiry
transitions without destroying user state. Endpoint
`POST /admin/users/{user_id}/tier` mutates the tier; the
`_http_exception_handler` rewrites `AUTH_SCOPE_DENIED` →
`USER_SUSPENDED` for any 403 raised against a non-anonymous caller with
an empty scope-set (the suspended-tier defining property). Ships in
v0.29.1.

- **INV-174-1 (closed tier vocabulary)**: `POST /admin/users/{user_id}/tier`
  validates `body.tier` against the hard-coded set `{anonymous, user, free,
  pro, admin, suspended}`. Out-of-set values return 422 `invalid_tier`.
  Vocabulary is NOT derived from `policies.yaml` at runtime — a typo in
  YAML must not silently expand the accepted set.

- **INV-174-2 (admin-only mutation)**: endpoint requires
  `admin.users.write.tier_change` scope. Listed explicitly in admin
  tier's scope set in `policies.yaml` for grep-discoverability (admin
  also carries `*` umbrella).

- **INV-174-3 (tier mutation primitive)**:
  `UserStore.update_user_tier(user_id, new_tier) -> User` is the storage
  primitive. Raises `LookupError` for unknown user_id (endpoint converts
  to 404 `user_not_found`).

- **INV-174-4 (suspended scope-set is exactly empty)**:
  `policies.yaml.tiers["suspended"].scopes == []`. The empty set is what
  makes the auth-denial work for free; the
  `_http_exception_handler` rewrite uses
  `ctx.user_id != "anonymous" and not ctx.scopes` as the
  suspended-detection heuristic since `SecurityContext` deliberately
  excludes `tier` (per `core/integration/types.py:64`).

- **INV-174-5 (uniform suspended error code via exception handler)**:
  The `_http_exception_handler` (registered for `StarletteHTTPException`)
  intercepts every 403 with `error_code: auth_scope_denied`; if the
  request's stashed `SecurityContext` has an empty scope-set (and
  non-anonymous user_id), it rewrites the detail to
  `{error_code: "user_suspended", message: "Account is suspended."}`.
  Single seam — covers every existing and future scope-deny site
  without per-endpoint refactor. The ctx is stashed by
  `get_security_context` on `request.state.security_context`.

- **INV-174-6 (/me carve-out)**: `/me` does NOT call `authorize()` and
  therefore never raises `AUTH_SCOPE_DENIED`. Suspended users with
  empty scopes reach the /me handler normally and see
  `{user_id, tier: "suspended", scopes: [], ...}`. Adding a scope check
  to /me without preserving the suspended-tier visibility would be a
  contract violation — the carve-out is structural, not coded.

- **INV-174-7 (audit emission)**: every tier-change attempt emits
  `conversation_api:admin:user:tier_changed` via `_audit_admin_action`
  with `actor_user_id`, `target_user_id`, `outcome ∈
  {success, denied}`, and `extra = {from_tier, to_tier, reason}` for
  successes; `extra = {reason: <reason_code>}` for denials
  (`invalid_tier`, `user_not_found`).

- **INV-174-8 (reversibility via audit replay)**: the user record does
  NOT carry a `previous_tier` column. Restoration of a suspended user
  requires reading the audit log to find the most recent
  `tier_changed` event with `to_tier="suspended"` and replaying its
  `from_tier` as the new target. Operational responsibility of SEA's
  billing integration; Worldtree provides only the read (audit log) and
  write (endpoint) surfaces.

- **INV-174-9 (no cross-tier session invalidation)**: a tier change for
  a user with active SSE turns in flight does NOT cancel those turns.
  The next request after the tier change picks up the new scope-set;
  in-flight streams complete under the old tier. If SEA needs
  immediate-cutoff semantics, that requires `disable_user`-style
  hard-revoke, not a tier change.

## 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.

## Amendment — Tier 3 motivational layer (issue #187, Phase 2.2)

Activates the `motivational` layer field on `POST /agents/define`, narrowing the
Phase 2.0 `layer_deferred` rejection (INV-181-3) to `valence` only. Full FN-level
spec at `docs/contracts/issues/187.contract.md`.

- **INV-187-1 (motivational-activated)**: `POST /agents/define` accepts a non-null
  `motivational` object `{goals, fears}`; `_tier3_validate_layer_fields` rejects
  only `valence` now. (Persona + memory were activated in Phase 2.1 / #189.)
- **INV-187-2 (define-validation)**: `validate_motivational_define_payload` enforces
  the documented 422 codes — `motivational_id_collision` (case-sensitive, across
  goals AND fears), `motivational_goal_invalid_type`,
  `motivational_salience_out_of_range`, `motivational_description_too_short`
  (< 20 chars after strip), `motivational_missing_required_field`. Unknown top-level
  OR nested (per goal/fear) keys → `validation_failed` (sub-models extra-forbid).
  Stricter than the Tier 1 `validate_motivation` (which only warns on short text).
- **INV-187-3 (per-agent-scope)**: motivational is per-agent, NOT
  per-(agent, end_user) — stored once on the row, identical across all end-users.
- **INV-187-4 (immutable-in-patch)**: `PATCH` with `motivational` → 422
  `field_not_mutable` (already covered by INV-181-6's `_IMMUTABLE_FIELDS` gate).
- **INV-187-5 (tier3-render-bridge)**: Tier 3 agents are NOT registered with the
  `persona_registry`; the stored config rides on the per-session `AgentContext`
  (`motivational_config`) and is rendered into the prompt per-turn in `stream_turn`
  via `_append_motivational_context_section`, before the memory-context section.
- **INV-187-6 (fear-signal-shape)**: fears carry `trigger_signals`; goals carry
  `positive_signals` + `negative_signals` (matches the `GoalConfig`/`FearConfig`
  substrate).
- **INV-187-7 (tier-uniformity)**: the render reuses `core.persona.goals.load_goals`
  + `render_motivational_context`, so a Tier 3 motivational config produces a
  byte-identical block to an equivalent Tier 1 `motivation.yaml`.
- **INV-187-8 (storage)**: persisted in `consumer_agents.tier3_layers_json` under
  the `"motivational"` key; round-trips via `ConsumerAgent.motivational`; null/omitted
  → `None` (no fabricated defaults; no migration).

### Audit

`agents.define` audit `extra` gains `presence_motivational: bool` alongside
`presence_persona` / `presence_memory`.

## Amendment — Tier 3 PATCH mutability policy (issue #188, Phase 2.3)

Settles which Tier 3 agent fields are editable post-define. #197 deleted the
STM tier between this issue's filing (2026-05-19) and its implementation, so the
"mutable memory dials" the original issue envisioned no longer exist; the policy
collapses to: `system_prompt` + `model` mutable, everything else fixed, with
`valence` distinguished from the immutable traits by error code. No new
endpoint, no new storage, no new invariant philosophy — a clarification +
error-code alignment + audit enrichment over the Phase 2.0 PATCH baseline.

- **INV-188-1 (valence-deferred-in-patch)**: `PATCH /agents/<id>` carrying a
  `valence` key (any value, including `null`) → 422 `layer_deferred` with
  `field: "valence"`, matching define-time (INV-181-3). Rationale: valence is
  a layer that does not exist yet, not a real-but-frozen trait; `layer_deferred`
  is the truthful reason and gives consumers ONE code for "valence unavailable"
  across both define and PATCH. The check precedes the DB lookup (INV-181-5/6
  ordering), so a `valence` PATCH against a missing agent still 422s, not 404s.
- **INV-188-2 (memory-wholesale-immutable-in-patch)**: `PATCH` carrying a
  `memory` key → 422 `field_not_mutable` with `field: "memory"`, rejected at the
  WHOLE-field level. No sub-field carve-out exists: `stm_capacity` /
  `stm_token_budget` are deprecated no-ops post-#197, `allows_world_scope` is
  create-time-only (toggling it after memory is written breaks scope-visibility
  invariants — memory scope policy must be fixed before any memory is written),
  and `embedder_version` is library-pinned. A real LTM tuning dial would warrant
  a deliberate per-sub-field PATCH contract at that time; pre-splitting for dead
  fields is not done. NOTE the deliberate define/PATCH asymmetry: `define`
  accept-and-ignores deprecated `stm_*` (201 + DeprecationWarning per
  INV-197-19), but `PATCH memory:{...}` rejects wholesale (422). Acceptable
  transitional artifact; disappears when the shims are removed.
- **INV-188-3 (patch-audit-before-after)**: a successful `agents.patch` audit
  event's `extra.changes` records before/after for each mutated field —
  `model: {before, after}` (literal values; allowlist enum, not PII) and
  `system_prompt: {before_bytes, after_bytes}` (byte-length only; raw prompt
  content is excluded as potential PII, consistent with `emit_consumer_agent_event`'s
  exclusion rule). `changes` contains only keys for fields actually present in
  the PATCH payload. `patched_fields` (the Phase 2.0 name list) is retained.
- **INV-188-4 (mutable-surface-unchanged)**: the mutable surface stays exactly
  `system_prompt` + `model` (per INV-181 Phase 2.0). PATCH re-enforces the
  define-time `system_prompt` byte-cap and `model` allowlist. #188 does NOT add
  model-swap capability/context-window validation — that gap (a swap to a
  smaller-context or non-tool model with no re-check of the existing prompt) is
  tracked as a separate follow-up (#219), not folded here.

## Amendment — model-assignment advisory warnings (issue #219)

`POST /agents/define` and `PATCH /agents/<id>` attach a best-effort, **non-
blocking** `warnings` array to their 2xx response when the assigned `model`
carries metadata risk (smaller context window, unknown window, or an explicit
capability downgrade). This is advisory-only by deliberate design: hard
rejection was rejected (Heid panel + operator, 2026-05-29) because model
metadata coverage is partial (`context_window` is 0/unknown for several
allowlisted models; `supports_tools` defaults true), the stored `system_prompt`
cap is bytes not tokens, Tier 3 agent rows store no tool/modality usage (tools
arrive per-session via Bifrost, so any capability concern is inherently
conditional), and runtime already classifies the real failure as
`CONTEXT_OVERFLOW`. The warning is a receipt-note for the owner who just made a
deliberate change, not a correctness gate.

- **INV-219-1 (advisory-not-blocking)**: neither define nor PATCH ever rejects
  on context-window or capability grounds. The allowlist check
  (`model_not_available`) and `system_prompt` byte-cap are the only model-
  related *rejections*; everything in #219 is a warning on an otherwise-2xx
  response. Correctness for over-budget prompts remains the runtime
  `CONTEXT_OVERFLOW` guard.
- **INV-219-2 (bounded-warning-codes)**: the closed code set is exactly —
  `model_context_window_unknown` (severity `info`): the assigned model's
  registry `context_window` is `0`/absent; `model_context_window_smaller`
  (severity `warning`): prior and new model both have known windows and
  new < prior (`details: {before, after}`); `model_capability_downgrade`
  (severity `warning`): the new model EXPLICITLY drops a capability the prior
  model advertised — `supports_tools`, `vision`, or `audio` (`details:
  {dropped: [...]}`). No token-aware "prompt won't fit" code — deferred until
  tokenizer-aware estimation exists; messages never claim a hard fit/failure.
- **INV-219-3 (when-evaluated, resulting-pair)**: warnings are computed
  whenever a model is *assigned*. At define, always (prior = None → only
  `model_context_window_unknown` can apply, since the comparative codes need a
  prior). At PATCH, only when the payload carries a `model` key whose value
  differs from the stored model (prior = stored model); a PATCH without `model`
  (e.g. `system_prompt`-only) emits no model warnings. The comparison is always
  against the *resulting* model.
- **INV-219-4 (capability-downgrade)**: a `model_capability_downgrade` fires
  only when BOTH prior and new models resolve to registry `ModelInfo` AND the
  new model's *effective* capability flags lack one the prior advertised
  (`supports_tools`, `vision`, or `audio`). The "both resolve" guard is the
  false-positive defense — an unresolvable model on either side yields no
  downgrade claim. Beyond that, comparison uses the registry's **effective**
  flags, which is asymmetric by capability because the data model collapses
  absent-to-default and does not preserve a "was this declared?" bit:
  - `supports_tools` defaults **true** (`ModelInfo` / `_build_model_info`), so
    a tools-drop requires the new catalog entry to set `supports_tools: false`
    *explicitly* — omission never triggers it.
  - `vision` / `audio` default **false** (`ModelCapabilities`), so a drop is
    detected whenever the prior advertised the capability and the new model does
    not carry it — whether the new entry says `false` explicitly OR omits it.
    This is the deliberate conservative reading: an undeclared modality is
    treated as unsupported. (A vision-capable model with sloppy metadata that
    omits its `vision` flag would thus be reported as a downgrade; the remedy is
    to declare the flag in the catalog, not to suppress the advisory.)

  Message phrasing is conditional ("if your sessions rely on these, e.g. Bifrost
  tools, they may be rejected") — the agent row does not record whether tools or
  modalities are actually used, so every capability warning is advisory by
  nature.
- **INV-219-5 (inline-response-shape)**: the `warnings` array is added inline to
  the define (201) and PATCH (200) response bodies — the existing flat
  `ConsumerAgentResponse` dict gains a `warnings` key (always present, `[]` when
  none). It is NOT added to the shared `ConsumerAgentResponse` pydantic model
  nor to `GET /agents/<id>` — only the two mutation handlers merge it into their
  returned dict, keeping persisted fields and the read path unchanged. Each
  entry is `{code, severity, message, details}`.
- **INV-219-6 (single-helper)**: a single pure helper
  `compute_model_swap_warnings(*, prior_model: str | None, new_model: str,
  registry)` is the only source of warning logic; both define and PATCH call
  it. It tolerates unresolvable specs / `None` `ModelInfo` / `context_window`
  `0` by treating them as "unknown" (emitting the unknown-window info code where
  applicable, never raising). Metadata improvements over time sharpen the
  warnings with no API or signature change.
