feat(tui): persona surface — sticky header + TabPane (v0.13.0)

Step 3 of the Worldtree #204 integration: visible persona-state UX.
Pairs with v0.11.0's AffectUpdate SSE event + v0.12.0's
get_persona_state HTTP client — together those gave the data; this
bump surfaces it.

Two surfaces (Option C: both):

Sticky persona-header (top of screen, dock=top, height=1):
- Shape: `agent_id · dominant_emotion · pad(P, A, D) · N emotions
  active` — concise enough for at-a-glance scan above the chat
- Starts hidden via `.empty` CSS class; height collapses to 0 when
  the agent has no persona surface
- Refreshes on AffectUpdate(status="current") snapshots

Persona TabPane (Ctrl+4):
- Full snapshot detail: dominant emotion, PAD axes with baseline +
  delta, mood drift, active emotions with intensity + decay
  (minutes-rounded), last_updated_at footer
- Replaced (not appended) on each new snapshot — snapshots are
  absolute state, not incremental

Lifecycle:
- on_mount spawns a Textual worker that calls get_persona_state to
  hydrate header + pane before turn 1
- PersonaNotConfigured (domari, muninn, Tier 3) → pane carries an
  italic placeholder, header stays empty
- AgentNotAvailable / AuthScopeDenied / network error → italic
  failure placeholder; audit-logged; never crashes
- Presenter's render() takes an optional `on_persona_snapshot`
  callback so AffectUpdate(current) refreshes both surfaces during
  a live turn (no widget coupling — App owns the callback)

Tests: 10 new (4 formatters, 3 presenter callback, 4 layout/binding/
hydration). Full suite: 313 passing.

docs/SPEC-PIN.md

@@ -7,11 +7,18 @@ documents the pin, the vendored artifacts, and the bump procedure.
 
 | Field | Value |
 |---|---|
-| Worldtree git SHA | `55101e909abcd2219833266b6f905c5bc956e0f0` |
-| Worldtree HEAD message | `memory: snapshot — #177 Vili v1 + persona async-decouple shipped as v0.19.0` |
-| Pinned on | 2026-05-20 |
-| Pinned by | brokkr-smithy-dev (initial scaffold) |
-| Worldtree version at pin | `v0.19.0` |
+| 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` |
+
+## 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-20 | `55101e9` | v0.19.0 | initial scaffold pin |
 
 ## Vendored artifacts
 

docs/conversation-api-spec.md

@@ -55,6 +55,69 @@ conversation_api:
 
 ---
 
+## Authorization model — agent invocation
+
+When you call `POST /sessions` against an agent, the authorization check that fires depends on **which kind of agent** you target. There are two distinct scope namespaces — the spelling differs by one character (`agent` vs `agents`) and the granting mechanism differs entirely. Confusing the two is a common source of bug reports.
+
+### Tier 1 — foundational agents (no `:` in agent_id)
+
+Agents bundled with Worldtree: `mimir`, `lofn`, `soong`, `forseti`, `domari`, `vili`, `actor`, `saga`, `bragi`, `leif`, `troi`, `cara`, `glados`, and any future Asgardian. The agent_id is a simple slug like `mimir` — no colon.
+
+> **About tiers:** Your `tier` is set on the `users` table row your API key resolves to, assigned at key-mint time (see `POST /admin/keys`). Tiers are `anonymous` (dev-mode unauthenticated), `user` (default for newly-issued keys), `free`/`pro` (subscription-shaped, not actively differentiated), and `admin`. The tier you have is visible via `GET /me`'s `tier` field. Tier-derived scopes come from `config/policies.yaml > tiers.<tier>.scopes` — there is no per-key scope override.
+
+**Authorization rule (singular `agent`):**
+
+```yaml
+- id: agent-call-baseline-allow
+  principal:
+    tiers: ["anonymous", "user", "free", "pro", "admin"]
+  action: "agent.call:*"
+  resource: "*"
+  effect: allow
+```
+
+This baseline rule lives at `config/policies.yaml`. Every authenticated tier — including the `user` tier that newly-issued keys default to — already passes this check for every Tier 1 agent. **There is no per-agent scope you can add to "grant" Tier 1 access; it's covered by tier.**
+
+If you get a 422 calling a Tier 1 agent (e.g., `lofn` rejecting with `end_user_id_required`), that's a **request-body validation**, not a scope denial. Check the `error_code` in the response detail — `END_USER_ID_REQUIRED` means the agent requires an `end_user_id` field in the request body; `AUTH_SCOPE_DENIED` (403) would be the actual scope problem. They're not interchangeable.
+
+### Tier 3 — consumer-defined agents (`:` in agent_id)
+
+Agents created at runtime via `POST /agents/define`. The agent_id is `<owner_user_id>:<agent_name>`, e.g., `acme:support-bot`. The `:` in the path is the trigger that switches the auth model.
+
+**Authorization is DB-backed per-resource, NOT policy-driven (plural `agents`):**
+
+```
+scope action checked: agents.call:<owner_user_id>:<agent_name>
+                       ^^^^^^
+                       PLURAL — different namespace from Tier 1
+```
+
+There is **no blanket allow rule** for `agents.call:*` in policy. The grant comes from the live `consumer_agents` table:
+
+- A non-soft-deleted row in `consumer_agents` owned by `ctx.user_id` IS the grant.
+- Cascade soft-delete and owner-initiated `DELETE` revoke it.
+- Missing row → policy defaults to deny (403 `auth_scope_denied`).
+
+To "add the scope" for a Tier 3 agent, you don't amend any config or call an admin endpoint — you `POST /agents/define` to register it under your `user_id`. Owning the row IS the grant. You cannot call another user's Tier 3 agent; ownership is checked at session-create (`row.user_id == ctx.user_id`).
+
+### Common pitfalls
+
+- **Singular vs plural.** Tier 1 uses `agent.call:*` (singular `agent`). Tier 3 uses `agents.call:<owner>:<name>` (plural `agents`). One character difference, two completely different mechanisms. There is no Tier 1 scope named `agent.call:mimir` or `agents.call:mimir` — Tier 1 is granted by baseline rule, not per-agent name.
+- **No scope-mutation API.** `POST /admin/keys` accepts `{user_id, label, tier}` only. There is no per-key scope override mechanism in the storage schema. To change a user's effective scopes, change their `tier`, not their key. Per-resource Tier 3 grants flow through `POST /agents/define` (and its DELETE counterpart), not through admin endpoints.
+- **422 vs 403.** A 422 is body-validation (e.g., `end_user_id_required`); a 403 is auth-policy denial (`auth_scope_denied`). Different fix paths. Read the `error_code` in `detail`.
+
+### Quick decision table for consumers
+
+| Target | Auth requirement |
+|---|---|
+| Tier 1 agent (e.g., `mimir`) | Authenticated tier ≥ `user`. No additional body requirements |
+| Tier 1 agent `lofn` (the default welcoming intermediary) | Authenticated tier ≥ `user` + `end_user_id` field required in request body. 422 `END_USER_ID_REQUIRED` if absent |
+| Tier 3 agent (any agent_id containing `:`) | `end_user_id` field required in body. AND the row must be owner-matched: `POST /agents/define` first to create a row under your `user_id`, then session-create works against your existing key. Cross-user Tier 3 invocation is rejected with 403 |
+
+> **Programmatic discovery of `end_user_id` requirements:** as of v0.22.x there is no field on `GET /agents` indicating which agents require `end_user_id` — the spec line above (lofn + Tier 3) is the authoritative list, and 422 `END_USER_ID_REQUIRED` is the fallback signal at request time. Adding a discoverable `requires_end_user_id` field on `AgentInfoResponse` is on the table as a small future capability; ping if you want to drive it.
+
+---
+
 ## GET /me
 
 Returns the authenticated principal's identity and key metadata. Lets a client verify its key on boot without triggering agent-config-loading side effects.
@@ -1878,6 +1941,45 @@ Tool-using turns cycle through `CallingLLM → ProcessingTools → CallingLLM
 
 Clients that don't need phase events can filter on `event["type"] != "worker_phase"` client-side. Existing SSE consumers that switch on `event["type"]` ignore this event type without code changes.
 
