Compare commits

...

2 Commits

Author SHA1 Message Date
vh 44138590ad 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.
2026-05-25 19:13:47 -07:00
vh d516537b08 feat(sessions): get_persona_state client + persona error taxonomy (v0.12.0)
Adds the read-side half of Worldtree #204's persona-state observability
surface. Pairs with v0.11.0's AffectUpdate SSE event — together they
let a consumer hydrate a persona pane on session-open (this GET) and
keep it live as turns fire (the SSE event).

Public surface:
- `get_persona_state(client, agent_id) -> dict[str, Any]` — GET
  /agents/{agent_id}/persona_state, returns the same `snapshot` dict
  shape as AffectUpdate.snapshot
- New exception types mapped from the spec's documented 4xx error_codes:
  - `PersonaNotConfigured` (404 persona_not_configured) — agent has
    no persona surface (domari, muninn, all Tier 3 in Phase 2.0)
  - `AgentNotAvailable` (404 agent_not_available) — unknown agent_id
  - `AuthScopeDenied` (403 auth_scope_denied) — key lacks the
    requested scope (persona.read here; reusable for future scoped
    endpoints)
- Other non-2xx falls through to the existing SessionApiFailed
  precedent so novel failure modes aren't silently absorbed

Tests: 6 new cases covering happy snapshot return, each typed 4xx
sub-code, unknown 404 fall-through, and 5xx SessionApiFailed parity.

Not yet consumed: TUI persona surface (Persona TabPane / sticky
header line). UX shape pending operator direction — step 3.
2026-05-25 18:55:12 -07:00
6 changed files with 586 additions and 8 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "ratatoskr"
version = "0.11.0"
version = "0.13.0"
description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard"
readme = "README.md"
requires-python = ">=3.12"
+82
View File
@@ -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)
+206 -6
View File
@@ -35,9 +35,13 @@ 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 (
@@ -187,6 +191,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.
@@ -280,6 +357,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 +370,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(
@@ -331,13 +415,20 @@ 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.7.1: Thinking deltas coalesce by newline before flushing.
# Worldtree emits Thinking events at token granularity; per-delta
@@ -796,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]] = [
@@ -806,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"
@@ -834,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.
@@ -875,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")
@@ -922,6 +1041,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 +1260,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 +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.
+109
View File
@@ -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
+187
View File
@@ -934,6 +934,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 +1254,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
Generated
+1 -1
View File
@@ -968,7 +968,7 @@ wheels = [
[[package]]
name = "ratatoskr"
version = "0.11.0"
version = "0.13.0"
source = { editable = "." }
dependencies = [
{ name = "httpx" },