feat(sse,tui): bump spec pin to v0.29.0 + AwaitingLlmFirstToken (v0.14.0)

Spec pin moved da93ca7 (v0.28.0) → 562001a (v0.29.0); vendored
conversation-api-spec.md + conversation_api.contract.md re-snapshotted.
The only material delta is Worldtree #201's awaiting_llm_first_token
SSE heartbeat — a top-level event (NOT a worker_phase extension, per
INV-053's three-field stability) that fires at a configurable interval
(default 5s) during the BuildingPrompt → CallingLLM gap.

Wire layer (sse_client.py):
- New `AwaitingLlmFirstToken` dataclass: sse_id / turn_id /
  elapsed_ms_since_building_prompt (server-authoritative monotonic)
- Added to Event union + _envelope_for_type dispatch branch
- Without this, ratatoskr would crash on any slow-first-token turn
  from a v0.29.0 server (unknown SSE event type → ValueError)

TUI layer (tui.py):
- Audit pipeline: per-event debug-pane line with elapsed in seconds
- Live transcript indicator: first heartbeat mounts a Static
  ("awaiting first token · 5.0s"); subsequent heartbeats update it
  in place; any non-heartbeat event removes it (the gap closed)
- Turn-summary line now carries heartbeat count
- Indicator demoted via .awaiting-label CSS so it reads as ambient
  progress, not content

Tests: 2 wire-layer (single + monotonic sequence) + 3 presenter
(audit line shape, single-mount semantic, indicator removal on gap
close). Suite: 318 passing.

docs/SPEC-PIN.md

@@ -7,17 +7,18 @@ documents the pin, the vendored artifacts, and the bump procedure.
 
 | Field | Value |
 |---|---|
-| Worldtree git SHA | `da93ca7cf613f1dc229a7a44d07fc1d7efc78e25` |
-| Worldtree HEAD message | `feat(#204): v0.28.0 — persona-state observability surface` |
-| Pinned on | 2026-05-25 |
-| Pinned by | ratatoskr-dev (bump for #204 affect_update SSE) |
-| Worldtree version at pin | `v0.28.0` |
+| Worldtree git SHA | `562001af28d752c3a60d449c7ddd09f44fa9dc9a` |
+| Worldtree HEAD message | `feat(#201): v0.29.0 — awaiting_llm_first_token SSE heartbeat` |
+| Pinned on | 2026-05-26 |
+| Pinned by | ratatoskr-dev (bump for #201 awaiting_llm_first_token SSE) |
+| Worldtree version at pin | `v0.29.0` |
 
 ## Pin history
 
 | Date | SHA | Version | Notable deltas consumed |
 |---|---|---|---|
-| 2026-05-25 | `da93ca7` | v0.28.0 | #204 — new SSE event `affect_update` (current/scheduled), new endpoint `GET /agents/{id}/persona_state` (not yet consumed), auth-model doc edits |
+| 2026-05-26 | `562001a` | v0.29.0 | #201 — new SSE event `awaiting_llm_first_token` (heartbeat during BuildingPrompt → CallingLLM gap, default 5s interval) |
+| 2026-05-25 | `da93ca7` | v0.28.0 | #204 — new SSE event `affect_update` (current/scheduled), new endpoint `GET /agents/{id}/persona_state`, auth-model doc edits |
 | 2026-05-20 | `55101e9` | v0.19.0 | initial scaffold pin |
 
 ## Vendored artifacts

docs/conversation-api-spec.md

@@ -1980,6 +1980,26 @@ Emitted after the post-turn appraisal task has been scheduled (per #177 Phase A'
 
 Bootstrap reads available via `GET /agents/{agent_id}/persona_state` (same `snapshot` shape, requires `persona.read` scope).
 
+### awaiting_llm_first_token
+
+Periodic heartbeat event (issue #201) emitted at a configurable interval during the gap between `worker_phase: phase="BuildingPrompt"` and `worker_phase: phase="CallingLLM"`. Solves the legitimate-slow first-token visibility gap: consumer TUIs can render a "thinking for Ns…" timer rather than a frozen line during heavy-CoT prompt warmup.
+
+```json
+{
+  "type": "awaiting_llm_first_token",
+  "turn_id": 42,
+  "elapsed_ms_since_building_prompt": 5012.3
+}
+```
+
+`elapsed_ms_since_building_prompt` is the server-authoritative wall-clock milliseconds since `BuildingPrompt` was emitted. Independent of network latency or clock skew.
+
+Heartbeats stop the moment the engine produces its first event (the `CallingLLM` marker). They do NOT re-fire during tool-roundtrip `CallingLLM` re-entries — the heartbeat is scoped to the FIRST `BuildingPrompt → CallingLLM` gap only.
+
+**Configuration:** `conversation_api.awaiting_llm_first_token_heartbeat_s` (default `5.0`). Per-agent override via `agent.conversation.awaiting_llm_first_token_heartbeat_s`. Value `0.0` disables emission entirely.
+
+Cancellation paths (stall watchdog, user-cancel) also stop the heartbeat — no `awaiting_llm_first_token` event appears after the terminal `cancelled` event.
+
 ### thinking
 
 Incremental reasoning/thinking content (from thinking-enabled models).

docs/conversation_api.contract.md

@@ -1881,6 +1881,73 @@ SQLite `consumer_agents` table.
   through `_publish`, so SSE resume / replay handles them with no
   special case.
 
+## Amendment — AwaitingLLMFirstToken heartbeat (issue #201, INV-201-1..7)
+
+Adds a periodic SSE heartbeat event during the gap between
+`BuildingPrompt` and `CallingLLM` so consumers can distinguish
+"engine is thinking" from "engine is wedged" without out-of-band
+server inspection. Filed by ratatoskr-dev; ships in v0.29.0.
+
+- **INV-201-1 (new top-level event type)**: `awaiting_llm_first_token`
+  is a new top-level SSE event type, sibling to `worker_phase` /
+  `tool_*` / `text` / `thinking` / `debug` / `done` / `affect_update`.
+  `_WORKER_PHASE_VOCAB` is NOT extended; INV-053 / INV-054 unchanged.
+  Same precedent as #204's `affect_update`.
+
+- **INV-201-2 (config-gated emission)**: Heartbeat emission requires
+  `awaiting_llm_first_token_heartbeat_s > 0.0`. When the resolved
+  value is `0.0`, the heartbeat task is never started and zero
+  `awaiting_llm_first_token` events emit for the turn. When > 0.0,
+  the task starts immediately after `_publish_phase("BuildingPrompt")`
+  and emits an event every `interval` seconds until cancelled.
+
+- **INV-201-3 (defense-in-depth cancellation)**: The heartbeat task
+  is cancelled at three sites (idempotent via the `_cancel_heartbeat`
+  helper): (a) immediately before `_publish_phase("CallingLLM")` on
+  the engine-first-event path; (b) inside the `cancelled`/`error`
+  handling that wraps `_handle_cancel` (covers stall + user-cancel
+  paths); (c) in the outer `finally` block alongside
+  `_clear_stall_timer`. After cancellation, no further
+  `awaiting_llm_first_token` events emit.
+
+- **INV-201-4 (wire shape)**: Payload is exactly `{type:
+  "awaiting_llm_first_token", turn_id: <int>,
+  elapsed_ms_since_building_prompt: <float>}` plus the composite `id:
+  "<turn_id>:<seq>"` stamped by `_publish`. No additional fields.
+  `elapsed_ms_since_building_prompt` is `(time.monotonic() -
+  building_prompt_t) * 1000.0` where `building_prompt_t` is captured
+  immediately before `BuildingPrompt` is published.
+
+- **INV-201-5 (first-gap-only scope)**: Heartbeat is scoped to the
+  FIRST `BuildingPrompt → CallingLLM` gap of the turn. Tool round-trip
+  `CallingLLM` re-entries (INV-058) emit ZERO
+  `awaiting_llm_first_token` events. Out-of-scope sub-phases
+  (`AwaitingToolResult`, `AwaitingNextLLMCall`) would be separate
+  follow-up features.
+
+- **INV-201-6 (replay participation)**: Heartbeat events flow through
+  `_publish → _replay_buffer + queue` per INV-060 — same replay
+  semantics as worker_phase events. On `Last-Event-ID` reconnect,
+  prior heartbeats replay identically.
+
+- **INV-201-7 (config resolution precedence)**: Per-agent
+  `agent.conversation.awaiting_llm_first_token_heartbeat_s` →
+  `api_cfg.awaiting_llm_first_token_heartbeat_s` → built-in `5.0`.
+  Negative values raise `ConfigurationError` at agent load; `0.0`
+  is valid and means "disabled." Mirrors the `_resolve_stall_timeout_s`
+  precedence pattern (INV-038).
+
+### Mechanism note
+
+The heartbeat task is a separate `asyncio.Task` (NOT `loop.call_later`,
+because heartbeats repeat at an interval rather than fire once at a
+timeout). An `asyncio.Queue` shared between the heartbeat task and the
+generator carries events; the generator uses
+`asyncio.wait(return_when=FIRST_COMPLETED)` to race the engine's
+`__anext__` against the heartbeat queue's `get` ONLY during the first
+iteration. After `CallingLLM` fires, the heartbeat task is cancelled
+and subsequent iterations use the original non-race pattern.
+
 ### Storage extension
 
 The `consumer_agents` table lives in `core/heimdall/storage/sqlite.py`

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "ratatoskr"
-version = "0.11.0"
+version = "0.14.0"
 description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard"
 readme = "README.md"
 requires-python = ">=3.12"
@@ -42,9 +42,9 @@ Repository = "https://gitea.phasefinal.com/vh/ratatoskr"
 # Ratatoskr is built against Worldtree at this commit; the vendored
 # spec snapshot in docs/ reflects that SHA.
 [tool.ratatoskr.spec-pin]
-worldtree-spec-rev = "da93ca7cf613f1dc229a7a44d07fc1d7efc78e25"
-worldtree-version = "v0.28.0"
-pinned-on = "2026-05-25"
+worldtree-spec-rev = "562001af28d752c3a60d449c7ddd09f44fa9dc9a"
+worldtree-version = "v0.29.0"
+pinned-on = "2026-05-26"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/ratatoskr"]

src/ratatoskr/sessions.py

@@ -89,6 +89,46 @@ class SessionApiFailed(Exception):
         self.body = body
 
 
+# Worldtree #204 / v0.28.0 — persona_state endpoint failure modes.
+class PersonaNotConfigured(Exception):
+    """Raised on HTTP 404 `persona_not_configured` from GET persona_state.
+
+    Agent exists but has no persona surface: persona-disabled Tier 1/2
+    agents (e.g. `domari`, `muninn`) and all Tier 3 consumer-defined
+    agents (Phase 2.0). Distinct from `AgentNotAvailable` which means the
+    agent_id is unknown entirely.
+    """
+
+    def __init__(self, *, agent_id: str) -> None:
+        super().__init__(f"persona not configured for agent_id: {agent_id!r}")
+        self.agent_id = agent_id
+
+
+class AgentNotAvailable(Exception):
+    """Raised on HTTP 404 `agent_not_available` from GET persona_state.
+
+    The agent_id is unknown to the server. Distinct from
+    `PersonaNotConfigured` (agent exists but has no persona).
+    """
+
+    def __init__(self, *, agent_id: str) -> None:
+        super().__init__(f"agent not available: {agent_id!r}")
+        self.agent_id = agent_id
+
+
+class AuthScopeDenied(Exception):
+    """Raised on HTTP 403 `auth_scope_denied` from a Heimdall-scoped endpoint.
+
+    The API key lacks the required scope (e.g. `persona.read` for
+    GET /agents/{id}/persona_state). User-tier keys carry `persona.read`
+    by default; this surfaces when a narrower key is in use.
+    """
+
+    def __init__(self, *, scope: str) -> None:
+        super().__init__(f"auth scope denied: required={scope!r}")
+        self.scope = scope
+
+
 async def list_sessions(
     client: httpx.AsyncClient,
     *,
@@ -200,3 +240,45 @@ async def list_agents(client: httpx.AsyncClient) -> list[AgentInfo]:
         )
         for item in body
     ]
+
+
+async def get_persona_state(
+    client: httpx.AsyncClient, agent_id: str
+) -> dict[str, Any]:
+    """GET /agents/{agent_id}/persona_state — fetch current persona snapshot.
+
+    Worldtree #204 / v0.28.0. Returns the same `snapshot` dict shape as the
+    `affect_update` SSE event's `status="current"` emission: pad,
+    dominant_emotion, emotions_active, baseline_pad, mood_drift,
+    last_updated_at. Bootstrap read for clients that want to populate a
+    persona pane on session-open without waiting for turn-1's `affect_update`.
+
+    Auth: requires Heimdall `persona.read` scope (user-tier default).
+
+    Failure modes (mapped to typed exceptions per the spec error_codes):
+      - 404 `persona_not_configured` → PersonaNotConfigured (persona-disabled
+        agents: domari / muninn, and all Tier 3 in Phase 2.0)
+      - 404 `agent_not_available` → AgentNotAvailable (unknown agent_id)
+      - 403 `auth_scope_denied` → AuthScopeDenied (key lacks persona.read)
+      - any other non-2xx → SessionApiFailed (preserves the broader-error
+        precedent from list_agents / list_sessions / create_session)
+    """
+    assert client is not None
+    assert agent_id and isinstance(agent_id, str)
+
+    resp = await client.get(f"/agents/{agent_id}/persona_state")
+    if resp.status_code == 200:
+        return resp.json()
+    # Discriminate the 4xx error_code sub-codes; everything else falls through.
+    try:
+        err = resp.json()
+        error_code = err.get("error_code") if isinstance(err, dict) else None
+    except ValueError:
+        error_code = None
+    if resp.status_code == 404 and error_code == "persona_not_configured":
+        raise PersonaNotConfigured(agent_id=agent_id)
+    if resp.status_code == 404 and error_code == "agent_not_available":
+        raise AgentNotAvailable(agent_id=agent_id)
+    if resp.status_code == 403 and error_code == "auth_scope_denied":
+        raise AuthScopeDenied(scope="persona.read")
+    raise SessionApiFailed(status=resp.status_code, body=resp.content)

src/ratatoskr/sse_client.py

@@ -111,6 +111,31 @@ class Cancelled:
     partial_message_id: int | None
 
 
+@dataclass(frozen=True)
+class AwaitingLlmFirstToken:
+    """SSE event `awaiting_llm_first_token`: heartbeat during slow first-token.
+
+    Fires at the configured interval (default 5s) during the gap between
+    `worker_phase` phase=BuildingPrompt and phase=CallingLLM. Lets clients
+    render a live "thinking for Ns…" indicator instead of a frozen line
+    during legitimate-slow first-token latency. Stops the moment CallingLLM
+    fires (defense-in-depth at three sites); no heartbeat after Cancelled
+    or stalled terminal events. Tool round-trip re-entries do NOT re-fire
+    heartbeats — INV-201-5 scopes the mechanism to the FIRST gap only.
+
+    `elapsed_ms_since_building_prompt` is server-authoritative
+    `time.monotonic()`-based — independent of network latency or clock
+    skew, monotonically increasing across the heartbeat sequence.
+
+    See docs/conversation-api-spec.md § awaiting_llm_first_token
+    (Worldtree #201, v0.29.0).
+    """
+
+    sse_id: SseId
+    turn_id: int
+    elapsed_ms_since_building_prompt: float
+
+
 @dataclass(frozen=True)
 class AffectUpdate:
     """SSE event `affect_update`: persona-state observability snapshot.
@@ -146,6 +171,7 @@ Event = (
     | Error
     | Cancelled
     | AffectUpdate
+    | AwaitingLlmFirstToken
 )
 
 
@@ -309,6 +335,15 @@ def _envelope_for_type(body: dict[str, Any], sse_id: SseId) -> Event:
             reason=body.get("reason"),
             partial_message_id=body.get("partial_message_id"),
         )
+    if t == "awaiting_llm_first_token":
+        # Worldtree #201 / v0.29.0: top-level heartbeat during BuildingPrompt
+        # → CallingLLM gap. Lets clients render live elapsed-time indicators
+        # instead of frozen lines on legitimate-slow first-token latency.
+        return AwaitingLlmFirstToken(
+            sse_id=sse_id,
+            turn_id=body["turn_id"],
+            elapsed_ms_since_building_prompt=body["elapsed_ms_since_building_prompt"],
+        )
     if t == "affect_update":
         # Worldtree #204 / v0.28.0: persona-state observability event.
         # status="current" carries full snapshot at turn start;

src/ratatoskr/tui.py

@@ -35,13 +35,18 @@ from textual.widgets import (
 from ratatoskr.cli import USER_AGENT, ParsedArgs, _format_duration_ms, _format_usage
 from ratatoskr.sessions import (
     AgentInfo,
+    AgentNotAvailable,
     AgentNotFound,
+    AuthScopeDenied,
+    PersonaNotConfigured,
     SessionApiFailed,
     create_session,
+    get_persona_state,
     list_agents,
 )
 from ratatoskr.sse_client import (
     AffectUpdate,
+    AwaitingLlmFirstToken,
     CancelAlreadyCompleted,
     CancelFailed,
     Cancelled,
@@ -187,6 +192,79 @@ def _ts() -> str:
     return now.strftime("%H:%M:%S") + f".{now.microsecond // 1000:03d}"
 
 
+def _format_persona_header(snapshot: dict) -> str:
+    """One-line persona summary for the sticky header widget.
+
+    Shape: `agent_id · dominant_emotion · pad(P, A, D) · N emotions active`.
+    Built for at-a-glance scanning above the chat area — concise enough to
+    fit one terminal row. Full detail lives in the Persona TabPane.
+    """
+    pad = snapshot.get("pad") or {}
+    emotions = snapshot.get("emotions_active") or []
+    dom = snapshot.get("dominant_emotion") or "?"
+    pieces = [
+        f"{snapshot.get('agent_id', '?')}",
+        f"{dom}",
+        f"pad({pad.get('pleasure', '?')}, {pad.get('arousal', '?')}, "
+        f"{pad.get('dominance', '?')})",
+    ]
+    if emotions:
+        pieces.append(f"{len(emotions)} emotion{'s' if len(emotions) != 1 else ''} active")
+    return " · ".join(pieces)
+
+
+def _format_persona_detail(snapshot: dict) -> str:
+    """Multi-line persona detail for the Persona TabPane.
+
+    Renders the full v0.28.0 snapshot shape: dominant_emotion, PAD with
+    baseline comparison, mood_drift deltas, active emotions list with
+    intensity + decay, last_updated_at footer.
+    """
+    pad = snapshot.get("pad") or {}
+    baseline = snapshot.get("baseline_pad") or {}
+    drift = snapshot.get("mood_drift") or {}
+    emotions = snapshot.get("emotions_active") or []
+    lines: list[str] = []
+    agent = snapshot.get("agent_id", "?")
+    lines.append(f"Persona snapshot · {agent}")
+    lines.append("")
+    lines.append(f"Dominant emotion: {snapshot.get('dominant_emotion', '?')}")
+    lines.append("")
+    lines.append("PAD")
+    for axis in ("pleasure", "arousal", "dominance"):
+        v = pad.get(axis, "?")
+        b = baseline.get(axis, "?")
+        delta = ""
+        if isinstance(v, (int, float)) and isinstance(b, (int, float)):
+            delta = f" (Δ {v - b:+.2f})"
+        lines.append(f"  {axis:<10} {v}  baseline {b}{delta}")
+    lines.append("")
+    if drift:
+        lines.append("Mood drift")
+        for key in ("valence_delta", "arousal_delta"):
+            if key in drift:
+                v = drift[key]
+                lines.append(f"  {key:<16} {v:+}" if isinstance(v, (int, float))
+                             else f"  {key:<16} {v}")
+        lines.append("")
+    lines.append(f"Active emotions ({len(emotions)})")
+    for e in emotions:
+        et = e.get("type", "?")
+        ei = e.get("intensity", "?")
+        decay = e.get("decay_remaining_s")
+        decay_str = (
+            f"  decay {decay / 60:.1f}m" if isinstance(decay, (int, float)) else ""
+        )
+        lines.append(f"  {et:<20} intensity {ei}{decay_str}")
+    if not emotions:
+        lines.append("  (none)")
+    last = snapshot.get("last_updated_at")
+    if last:
+        lines.append("")
+        lines.append(f"Last updated: {last}")
+    return "\n".join(lines)
+
+
 def _audit_line(event: Event) -> str:
     """One-line wire-level audit summary for the debug pane.
 
@@ -218,6 +296,12 @@ def _audit_line(event: Event) -> str:
         )
     elif isinstance(event, Cancelled):
         detail = f"turn_id={event.turn_id} reason={event.reason!r}"
+    elif isinstance(event, AwaitingLlmFirstToken):
+        # Worldtree #201 / v0.29.0. Compact: turn_id + elapsed in seconds
+        # (the heartbeat itself fires every 5s by default; seconds-rounding
+        # is the natural unit for operator scanning).
+        secs = event.elapsed_ms_since_building_prompt / 1000.0
+        detail = f"turn_id={event.turn_id} elapsed={secs:.1f}s"
     elif isinstance(event, AffectUpdate):
         # Worldtree #204 / v0.28.0. status="current" carries the full
         # snapshot; surface dominant_emotion + PAD inline so the operator
@@ -270,6 +354,13 @@ class TuiPresenterState:
     thinking_delta_count: int = 0
     thinking_byte_count: int = 0
     turn_start_ts: float = 0.0
+    # v0.14.0: Worldtree #201 heartbeat surface. First
+    # `awaiting_llm_first_token` mounts a Static; subsequent heartbeats
+    # update it in place; any non-heartbeat event clears it (the gap
+    # closed). awaiting_widget is the Static reference (None when
+    # closed); heartbeat_count tracks emissions for the turn-summary.
+    awaiting_widget: object = None
+    heartbeat_count: int = 0
 
     def render(
         self,
@@ -280,6 +371,7 @@ class TuiPresenterState:
         debug_log: RichLog,
         thinking_log: RichLog,
         raw: bool,
+        on_persona_snapshot: object = None,
     ) -> None:
         """Render one Worldtree SSE event with the TUI hierarchy + coalescing.
 
@@ -292,6 +384,12 @@ class TuiPresenterState:
           - `thinking_log` (RichLog) = streaming Thinking deltas inline
             (coalesced on `\n`); Rule(start)/Rule(end) wrap each run.
 
+        v0.13.0: optional `on_persona_snapshot` callback receives the
+        snapshot dict whenever AffectUpdate(status="current") arrives.
+        Lets the App surface the snapshot to the persona-header + Persona
+        pane without the presenter needing direct widget access. Default
+        None — presenter falls back to audit-only routing.
+
         Exceptions caught at the presenter boundary (INV-009 fallback).
         """
         assert isinstance(
@@ -299,6 +397,7 @@ class TuiPresenterState:
             (
                 WorkerPhase, Thinking, Text, TextBoundary,
                 ToolStart, ToolResult, Done, Error, Cancelled, AffectUpdate,
+                AwaitingLlmFirstToken,
             ),
         )
         from rich.text import Text as RichText
@@ -331,14 +430,52 @@ class TuiPresenterState:
                 if self.turn_start_ts == 0.0:
                     self.turn_start_ts = _time.monotonic()
                 debug_log.write(_dim(_audit_line(event)))
-            # v0.11.0: AffectUpdate is debug-pane-only for now (the audit
-            # line emitted above is the complete handling). Return early
-            # so the event doesn't pass through the thinking-close path
-            # or fall into the unknown-event ValueError branch. A full
-            # persona surface (Persona TabPane, sticky header line, or
-            # similar) is deferred to a later bump pending UX direction.
+            # v0.11.0 → v0.13.0: AffectUpdate gets the audit line (above)
+            # plus a callback to the App so the persona-header + Persona
+            # pane refresh from the snapshot. status="scheduled" carries
+            # no snapshot — the callback is skipped and the next turn's
+            # status="current" lands the actual update.
             if isinstance(event, AffectUpdate):
+                if event.snapshot is not None and on_persona_snapshot is not None:
+                    try:
+                        on_persona_snapshot(event.snapshot)
+                    except Exception:
+                        # Persona surface failure must not break the SSE
+                        # stream — the audit line above already records
+                        # the event regardless.
+                        pass
                 return
+            # v0.14.0: AwaitingLlmFirstToken (Worldtree #201) gets the audit
+            # line (above) plus a live transcript indicator. First heartbeat
+            # mounts a Static; subsequent heartbeats update it in place.
+            # Any non-heartbeat event below closes the gap and the indicator
+            # is removed (the first text/thinking/done arrived).
+            if isinstance(event, AwaitingLlmFirstToken):
+                from rich.text import Text as RichText
+                self.heartbeat_count += 1
+                secs = event.elapsed_ms_since_building_prompt / 1000.0
+                label = RichText(
+                    f"awaiting first token · {secs:.1f}s",
+                    style=_AU_DEMOTED,
+                )
+                try:
+                    if self.awaiting_widget is None:
+                        self.awaiting_widget = Static(label, classes="awaiting-label")
+                        transcript.mount(self.awaiting_widget)
+                    else:
+                        self.awaiting_widget.update(label)
+                    transcript.scroll_end(animate=False)
+                except Exception:
+                    pass
+                return
+            # Any non-heartbeat event past this point means the gap closed —
+            # remove the awaiting indicator if it's still mounted.
+            if self.awaiting_widget is not None:
+                try:
+                    self.awaiting_widget.remove()
+                except Exception:
+                    pass
+                self.awaiting_widget = None
             # v0.7.1: Thinking deltas coalesce by newline before flushing.
             # Worldtree emits Thinking events at token granularity; per-delta
             # RichLog writes produce one visual line per token (per-token-per-
@@ -434,6 +571,7 @@ class TuiPresenterState:
                     f"text_bytes={self.text_byte_count} "
                     f"thinking_deltas={self.thinking_delta_count} "
                     f"thinking_bytes={self.thinking_byte_count} "
+                    f"heartbeats={self.heartbeat_count} "
                     f"elapsed_ms={elapsed_ms}"
                 ))
                 # Terminal event: finalize the response widget (clear ref so
@@ -747,6 +885,14 @@ class RatatoskrApp(App[int]):
         height: auto;
         padding: 0 1;
     }