+### affect_update
+
+Persona-state observability event (issue #204). Fires twice per turn for agents with `persona.enabled: true` on non-ephemeral sessions; suppressed entirely for persona-disabled agents (e.g. `domari`, `muninn`), Tier 3 consumer-defined agents (Phase 2.0), and ephemeral sessions.
+
+**Start-of-turn — `status: "current"`:**
+
+Emitted immediately at the start of each qualifying turn, before any `worker_phase` event. Carries the agent's current persona snapshot reflecting all prior turns' completed appraisals.
+
+```json
+{
+  "type": "affect_update",
+  "status": "current",
+  "turn_id": 42,
+  "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"
+  }
+}
+```
+
+**End-of-turn — `status: "scheduled"`:**
+
+Emitted after the post-turn appraisal task has been scheduled (per #177 Phase A's fire-and-forget discipline) and before `done`. Lightweight notification — no PAD numbers, since the appraisal is still running asynchronously. The result lands in the NEXT turn's `status: "current"` snapshot.
+
+```json
+{"type": "affect_update", "status": "scheduled", "turn_id": 42}
+```
+
+`scheduled` is skipped on turn failure/cancel paths (the appraisal was never reached); `current` still fires unconditionally for qualifying turns.
+
+Bootstrap reads available via `GET /agents/{agent_id}/persona_state` (same `snapshot` shape, requires `persona.read` scope).
+
 ### thinking
 
 Incremental reasoning/thinking content (from thinking-enabled models).

docs/conversation_api.contract.md

@@ -1851,6 +1851,36 @@ SQLite `consumer_agents` table.
   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.
+
 ### 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.8.2"
+version = "0.13.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 = "55101e909abcd2219833266b6f905c5bc956e0f0"
-worldtree-version = "v0.19.0"
-pinned-on = "2026-05-20"
+worldtree-spec-rev = "da93ca7cf613f1dc229a7a44d07fc1d7efc78e25"
+worldtree-version = "v0.28.0"
+pinned-on = "2026-05-25"
 
 [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,30 @@ class Cancelled:
     partial_message_id: int | None
 
 
+@dataclass(frozen=True)
+class AffectUpdate:
+    """SSE event `affect_update`: persona-state observability snapshot.
+
+    Two emissions per qualifying turn (persona-enabled agent on non-
+    ephemeral session): `status="current"` at turn start carrying the full
+    snapshot, `status="scheduled"` after post-turn appraisal kicks off
+    (lightweight — `snapshot` is None). Suppressed entirely for persona-
+    disabled agents (e.g. `domari`, `muninn`), Tier 3 consumer-defined
+    agents (Phase 2.0), and ephemeral sessions.
+
+    Bootstrap reads available via `GET /agents/{agent_id}/persona_state`
+    (same `snapshot` shape, requires `persona.read` scope).
+
+    See docs/conversation-api-spec.md § affect_update (Worldtree #204,
+    v0.28.0).
+    """
+
+    sse_id: SseId
+    status: str  # "current" | "scheduled"
+    turn_id: int
+    snapshot: dict[str, Any] | None  # None when status="scheduled"
+
+
 Event = (
     WorkerPhase
     | Thinking
@@ -121,6 +145,7 @@ Event = (
     | Done
     | Error
     | Cancelled
+    | AffectUpdate
 )
 
 
@@ -284,6 +309,17 @@ 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 == "affect_update":
+        # Worldtree #204 / v0.28.0: persona-state observability event.
+        # status="current" carries full snapshot at turn start;
+        # status="scheduled" omits snapshot (lightweight post-appraisal-
+        # kickoff notification).
+        return AffectUpdate(
+            sse_id=sse_id,
+            status=body["status"],
+            turn_id=body["turn_id"],
+            snapshot=body.get("snapshot"),
+        )
     raise ValueError(f"unknown SSE event type: {t!r}")
 
 

src/ratatoskr/tui.py

@@ -10,13 +10,15 @@ from __future__ import annotations
 
 import asyncio
 import sys
+import time as _time
 from dataclasses import dataclass
+from datetime import datetime as _datetime
 from typing import ClassVar, Literal
 
 import httpx
 from textual.app import App, ComposeResult
 from textual.binding import Binding
-from textual.containers import Horizontal, Vertical
+from textual.containers import Horizontal, Vertical, VerticalScroll
 from textual.theme import Theme
 from textual.widgets import (
     Footer,
@@ -33,12 +35,17 @@ 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,
     CancelAlreadyCompleted,
     CancelFailed,
     Cancelled,
@@ -178,6 +185,135 @@ def _plain_label(event: Event) -> str:
     return f"[unknown_event] {type(event).__name__}"
 
 
+def _ts() -> str:
+    """HH:MM:SS.fff wall-clock timestamp for debug-pane log lines."""
+    now = _datetime.now()
+    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.
+
+    v0.10.0: every SSE event arrival lands as one of these in the debug
+    pane (Text and Thinking deltas are aggregated into the turn summary
+    instead — token-rate per-delta lines would drown the pane). Shape:
+    `[HH:MM:SS.fff] event_type sse_id=T:S key=val …`.
+    """
+    sid = getattr(event, "sse_id", None)
+    sid_str = f"{sid.turn_id}:{sid.seq}" if sid is not None else "-"
+    kind = type(event).__name__.lower()
+    if isinstance(event, WorkerPhase):
+        detail = f"phase={event.phase} turn_id={event.turn_id}"
+    elif isinstance(event, ToolStart):
+        detail = f"name={event.name} args={event.arguments!r:.80}"
+    elif isinstance(event, ToolResult):
+        detail = f"name={event.name} duration_ms={event.duration_ms}"
+    elif isinstance(event, TextBoundary):
+        detail = f"kind={event.kind} char_offset={event.char_offset}"
+    elif isinstance(event, Done):
+        detail = (
+            f"turn_id={event.sse_id.turn_id} model={event.model} "
+            f"duration_ms={event.duration_ms}"
+        )
+    elif isinstance(event, Error):
+        detail = (
+            f"turn_id={event.sse_id.turn_id} code={event.error_code} "
+            f"message={event.message!r:.80}"
+        )
+    elif isinstance(event, Cancelled):
+        detail = f"turn_id={event.turn_id} reason={event.reason!r}"
+    elif isinstance(event, AffectUpdate):
+        # Worldtree #204 / v0.28.0. status="current" carries the full
+        # snapshot; surface dominant_emotion + PAD inline so the operator
+        # sees persona drift at a glance. status="scheduled" is
+        # lightweight — no PAD, just the appraisal-kickoff marker.
+        if event.snapshot is not None:
+            pad = event.snapshot.get("pad") or {}
+            detail = (
+                f"status={event.status} turn_id={event.turn_id} "
+                f"dominant_emotion={event.snapshot.get('dominant_emotion')!r} "
+                f"pad=({pad.get('pleasure')},{pad.get('arousal')},{pad.get('dominance')})"
+            )
+        else:
+            detail = f"status={event.status} turn_id={event.turn_id}"
+    else:  # Text / Thinking handled by counter path; fallback for safety
+        detail = ""
+    return f"[{_ts()}] {kind} sse_id={sid_str} {detail}".rstrip()
+
+
 @dataclass(slots=True)
 class TuiPresenterState:
     """Per-turn presenter state for TUI mode (issue #12).
@@ -194,35 +330,51 @@ class TuiPresenterState:
     # only on `\n` boundaries (one written line per natural paragraph) or
     # when the run closes (any leftover tail).
     thinking_chunk_buffer: str = ""
-    # v0.8.1: same pattern for Text deltas. Pre-v0.8.1 the Text deltas
-    # streamed into a dedicated #current-text Static below the transcript;
-    # that Static (docked-bottom, height: auto) grew during streaming and
-    # visually OVERLAPPED the transcript above (Textual didn't dynamically
-    # resize the 1fr transcript while the dock-bottom child expanded).
-    # The Static is gone in v0.8.1 — Text deltas coalesce on `\n` and write
-    # directly to `log` (transcript), the same shape thinking uses.
+    # v0.9.0: Text accumulator for live Markdown rendering. Worldtree emits
+    # Text deltas at token granularity; each delta appends to this buffer
+    # and the current_response_widget re-renders Markdown(text_chunk_buffer)
+    # in place. On terminal event the widget is finalized + reference clears.
     text_chunk_buffer: str = ""
+    # v0.9.0: reference to the Static widget holding the current turn's
+    # response Markdown Renderable. None between turns.
+    current_response_widget: object = None
+    # v0.10.0: per-turn counters for the debug-pane turn-summary line. Text
+    # and Thinking events arrive at token rate; emitting per-delta debug
+    # lines would drown the pane. Instead we count them and surface
+    # aggregated totals when the turn closes.
+    text_delta_count: int = 0
+    text_byte_count: int = 0
+    thinking_delta_count: int = 0
+    thinking_byte_count: int = 0
+    turn_start_ts: float = 0.0
 
     def render(
         self,
         event: Event,
         *,
-        log: RichLog,
+        transcript: "VerticalScroll",
         tools_log: RichLog,
         debug_log: RichLog,
         thinking_log: RichLog,
         raw: bool,
+        on_persona_snapshot: object = None,
     ) -> None:
         """Render one Worldtree SSE event with the TUI hierarchy + coalescing.
 
-        v0.8.1 routing:
-          - `log` (transcript) = chat content: user-prompt echo (written
-            outside the presenter), coalesced Text deltas, terminal labels,
-            optional post-Done Markdown body.
-          - `tools_log` = ToolStart + ToolResult.
-          - `debug_log` = WorkerPhase + TextBoundary.
-          - `thinking_log` = streaming Thinking deltas inline (coalesced on
-            `\n`). Rule(start)/Rule(end) wrap each run.
+        v0.9.0 routing:
+          - `transcript` (VerticalScroll) = chat content: each turn mounts
+            child widgets (turn-header / prompt-echo / response Markdown /
+            done-label). Live Markdown rendering during Text streaming.
+          - `tools_log` (RichLog) = ToolStart + ToolResult.
+          - `debug_log` (RichLog) = WorkerPhase + TextBoundary.
+          - `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).
         """
@@ -230,7 +382,7 @@ class TuiPresenterState:
             event,
             (
                 WorkerPhase, Thinking, Text, TextBoundary,
-                ToolStart, ToolResult, Done, Error, Cancelled,
+                ToolStart, ToolResult, Done, Error, Cancelled, AffectUpdate,
             ),
         )
         from rich.text import Text as RichText
@@ -240,6 +392,44 @@ class TuiPresenterState:
             return RichText(s, style=_AU_DEMOTED)
 
         try:
+            # v0.10.0: per-event audit log line to debug pane. Text and
+            # Thinking arrive at token rate, so we count them rather than
+            # emit a line per delta — totals are reported in the turn-
+            # summary on Done/Error/Cancelled. Everything else gets one
+            # debug-pane line per arrival with timestamp + sse_id + a short
+            # event-specific summary, giving the operator a wire-level
+            # timeline of what the server sent.
+            if isinstance(event, Text):
+                if self.text_delta_count == 0:
+                    if self.turn_start_ts == 0.0:
+                        self.turn_start_ts = _time.monotonic()
+                self.text_delta_count += 1
+                self.text_byte_count += len(event.content)
+            elif isinstance(event, Thinking):
+                if self.thinking_delta_count == 0:
+                    if self.turn_start_ts == 0.0:
+                        self.turn_start_ts = _time.monotonic()
+                self.thinking_delta_count += 1
+                self.thinking_byte_count += len(event.content)
+            else:
+                if self.turn_start_ts == 0.0:
+                    self.turn_start_ts = _time.monotonic()
+                debug_log.write(_dim(_audit_line(event)))
+            # 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.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-
@@ -288,47 +478,95 @@ class TuiPresenterState:
                 # coalesced on `\n`. Same pattern as Thinking (v0.7.1).
                 # The pre-v0.8.1 #current-text Static is gone — its dock-
                 # bottom growth was overlapping the transcript visually.
+                #
+                # v0.9.0: Text deltas accumulate in text_chunk_buffer and
+                # the current_response_widget renders Markdown(buffer) in
+                # place. First Text delta of the turn mounts a fresh Static
+                # holding the Markdown Renderable; subsequent deltas update
+                # the same widget. Live markdown rendering — no post-Done
+                # re-render needed.
+                from rich.markdown import Markdown
+
                 self.text_chunk_buffer += event.content
-                while "\n" in self.text_chunk_buffer:
-                    line, _, rest = self.text_chunk_buffer.partition("\n")
-                    if line:
-                        log.write(line)
-                    self.text_chunk_buffer = rest
+                # --raw bypasses Markdown rendering — useful for debugging
+                # the raw text stream surface, and matches the pre-v0.9.0
+                # --raw semantics (which dropped the post-Done Markdown re-
+                # render). In raw mode the response widget holds plain str.
+                rendered = (
+                    self.text_chunk_buffer if raw else Markdown(self.text_chunk_buffer)
+                )
+                if self.current_response_widget is None:
+                    self.current_response_widget = Static(
+                        rendered, classes="response-md"
+                    )
+                    transcript.mount(self.current_response_widget)
+                else:
+                    self.current_response_widget.update(rendered)
+                transcript.scroll_end(animate=False)
                 return
             if isinstance(event, (Done, Error, Cancelled)):
-                # Terminal event: flush any remaining text tail before the
-                # label / Markdown body lands.
-                if self.text_chunk_buffer:
-                    log.write(self.text_chunk_buffer)
-                    self.text_chunk_buffer = ""
-                # Terminal labels tinted per outcome (Aurora green / Dawn red
-                # / Dawn yellow) for at-a-glance scanning.
+                # v0.10.0: emit turn-summary to debug pane before clearing
+                # counters. Aggregates the per-event totals (Text + Thinking
+                # deltas don't get per-event audit lines because they arrive
+                # at token rate; the summary surfaces what was elided).
+                elapsed_ms = (
+                    int((_time.monotonic() - self.turn_start_ts) * 1000)
+                    if self.turn_start_ts
+                    else 0
+                )
+                turn_id = (
+                    event.sse_id.turn_id
+                    if hasattr(event, "sse_id")
+                    else getattr(event, "turn_id", "?")
+                )
+                debug_log.write(_dim(
+                    f"[{_ts()}] turn_summary turn_id={turn_id} "
+                    f"text_deltas={self.text_delta_count} "
+                    f"text_bytes={self.text_byte_count} "
+                    f"thinking_deltas={self.thinking_delta_count} "
+                    f"thinking_bytes={self.thinking_byte_count} "
+                    f"elapsed_ms={elapsed_ms}"
+                ))
+                # Terminal event: finalize the response widget (clear ref so
+                # the next turn mounts a fresh one). The accumulated text is
+                # already rendered as Markdown in the widget — no post-Done
+                # re-render, no double-print.
+                self.text_chunk_buffer = ""
+                self.current_response_widget = None
+                # Terminal labels mount as styled Statics. Tinted per outcome
+                # (Aurora green / Dawn red / Dawn yellow) for at-a-glance
+                # scanning.
                 if isinstance(event, Done):
-                    log.write(RichText(
-                        f"[done] turn_id={event.sse_id.turn_id} model={event.model} "
-                        f"duration={_format_duration_ms(event.duration_ms)} "
-                        f"usage {_format_usage(event.usage, arrow='→')}",
-                        style=_AU_SUCCESS,
+                    transcript.mount(Static(
+                        RichText(
+                            f"[done] turn_id={event.sse_id.turn_id} "
+                            f"model={event.model} "
+                            f"duration={_format_duration_ms(event.duration_ms)} "
+                            f"usage {_format_usage(event.usage, arrow='→')}",
+                            style=_AU_SUCCESS,
+                        ),
+                        classes="done-label",
                     ))
-                    # v0.8.2: post-Done Markdown body re-render dropped. Pre-
-                    # v0.8.2 the transcript got BOTH the streamed text AND
-                    # the Markdown(response) re-render — same content twice,
-                    # operator-flagged as "double prints". The streamed text
-                    # IS the response now; markdown formatting (bold, lists,
-                    # code) renders as plain text. Matches thinking pane's
-                    # stream-as-content semantics (no post-close re-render).
                 elif isinstance(event, Error):
-                    log.write(RichText(
-                        f"[error] turn_id={event.sse_id.turn_id} code={event.error_code} "
-                        f"message={event.message!r}",
-                        style=_AU_ERROR,
+                    transcript.mount(Static(
+                        RichText(
+                            f"[error] turn_id={event.sse_id.turn_id} "
+                            f"code={event.error_code} message={event.message!r}",
+                            style=_AU_ERROR,
+                        ),
+                        classes="error-label",
                     ))
                 else:  # Cancelled
-                    log.write(RichText(
-                        f"[cancelled] turn_id={event.turn_id} reason={event.reason!r} "
-                        f"partial_message_id={event.partial_message_id}",
-                        style=_AU_WARNING,
+                    transcript.mount(Static(
+                        RichText(
+                            f"[cancelled] turn_id={event.turn_id} "
+                            f"reason={event.reason!r} "
+                            f"partial_message_id={event.partial_message_id}",
+                            style=_AU_WARNING,
+                        ),
+                        classes="cancelled-label",
                     ))
+                transcript.scroll_end(animate=False)
                 return
             if isinstance(event, WorkerPhase):
                 # v0.5.0: telemetry → Debug pane, not transcript.
@@ -360,22 +598,24 @@ class TuiPresenterState:
             # the original event AND a render_error line with the class name only
             # (NO exception message — security clause). Volva F1 fix.
             #
-            # v0.6.0 routing-under-failure preservation — fallback writes go
-            # to the same destination the successful render would have used:
-            #   - ToolStart/ToolResult       → tools_log
-            #   - Thinking                   → thinking_log
-            #   - WorkerPhase/TextBoundary   → debug_log
-            #   - everything else            → log
+            # v0.9.0 routing-under-failure: panes (RichLog) still write Strip
+            # lines; transcript (VerticalScroll) mounts a Static instead.
             if isinstance(event, (ToolStart, ToolResult)):
-                target = tools_log
+                tools_log.write(_plain_label(event))
+                tools_log.write(f"[render_error] {type(exc).__name__}")
             elif isinstance(event, Thinking):
-                target = thinking_log
+                thinking_log.write(_plain_label(event))
+                thinking_log.write(f"[render_error] {type(exc).__name__}")
             elif isinstance(event, (WorkerPhase, TextBoundary)):
-                target = debug_log
+                debug_log.write(_plain_label(event))
+                debug_log.write(f"[render_error] {type(exc).__name__}")
             else:
-                target = log
-            target.write(_plain_label(event))
-            target.write(f"[render_error] {type(exc).__name__}")
+                # Transcript-bound event (Text / Done / Error / Cancelled).
+                transcript.mount(Static(_plain_label(event), classes="error-label"))
+                transcript.mount(
+                    Static(f"[render_error] {type(exc).__name__}", classes="error-label")
+                )
+                transcript.scroll_end(animate=False)
 
 
 class AgentPickerApp(App[str | None]):
@@ -564,11 +804,40 @@ class RatatoskrApp(App[int]):
     }
     /* v0.6.5: thinking-current Static removed; thinking now streams
        directly into thinking-log so the whole pane scrolls naturally. */
-    #transcript {
+    /* v0.9.0: transcript is a VerticalScroll container holding dynamically
+       mounted Statics + Markdown widgets per turn. Live Markdown rendering
+       replaces the v0.8.x RichLog approach which couldn't render Markdown
+       in-flight (only on Done as a re-render → double-print bug). */
+    #transcript-scroll {
         height: 1fr;
         background: $background;
         padding: 0 1;
     }
+    /* Per-turn mounted widgets carry id-prefix conventions:
+       - .turn-header   "── turn N ──" (dim)
+       - .prompt-echo   "❯ user input"  (aurora bright cyan)
+       - .response-md   Markdown(accumulated_text) — updated live
+       - .done-label    "[done] turn_id=…" (aurora green)
+       - .error-label   "[error] …"        (dawn red)
+       - .cancelled-label "[cancelled] …"  (dawn yellow)
+    */
+    .turn-header {
+        height: auto;
+        padding: 0 1;
+        color: $au-dark-60;
+    }
+    .prompt-echo {
+        height: auto;
+        padding: 0 1;
+    }
+    .response-md {
+        height: auto;
+        padding: 0 1;
+    }
+    .done-label, .error-label, .cancelled-label {
+        height: auto;
+        padding: 0 1;
+    }
     /* 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. */
@@ -618,6 +887,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]] = [
@@ -628,6 +911,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"
@@ -656,6 +940,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.
@@ -670,7 +959,12 @@ class RatatoskrApp(App[int]):
         # work without widget-level markup=True.
         with Horizontal(id="main-row"):
             with Vertical(id="left-column"):
-                yield RichLog(id="transcript", wrap=True, markup=False, highlight=False)
+                # v0.9.0: transcript is a VerticalScroll holding per-turn
+                # mounted widgets (turn header, prompt echo, response Markdown,
+                # done label). Live Markdown rendering happens via Static
+                # widgets holding `Markdown` Renderables, updated as Text
+                # deltas arrive.
+                yield VerticalScroll(id="transcript-scroll")
                 yield Input(id="prompt", placeholder="Type a message and press Enter")
             with Vertical(id="right-column"):
                 with TabbedContent(id="side-panes"):
@@ -692,6 +986,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")
@@ -731,22 +1033,115 @@ class RatatoskrApp(App[int]):
         )
         self.state = "idle"
         self._set_hint(self.HINT_IDLE)
+        # v0.10.0: startup audit so the debug pane carries a complete
+        # session bootstrap line (server URL, agent, end_user_id, raw flag,
+        # session tail) before the first turn fires.
+        self._audit(
+            f"app_mounted server={self.args.server_url} agent_id={self.agent_id!r} "
+            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: Write `── turn N ──` Rule headers across every pane so
-        operators can visually correlate sections during cross-pane
-        debugging. Called from `_stream_turn_worker` on first event of
-        each new turn (idempotent per turn via active_turn_id guard).
+        """v0.6.0: turn-ID headers across every pane for cross-pane
+        correlation. v0.9.0: transcript is a VerticalScroll; mounts a
+        Static with rule-style text instead of writing a Rule Renderable
+        to RichLog. Other panes still use RichLog.write(Rule).
         """
         from rich.rule import Rule
+        from rich.text import Text as RichText
 
         title = f"turn {turn_id}"
         rule = Rule(title=title, style=_AU_DEMOTED)
         try:
-            self.query_one("#transcript", RichLog).write(rule)
+            # Transcript (VerticalScroll): mount a styled Static.
+            transcript = self.query_one("#transcript-scroll", VerticalScroll)
+            transcript.mount(
+                Static(
+                    RichText(f"── turn {turn_id} ──", style=_AU_DEMOTED),
+                    classes="turn-header",
+                )
+            )
+            # Other panes (RichLog): write the Rule Renderable.
             self.query_one("#tools-log", RichLog).write(rule)
             self.query_one("#debug-log", RichLog).write(rule)
             self.query_one("#thinking-log", RichLog).write(rule)
+            transcript.scroll_end(animate=False)
         except Exception:
             # Defensive: widget tree may be tearing down — never let a
             # turn-header write block the SSE consumer.
@@ -761,13 +1156,51 @@ class RatatoskrApp(App[int]):
             # Widget may be gone during shutdown; ignore.
             pass
 
+    def _audit(self, line: str) -> None:
+        """Write a timestamped audit line to the debug pane.
+
+        v0.10.0: shared sink for app-level events that don't pass through
+        the presenter — state transitions, worker spawn/cancel, cancel POST
+        lifecycle, startup probes. The presenter's per-event audit lives at
+        `_audit_line()`; this is its app-side counterpart.
+        """
+        try:
+            from rich.text import Text as RichText
+            self.query_one("#debug-log", RichLog).write(
+                RichText(f"[{_ts()}] {line}", style=_AU_DEMOTED)
+            )
+        except Exception:
+            # Widget may not exist yet (pre-mount) or be tearing down.
+            pass
+
+    def _transition(
+        self, new_state: Literal["idle", "streaming", "cancelling"], reason: str
+    ) -> None:
+        """Set self.state with debug-pane audit log.
+
+        Every state machine transition flows through here so the debug pane
+        carries a complete idle→streaming→cancelling→idle timeline with the
+        triggering reason. Cheap; safe to call from any context.
+        """
+        old = self.state
+        self.state = new_state
+        if old != new_state:
+            self._audit(f"state {old} → {new_state} reason={reason}")
+
     async def on_input_submitted(self, event: Input.Submitted) -> None:
-        """Echo user prompt, spawn stream worker; busy notice if not idle."""
+        """Echo user prompt, spawn stream worker; busy notice if not idle.
+
+        v0.9.0: prompt echo mounts as a Static in the transcript VerticalScroll
+        (was log.write to RichLog).
+        """
         if event.input.id != "prompt":
             return
-        log = self.query_one("#transcript", RichLog)
+        transcript = self.query_one("#transcript-scroll", VerticalScroll)
         if self.state != "idle":
-            log.write("[busy] turn in flight; input ignored")
+            transcript.mount(
+                Static("[busy] turn in flight; input ignored", classes="error-label")
+            )
+            transcript.scroll_end(animate=False)
             event.input.value = ""
             return
         content = event.input.value.strip()
@@ -776,54 +1209,78 @@ class RatatoskrApp(App[int]):
         # v0.4.1 retheme: operator's voice gets Australis bright cyan so it
         # stands out against the default-foreground assistant text below it.
         from rich.text import Text as RichText
-        log.write(RichText(f"❯ {content}", style=_AU_USER_ECHO))  # noqa: RUF001
+        transcript.mount(
+            Static(
+                RichText(f"❯ {content}", style=_AU_USER_ECHO),  # noqa: RUF001
+                classes="prompt-echo",
+            )
+        )
+        transcript.scroll_end(animate=False)
         event.input.value = ""
-        self.state = "streaming"
+        self._transition("streaming", "input_submitted")
+        self._audit(f"worker_spawn content_len={len(content)}")
         self._set_hint(self.HINT_STREAMING)
         self.stream_worker = self.run_worker(
             self._stream_turn_worker(content), exclusive=True
         )
 
     async def _stream_turn_worker(self, content: str) -> None:
-        """Drive stream_turn, render events via TuiPresenterState (issue #12)."""
+        """Drive stream_turn, render events via TuiPresenterState.
+
+        v0.9.0: transcript is a VerticalScroll; the presenter's `transcript`
+        argument is the container, and the presenter mounts Static / Markdown-
+        backed widgets directly. Wire-error labels mount as `error-label`
+        Statics into the transcript-scroll.
+        """
         assert self.state == "streaming"
         assert self.client is not None
         assert content
-        log = self.query_one("#transcript", RichLog)
+        transcript = self.query_one("#transcript-scroll", VerticalScroll)
         tools_log = self.query_one("#tools-log", RichLog)
         debug_log = self.query_one("#debug-log", RichLog)
         thinking_log = self.query_one("#thinking-log", RichLog)
         presenter = TuiPresenterState()
+
+        def _mount_wire_error(label: str) -> None:
+            try:
+                transcript.mount(Static(label, classes="error-label"))
+                transcript.scroll_end(animate=False)
+            except Exception:
+                pass
+
         try:
             async for event in stream_turn(self.client, self.session_id, content):
                 if self.active_turn_id is None:
                     self.active_turn_id = event.sse_id.turn_id
-                    # v0.6.0: turn-ID headers across all panes so the
-                    # operator can visually correlate sections during
-                    # cross-pane debugging.
                     self._write_turn_headers(self.active_turn_id)
                 presenter.render(
                     event,
-                    log=log,
+                    transcript=transcript,
                     tools_log=tools_log,
                     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
         except SseConnectFailed as exc:
-            log.write(f"[sse_connect_failed] status={exc.status} body={exc.body!r}")
+            self._audit(f"sse_connect_failed status={exc.status} body={exc.body!r:.120}")
+            _mount_wire_error(f"[sse_connect_failed] status={exc.status} body={exc.body!r}")
         except SseConnectionDropped as exc:
-            log.write(f"[connection_dropped] last_seen={exc.last_seen_sse_id}")
+            self._audit(f"connection_dropped last_seen={exc.last_seen_sse_id}")
+            _mount_wire_error(f"[connection_dropped] last_seen={exc.last_seen_sse_id}")
         except MalformedSseId as exc:
-            log.write(f"[malformed_sse_id] raw={exc.raw!r}")
+            self._audit(f"malformed_sse_id raw={exc.raw!r}")
+            _mount_wire_error(f"[malformed_sse_id] raw={exc.raw!r}")
         except MalformedSseData as exc:
-            log.write(f"[malformed_sse_data] raw={exc.raw!r}")
+            self._audit(f"malformed_sse_data raw={exc.raw!r:.120}")
+            _mount_wire_error(f"[malformed_sse_data] raw={exc.raw!r}")
         except TurnIdFlip as exc:
-            log.write(f"[turn_id_flip] expected={exc.established} got={exc.got}")
+            self._audit(f"turn_id_flip expected={exc.established} got={exc.got}")
+            _mount_wire_error(f"[turn_id_flip] expected={exc.established} got={exc.got}")
         finally:
-            self.state = "idle"
+            self._transition("idle", "worker_finally")
             self.active_turn_id = None
             self._set_hint(self.HINT_IDLE)
 
@@ -835,26 +1292,35 @@ class RatatoskrApp(App[int]):
         """Two-stage Ctrl-C state machine per INV-003."""
         assert self.state in ("idle", "streaming", "cancelling")
         if self.state == "idle":
+            self._audit("ctrl_c state=idle action=exit code=0")
             self.exit(0)
         elif self.state == "streaming":
             if self.active_turn_id is None:
+                self._audit("ctrl_c state=streaming active_turn_id=None action=force_exit code=3")
                 if self.stream_worker is not None:
                     self.stream_worker.cancel()
                 self.exit(3)
                 return
-            self.state = "cancelling"
+            self._audit(f"ctrl_c state=streaming turn_id={self.active_turn_id} action=cancel_post")
+            self._transition("cancelling", "ctrl_c_cancel_post_issued")
             self._set_hint(self.HINT_CANCELLING)
-            log = self.query_one("#transcript", RichLog)
+            transcript = self.query_one("#transcript-scroll", VerticalScroll)
             self.run_worker(
-                _cancel_via_sse(self.client, self.session_id, self.active_turn_id, log=log)
+                _cancel_via_sse(
+                    self.client, self.session_id, self.active_turn_id,
+                    transcript=transcript,
+                    audit=self._audit,
+                )
             )
         elif self.state == "cancelling":
+            self._audit("ctrl_c state=cancelling action=force_exit code=3")
             if self.stream_worker is not None:
                 self.stream_worker.cancel()
             self.exit(3)
 
     def action_quit(self) -> None:
         """Ctrl-D — immediate exit regardless of state."""
+        self._audit(f"ctrl_d state={self.state} action=exit code=0")
         if self.stream_worker is not None and not self.stream_worker.is_finished:
             self.stream_worker.cancel()
         self.exit(0)
@@ -880,6 +1346,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.
@@ -995,12 +1466,33 @@ async def _cancel_via_sse(
     session_id: str,
     turn_id: int,
     *,
-    log: RichLog,
+    transcript: VerticalScroll,
+    audit: "Callable[[str], None] | None" = None,
 ) -> None:
-    """Fire-and-forget cancel; never raises (mirrors cli._cancel_and_log; #3 INV-009)."""
+    """Fire-and-forget cancel; never raises (mirrors cli._cancel_and_log; #3 INV-009).
+
+    v0.9.0: mounts a `[cancel_failed]` Static into the transcript-scroll
+    container on failure (was log.write to RichLog).
+    v0.10.0: optional `audit` callback (RatatoskrApp._audit) receives one
+    line on POST issue + one on POST result, so the debug pane carries the
+    full cancel lifecycle. Defaults to no-op for legacy callers.
+    """
     assert client is not None
     assert isinstance(turn_id, int) and turn_id > 0
+    if audit is not None:
+        audit(f"cancel_post issued session_id={session_id} turn_id={turn_id}")
     try:
         await cancel_turn(client, session_id, turn_id)
+        if audit is not None:
+            audit(f"cancel_post ok turn_id={turn_id}")
     except (CancelFailed, CancelTurnNotFound, CancelAlreadyCompleted, httpx.RequestError) as exc:
-        log.write(f"[cancel_failed] {type(exc).__name__}: {exc}")
+        if audit is not None:
+            audit(f"cancel_post failed turn_id={turn_id} {type(exc).__name__}: {exc!s:.120}")
+        try:
+            transcript.mount(Static(
+                f"[cancel_failed] {type(exc).__name__}: {exc}",
+                classes="error-label",
+            ))
+            transcript.scroll_end(animate=False)
+        except Exception:
+            pass

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

@@ -5,6 +5,7 @@ import pytest
 import respx
 
 from ratatoskr.sse_client import (
+    AffectUpdate,
     CancelAlreadyCompleted,
     Cancelled,
     CancelResult,
@@ -877,3 +878,80 @@ class TestEmptyDataSkipped:
         assert exc_info.value.raw == "x" * 200
         # Exception message also only contains the truncated form
         assert "x" * 5000 not in str(exc_info.value)
+
+
+class TestAffectUpdate:
+    """Worldtree #204 / v0.28.0 — persona-state observability SSE event.
+
+    Two emissions per qualifying turn (persona-enabled agent, non-ephemeral
+    session): `status: "current"` at turn start with full snapshot, then
+    `status: "scheduled"` near turn end (lightweight, no snapshot).
+
+    See docs/conversation-api-spec.md § affect_update.
+    """
+
+    @respx.mock
+    async def test_current_status_parsed_with_snapshot(self) -> None:
+        """current_status_parsed_with_snapshot [tracer]: status=current carries
+        the full snapshot dict; AffectUpdate.snapshot is populated with the
+        nested PAD / dominant_emotion / emotions_active fields.
+        """
+        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",
+        }
+        stream = _sse_chunk(
+            "42:1",
+            {
+                "type": "affect_update",
+                "status": "current",
+                "turn_id": 42,
+                "snapshot": snapshot,
+            },
+        ) + _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")]
+        affect = events[0]
+        assert isinstance(affect, AffectUpdate)
+        assert affect.status == "current"
+        assert affect.turn_id == 42
+        assert affect.snapshot == snapshot
+        assert affect.sse_id == SseId(42, 1)
+
+    @respx.mock
+    async def test_scheduled_status_parsed_no_snapshot(self) -> None:
+        """scheduled_status_parsed_no_snapshot [trace]: status=scheduled carries
+        no snapshot field; AffectUpdate.snapshot is None.
+        """
+        stream = (
+            _sse_chunk("42:1", {"type": "text", "content": "x"})
+            + _sse_chunk(
+                "42:2",
+                {"type": "affect_update", "status": "scheduled", "turn_id": 42},
+            )
+            + _sse_chunk("42:3", _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")]
+        affect = next(e for e in events if isinstance(e, AffectUpdate))
+        assert affect.status == "scheduled"
+        assert affect.turn_id == 42
+        assert affect.snapshot is None
+        assert affect.sse_id == SseId(42, 2)

uv.lock

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