+    /* v0.14.0: Worldtree #201 — live "awaiting first token · Ns" indicator
+       in the transcript during the BuildingPrompt → CallingLLM gap.
+       Demoted styling so it reads as ambient progress, not content. */
+    .awaiting-label {
+        height: auto;
+        padding: 0 1;
+        color: $au-dark-60;
+    }
     /* v0.8.1: #current-text Static removed. Streaming text now coalesces
        on `\n` and writes directly to #transcript (same pattern as v0.7.1
        thinking fix). Eliminates the dock-bottom-growth-overlap bug. */
@@ -796,6 +942,20 @@ class RatatoskrApp(App[int]):
         color: $au-dark-60;
         padding: 0 1;
     }
+    /* v0.13.0: sticky persona-header — one-line agent persona summary
+       above the main row. Empty (height: 0) when the agent has no
+       persona surface (PersonaNotConfigured) so the chat layout
+       collapses cleanly. */
+    #persona-header {
+        dock: top;
+        height: 1;
+        color: $au-bright-80;
+        background: $surface;
+        padding: 0 1;
+    }
+    #persona-header.empty {
+        display: none;
+    }
     """
 
     BINDINGS: ClassVar[list[Binding]] = [
@@ -806,6 +966,7 @@ class RatatoskrApp(App[int]):
         Binding("ctrl+1", "focus_tools", "Tools tab", priority=False),
         Binding("ctrl+2", "focus_debug", "Debug tab", priority=False),
         Binding("ctrl+3", "focus_thinking", "Thinking tab", priority=False),
+        Binding("ctrl+4", "focus_persona", "Persona tab", priority=False),
     ]
 
     HINT_IDLE = "Ctrl-C twice to exit"
@@ -834,6 +995,11 @@ class RatatoskrApp(App[int]):
 
     def compose(self) -> ComposeResult:
         yield Header()
+        # v0.13.0: sticky persona-header docks at the top, above main-row.
+        # One-line summary refreshed on each AffectUpdate(status=current).
+        # Starts in the .empty CSS class (height collapses to 0) until
+        # on_mount's get_persona_state hydration succeeds.
+        yield Static("", id="persona-header", classes="empty")
         # v0.6.0 layout: left column is content-only (transcript + streaming
         # text Static + prompt). Right column hosts thinking-current live
         # preview above TabbedContent cycling Tools / Debug / Thinking.
@@ -875,6 +1041,14 @@ class RatatoskrApp(App[int]):
                         yield RichLog(
                             id="thinking-log", wrap=True, markup=False, highlight=False
                         )
+                    with TabPane("Persona", id="persona-tab"):
+                        # v0.13.0: full persona-snapshot detail (PAD,
+                        # mood drift, active emotions). Replaced (not
+                        # appended) on each AffectUpdate(current) — the
+                        # snapshot is absolute state, not incremental.
+                        yield RichLog(
+                            id="persona-log", wrap=True, markup=False, highlight=False
+                        )
         # INV-002 + INV-003: visible identity + hint widgets (Footer-area).
         # pane-name widget displays current side-pane name.
         yield Static("", id="identity")
@@ -922,6 +1096,81 @@ class RatatoskrApp(App[int]):
             f"session={self.session_id[-8:]} raw={self.args.raw} "
             f"end_user_id={getattr(self.args, 'end_user_id', None)!r}"
         )
+        # v0.13.0: hydrate persona surface on mount via
+        # GET /agents/{id}/persona_state. Spawns as a Textual worker so the
+        # network call doesn't block mount. Agents without a persona
+        # surface (PersonaNotConfigured) get a placeholder + empty header.
+        if self.agent_id is not None:
+            self.run_worker(self._hydrate_persona())
+
+    async def _hydrate_persona(self) -> None:
+        """Hydrate persona-header + Persona pane via GET /agents/{id}/persona_state.
+
+        Failure modes are absorbed (this is best-effort observability):
+          - PersonaNotConfigured: pane shows placeholder, header stays empty
+          - AgentNotAvailable / AuthScopeDenied: error placeholder; header empty
+          - Network error: error placeholder; header empty
+        On 200: header populated, pane shows full detail, audit logged.
+        """
+        assert self.client is not None and self.agent_id is not None
+        from rich.text import Text as RichText
+        try:
+            snapshot = await get_persona_state(self.client, self.agent_id)
+            self._update_persona_surfaces(snapshot)
+            self._audit(
+                f"persona_hydrated agent_id={self.agent_id!r} "
+                f"dominant_emotion={snapshot.get('dominant_emotion')!r}"
+            )
+        except PersonaNotConfigured:
+            self._set_persona_placeholder(
+                f"(persona not configured for {self.agent_id})"
+            )
+            self._audit(f"persona_not_configured agent_id={self.agent_id!r}")
+        except (AgentNotAvailable, AuthScopeDenied, SessionApiFailed, Exception) as exc:
+            # Best-effort — never let a persona hydration failure crash the
+            # TUI. Surface the error in the persona pane and audit log.
+            self._set_persona_placeholder(
+                f"(persona hydration failed: {type(exc).__name__})"
+            )
+            self._audit(
+                f"persona_hydration_failed agent_id={self.agent_id!r} "
+                f"err={type(exc).__name__}: {exc!s:.120}"
+            )
+
+    def _update_persona_surfaces(self, snapshot: dict) -> None:
+        """Update sticky header + Persona pane from a fresh snapshot.
+
+        Called on bootstrap (on_mount) and on each AffectUpdate(current).
+        Header gets the compact one-liner; pane gets the full detail.
+        """
+        from rich.text import Text as RichText
+        try:
+            header = self.query_one("#persona-header", Static)
+            header.update(RichText(_format_persona_header(snapshot)))
+            header.remove_class("empty")
+        except Exception:
+            pass
+        try:
+            log = self.query_one("#persona-log", RichLog)
+            log.clear()
+            log.write(_format_persona_detail(snapshot))
+        except Exception:
+            pass
+
+    def _set_persona_placeholder(self, text: str) -> None:
+        """Render an italic-dim placeholder in the Persona pane; keep header empty.
+
+        Used when persona hydration returns PersonaNotConfigured or fails —
+        the pane stays usable as documentation of *why* it's empty without
+        the sticky header consuming a row for nothing.
+        """
+        from rich.text import Text as RichText
+        try:
+            log = self.query_one("#persona-log", RichLog)
+            log.clear()
+            log.write(RichText(text, style=f"{_AU_DEMOTED_FAINT} italic"))
+        except Exception:
+            pass
 
     def _write_turn_headers(self, turn_id: int) -> None:
         """v0.6.0: turn-ID headers across every pane for cross-pane
@@ -1066,6 +1315,7 @@ class RatatoskrApp(App[int]):
                     debug_log=debug_log,
                     thinking_log=thinking_log,
                     raw=self.args.raw,
+                    on_persona_snapshot=self._update_persona_surfaces,
                 )
                 if isinstance(event, (Done, Error, Cancelled)):
                     break
@@ -1151,6 +1401,11 @@ class RatatoskrApp(App[int]):
         self.query_one("#side-panes", TabbedContent).active = "thinking-tab"
         self.query_one("#pane-name", Static).update("Thinking")
 
+    def action_focus_persona(self) -> None:
+        """v0.13.0: Ctrl+4 activates the Persona tab. INV-016 preserves Input focus."""
+        self.query_one("#side-panes", TabbedContent).active = "persona-tab"
+        self.query_one("#pane-name", Static).update("Persona")
+
 
 def run_tui(args: ParsedArgs) -> int:
     """Sync entry point — delegates to the async resolve-then-run flow.

tests/test_sessions.py

@@ -6,11 +6,15 @@ import respx
 
 from ratatoskr.sessions import (
     AgentInfo,
+    AgentNotAvailable,
     AgentNotFound,
+    AuthScopeDenied,
     InvalidCursor,
+    PersonaNotConfigured,
     SessionApiFailed,
     SessionPage,
     create_session,
+    get_persona_state,
     list_agents,
     list_sessions,
 )
@@ -556,3 +560,108 @@ class TestListAgents:
             with pytest.raises(SessionApiFailed) as excinfo:
                 await list_agents(client)
         assert excinfo.value.status == 401
+
+
+class TestGetPersonaState:
+    """Worldtree #204 / v0.28.0 — GET /agents/{agent_id}/persona_state.
+
+    Bootstrap read for the persona snapshot — same shape as `affect_update`'s
+    `current` snapshot. Auth via `persona.read` scope (user-tier default).
+    """
+
+    @respx.mock
+    async def test_happy_full_snapshot(self) -> None:
+        """happy_full_snapshot [happy,tracer]: 200 → snapshot dict with pad +
+        dominant_emotion + emotions_active + baseline_pad + mood_drift.
+        """
+        snapshot = {
+            "agent_id": "mimir",
+            "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
+            "dominant_emotion": "curiosity",
+            "emotions_active": [
+                {"type": "curiosity", "intensity": 0.6, "decay_remaining_s": 202.7}
+            ],
+            "baseline_pad": {"pleasure": 0.50, "arousal": 0.40, "dominance": 0.50},
+            "mood_drift": {"valence_delta": 0.02, "arousal_delta": 0.07},
+            "last_updated_at": "2026-05-25T22:30:18+00:00",
+        }
+        respx.get("https://w.example/agents/mimir/persona_state").mock(
+            return_value=httpx.Response(200, json=snapshot)
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            result = await get_persona_state(client, "mimir")
+        assert result == snapshot
+
+    @respx.mock
+    async def test_persona_not_configured_404(self) -> None:
+        """persona_not_configured_404 [error]: 404 with error_code
+        persona_not_configured → PersonaNotConfigured. Agent exists but has
+        no persona surface (e.g. domari, muninn, Tier 3).
+        """
+        respx.get("https://w.example/agents/domari/persona_state").mock(
+            return_value=httpx.Response(
+                404, json={"error_code": "persona_not_configured", "message": "no persona"}
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(PersonaNotConfigured) as exc_info:
+                await get_persona_state(client, "domari")
+        assert exc_info.value.agent_id == "domari"
+
+    @respx.mock
+    async def test_agent_not_available_404(self) -> None:
+        """agent_not_available_404 [error]: 404 with error_code
+        agent_not_available → AgentNotAvailable. Distinct from
+        persona_not_configured — the agent_id itself is unknown.
+        """
+        respx.get("https://w.example/agents/bogus/persona_state").mock(
+            return_value=httpx.Response(
+                404, json={"error_code": "agent_not_available", "message": "unknown agent"}
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(AgentNotAvailable) as exc_info:
+                await get_persona_state(client, "bogus")
+        assert exc_info.value.agent_id == "bogus"
+
+    @respx.mock
+    async def test_auth_scope_denied_403(self) -> None:
+        """auth_scope_denied_403 [error]: 403 with error_code auth_scope_denied
+        → AuthScopeDenied. Key lacks `persona.read` scope.
+        """
+        respx.get("https://w.example/agents/mimir/persona_state").mock(
+            return_value=httpx.Response(
+                403,
+                json={"error_code": "auth_scope_denied", "message": "missing persona.read"},
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(AuthScopeDenied) as exc_info:
+                await get_persona_state(client, "mimir")
+        assert exc_info.value.scope == "persona.read"
+
+    @respx.mock
+    async def test_404_unknown_error_code_falls_through(self) -> None:
+        """404_unknown_error_code_falls_through [adversarial]: 404 without the
+        two known error codes → SessionApiFailed (don't swallow novel failure
+        modes as something more specific than they are).
+        """
+        respx.get("https://w.example/agents/mimir/persona_state").mock(
+            return_value=httpx.Response(404, json={"error_code": "novel_404"})
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(SessionApiFailed) as exc_info:
+                await get_persona_state(client, "mimir")
+        assert exc_info.value.status == 404
+
+    @respx.mock
+    async def test_500_unexpected_status(self) -> None:
+        """500_unexpected_status [error]: 5xx → SessionApiFailed (matches the
+        list_agents / list_sessions / create_session precedent)."""
+        respx.get("https://w.example/agents/mimir/persona_state").mock(
+            return_value=httpx.Response(500, content=b"boom")
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            with pytest.raises(SessionApiFailed) as exc_info:
+                await get_persona_state(client, "mimir")
+        assert exc_info.value.status == 500

tests/test_sse_client.py

@@ -6,6 +6,7 @@ import respx
 
 from ratatoskr.sse_client import (
     AffectUpdate,
+    AwaitingLlmFirstToken,
     CancelAlreadyCompleted,
     Cancelled,
     CancelResult,
@@ -955,3 +956,87 @@ class TestAffectUpdate:
         assert affect.turn_id == 42
         assert affect.snapshot is None
         assert affect.sse_id == SseId(42, 2)
+
+
+class TestAwaitingLlmFirstToken:
+    """Worldtree #201 / v0.29.0 — `awaiting_llm_first_token` SSE heartbeat.
+
+    Top-level event (not a worker_phase extension) fired during the
+    BuildingPrompt → CallingLLM gap at the configured interval (default
+    5s). Server-authoritative elapsed_ms is time.monotonic()-based and
+    monotonically increasing across the heartbeat sequence.
+
+    See docs/conversation-api-spec.md § awaiting_llm_first_token.
+    """
+
+    @respx.mock
+    async def test_single_heartbeat_parsed(self) -> None:
+        """single_heartbeat_parsed [tracer]: type=awaiting_llm_first_token →
+        AwaitingLlmFirstToken(turn_id, elapsed_ms_since_building_prompt).
+        """
+        stream = _sse_chunk(
+            "42:1",
+            {
+                "type": "awaiting_llm_first_token",
+                "turn_id": 42,
+                "elapsed_ms_since_building_prompt": 5012.3,
+            },
+        ) + _sse_chunk("42:2", _DONE_42_6)
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        beat = events[0]
+        assert isinstance(beat, AwaitingLlmFirstToken)
+        assert beat.turn_id == 42
+        assert beat.elapsed_ms_since_building_prompt == 5012.3
+        assert beat.sse_id == SseId(42, 1)
+
+    @respx.mock
+    async def test_heartbeat_sequence_monotonic(self) -> None:
+        """heartbeat_sequence_monotonic [scenario]: three consecutive heartbeats
+        in one turn — elapsed_ms_since_building_prompt monotonically increases,
+        all carry the same turn_id.
+        """
+        stream = (
+            _sse_chunk(
+                "42:1",
+                {
+                    "type": "awaiting_llm_first_token",
+                    "turn_id": 42,
+                    "elapsed_ms_since_building_prompt": 5000.0,
+                },
+            )
+            + _sse_chunk(
+                "42:2",
+                {
+                    "type": "awaiting_llm_first_token",
+                    "turn_id": 42,
+                    "elapsed_ms_since_building_prompt": 10005.4,
+                },
+            )
+            + _sse_chunk(
+                "42:3",
+                {
+                    "type": "awaiting_llm_first_token",
+                    "turn_id": 42,
+                    "elapsed_ms_since_building_prompt": 15011.8,
+                },
+            )
+            + _sse_chunk("42:4", _DONE_42_6)
+        )
+        respx.post("https://w.example/sessions/s1/messages").mock(
+            return_value=httpx.Response(
+                200, headers={"content-type": "text/event-stream"}, content=stream
+            )
+        )
+        async with httpx.AsyncClient(base_url="https://w.example") as client:
+            events = [e async for e in stream_turn(client, "s1", "hi")]
+        beats = [e for e in events if isinstance(e, AwaitingLlmFirstToken)]
+        assert len(beats) == 3
+        elapsed = [b.elapsed_ms_since_building_prompt for b in beats]
+        assert elapsed == sorted(elapsed)  # monotonically increasing
+        assert all(b.turn_id == 42 for b in beats)

tests/test_tui.py

@@ -719,6 +719,95 @@ class TestPresenterAuditLogging:
         assert not tools_log.write.called
         assert not thinking_log.write.called
 
+    def test_awaiting_llm_first_token_mounts_indicator(self) -> None:
+        """awaiting_llm_first_token_mounts_indicator [v0.14.0]: first heartbeat
+        mounts a Static into the transcript and bumps heartbeat_count;
+        debug-pane audit line carries turn_id + elapsed in seconds.
+        """
+        from ratatoskr.sse_client import AwaitingLlmFirstToken
+        from ratatoskr.tui import TuiPresenterState
+
+        transcript = MagicMock()
+        debug_log = MagicMock()
+        state = TuiPresenterState()
+        state.render(
+            AwaitingLlmFirstToken(
+                sse_id=SID, turn_id=42, elapsed_ms_since_building_prompt=5012.3
+            ),
+            transcript=transcript,
+            tools_log=MagicMock(),
+            debug_log=debug_log,
+            thinking_log=MagicMock(),
+            raw=False,
+        )
+        assert state.heartbeat_count == 1
+        assert state.awaiting_widget is not None
+        assert transcript.mount.call_count == 1
+        audit = _text_of(debug_log.write.call_args[0][0])
+        assert "awaitingllmfirsttoken" in audit
+        assert "turn_id=42" in audit
+        assert "elapsed=5.0s" in audit
+
+    def test_awaiting_subsequent_heartbeats_update_in_place(self) -> None:
+        """awaiting_subsequent_heartbeats_update_in_place [v0.14.0]: second+
+        heartbeats reuse the existing Static (no new mount); heartbeat_count
+        tracks the total.
+        """
+        from ratatoskr.sse_client import AwaitingLlmFirstToken
+        from ratatoskr.tui import TuiPresenterState
+
+        transcript = MagicMock()
+        state = TuiPresenterState()
+        for elapsed in (5000.0, 10005.4, 15011.8):
+            state.render(
+                AwaitingLlmFirstToken(
+                    sse_id=SID, turn_id=42, elapsed_ms_since_building_prompt=elapsed
+                ),
+                transcript=transcript,
+                tools_log=MagicMock(),
+                debug_log=MagicMock(),
+                thinking_log=MagicMock(),
+                raw=False,
+            )
+        assert state.heartbeat_count == 3
+        assert transcript.mount.call_count == 1  # mounted once on first
+
+    def test_awaiting_indicator_removed_when_gap_closes(self) -> None:
+        """awaiting_indicator_removed_when_gap_closes [v0.14.0]: any non-
+        heartbeat event after one or more heartbeats removes the indicator
+        and clears the awaiting_widget reference. Text event simulates the
+        gap closing (CallingLLM fires, text begins).
+        """
+        from ratatoskr.sse_client import AwaitingLlmFirstToken
+        from ratatoskr.tui import TuiPresenterState
+
+        transcript = MagicMock()
+        state = TuiPresenterState()
+        state.render(
+            AwaitingLlmFirstToken(
+                sse_id=SID, turn_id=42, elapsed_ms_since_building_prompt=5000.0
+            ),
+            transcript=transcript,
+            tools_log=MagicMock(),
+            debug_log=MagicMock(),
+            thinking_log=MagicMock(),
+            raw=False,
+        )
+        widget = state.awaiting_widget
+        assert widget is not None
+        state.render(
+            Text(sse_id=SID, content="hello"),
+            transcript=transcript,
+            tools_log=MagicMock(),
+            debug_log=MagicMock(),
+            thinking_log=MagicMock(),
+            raw=False,
+        )
+        # State reference cleared (the widget itself is a real Static whose
+        # .remove() schedules removal — we verify the cleanup intent via
+        # the state field, which is the contract callers actually observe).
+        assert state.awaiting_widget is None
+
     def test_affect_update_scheduled_has_no_pad_detail(self) -> None:
         """affect_update_scheduled_has_no_pad_detail [v0.11.0]: status=scheduled
         carries no snapshot — the audit line omits dominant_emotion / pad and
@@ -934,6 +1023,156 @@ class TestCancelViaSse:
         assert audit_lines[1].startswith("cancel_post failed turn_id=42 CancelFailed")
 
 
+class TestPersonaFormatters:
+    """v0.13.0 — _format_persona_header / _format_persona_detail rendering."""
+
+    def test_header_compact_summary(self) -> None:
+        """header_compact_summary: agent_id · dominant_emotion · pad(P,A,D) · N emotions."""
+        from ratatoskr.tui import _format_persona_header
+
+        snapshot = {
+            "agent_id": "mimir",
+            "dominant_emotion": "curiosity",
+            "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
+            "emotions_active": [
+                {"type": "curiosity"}, {"type": "joy"},
+            ],
+        }
+        line = _format_persona_header(snapshot)
+        assert "mimir" in line
+        assert "curiosity" in line
+        assert "pad(0.52, 0.47, 0.5)" in line
+        assert "2 emotions active" in line
+
+    def test_header_singular_emotion(self) -> None:
+        """header_singular_emotion: single emotion → '1 emotion active' (no 's')."""
+        from ratatoskr.tui import _format_persona_header
+
+        line = _format_persona_header(
+            {
+                "agent_id": "mimir",
+                "dominant_emotion": "calm",
+                "pad": {"pleasure": 0.5, "arousal": 0.4, "dominance": 0.5},
+                "emotions_active": [{"type": "calm"}],
+            }
+        )
+        assert "1 emotion active" in line
+        assert "1 emotions" not in line
+
+    def test_header_no_emotions_drops_count(self) -> None:
+        """header_no_emotions_drops_count: empty emotions list → no count suffix."""
+        from ratatoskr.tui import _format_persona_header
+
+        line = _format_persona_header(
+            {
+                "agent_id": "mimir",
+                "dominant_emotion": "?",
+                "pad": {"pleasure": 0.5, "arousal": 0.4, "dominance": 0.5},
+                "emotions_active": [],
+            }
+        )
+        assert "emotion" not in line  # neither "1 emotion" nor "N emotions"
+
+    def test_detail_renders_full_snapshot(self) -> None:
+        """detail_renders_full_snapshot: PAD axes + baseline + delta + drift +
+        emotions + last_updated_at all surface in the multi-line render.
+        """
+        from ratatoskr.tui import _format_persona_detail
+
+        snapshot = {
+            "agent_id": "mimir",
+            "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
+            "dominant_emotion": "curiosity",
+            "emotions_active": [
+                {"type": "curiosity", "intensity": 0.6, "decay_remaining_s": 202.7}
+            ],
+            "baseline_pad": {"pleasure": 0.50, "arousal": 0.40, "dominance": 0.50},
+            "mood_drift": {"valence_delta": 0.02, "arousal_delta": 0.07},
+            "last_updated_at": "2026-05-25T22:30:18+00:00",
+        }
+        detail = _format_persona_detail(snapshot)
+        assert "Persona snapshot · mimir" in detail
+        assert "Dominant emotion: curiosity" in detail
+        assert "pleasure" in detail
+        assert "baseline 0.5" in detail
+        assert "+0.02" in detail or "0.02" in detail
+        assert "valence_delta" in detail
+        assert "curiosity" in detail
+        assert "intensity 0.6" in detail
+        assert "decay 3.4m" in detail
+        assert "2026-05-25T22:30:18+00:00" in detail
+
+
+class TestPresenterPersonaCallback:
+    """v0.13.0 — presenter wires AffectUpdate snapshots into a callback."""
+
+    def test_current_invokes_callback_with_snapshot(self) -> None:
+        """current_invokes_callback_with_snapshot: AffectUpdate(current, snapshot)
+        calls on_persona_snapshot(snapshot)."""
+        from ratatoskr.sse_client import AffectUpdate
+        from ratatoskr.tui import TuiPresenterState
+
+        captured: list = []
+        state = TuiPresenterState()
+        snapshot = {"agent_id": "mimir", "pad": {"pleasure": 0.5}}
+        state.render(
+            AffectUpdate(sse_id=SID, status="current", turn_id=42, snapshot=snapshot),
+            transcript=MagicMock(),
+            tools_log=MagicMock(),
+            debug_log=MagicMock(),
+            thinking_log=MagicMock(),
+            raw=False,
+            on_persona_snapshot=captured.append,
+        )
+        assert captured == [snapshot]
+
+    def test_scheduled_does_not_invoke_callback(self) -> None:
+        """scheduled_does_not_invoke_callback: status=scheduled has no snapshot,
+        so the callback is skipped (would be called with None otherwise)."""
+        from ratatoskr.sse_client import AffectUpdate
+        from ratatoskr.tui import TuiPresenterState
+
+        captured: list = []
+        state = TuiPresenterState()
+        state.render(
+            AffectUpdate(sse_id=SID, status="scheduled", turn_id=42, snapshot=None),
+            transcript=MagicMock(),
+            tools_log=MagicMock(),
+            debug_log=MagicMock(),
+            thinking_log=MagicMock(),
+            raw=False,
+            on_persona_snapshot=captured.append,
+        )
+        assert captured == []
+
+    def test_callback_exception_swallowed(self) -> None:
+        """callback_exception_swallowed: a raising callback does NOT crash the
+        presenter — the audit line still landed (it precedes the callback).
+        """
+        from ratatoskr.sse_client import AffectUpdate
+        from ratatoskr.tui import TuiPresenterState
+
+        def boom(_snap: dict) -> None:
+            raise RuntimeError("widget tearing down")
+
+        debug_log = MagicMock()
+        state = TuiPresenterState()
+        # Should NOT raise
+        state.render(
+            AffectUpdate(
+                sse_id=SID, status="current", turn_id=42, snapshot={"agent_id": "x"}
+            ),
+            transcript=MagicMock(),
+            tools_log=MagicMock(),
+            debug_log=debug_log,
+            thinking_log=MagicMock(),
+            raw=False,
+            on_persona_snapshot=boom,
+        )
+        # Audit line still emitted (it runs before the callback)
+        assert debug_log.write.called
+
+
 class TestAppMount:
     """on_mount narrows per issue #6: only identity-widget population.
 
@@ -1104,6 +1343,43 @@ class TestLayoutShape:
             await pilot.pause()
             assert app.query_one("#side-panes", TabbedContent).active == "debug-tab"
 
+    async def test_persona_tab_exists(self) -> None:
+        """persona_tab_exists [v0.13.0]: right column has Persona TabPane +
+        #persona-log RichLog as descendant.
+        """
+        from textual.widgets import RichLog, TabPane
+
+        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            persona_tab = app.query_one("#persona-tab", TabPane)
+            persona_log = app.query_one("#persona-log", RichLog)
+            assert persona_log in persona_tab.walk_children()
+
+    async def test_ctrl_4_activates_persona_tab(self) -> None:
+        """ctrl_4_activates_persona_tab [v0.13.0]: Ctrl+4 → active == 'persona-tab'."""
+        from textual.widgets import TabbedContent
+
+        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.press("ctrl+4")
+            await pilot.pause()
+            assert app.query_one("#side-panes", TabbedContent).active == "persona-tab"
+
+    async def test_persona_header_starts_empty(self) -> None:
+        """persona_header_starts_empty [v0.13.0]: sticky persona-header widget
+        exists, starts hidden (height collapsed via .empty class) until
+        hydration succeeds.
+        """
+        from textual.widgets import Static
+
+        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            header = app.query_one("#persona-header", Static)
+            assert "empty" in header.classes
+
     async def test_done_label_styled_success(self) -> None:
         """done_label_styled_success [v0.9.0]: [done] label mounts as Static
         carrying a RichText with Aurora green style. Inspect the mounted

uv.lock

@@ -968,7 +968,7 @@ wheels = [
 
 [[package]]
 name = "ratatoskr"
-version = "0.11.0"
+version = "0.14.0"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },