diff --git a/persistent-memory.md b/persistent-memory.md index 66c56be..79a8ded 100644 --- a/persistent-memory.md +++ b/persistent-memory.md @@ -1,6 +1,6 @@ # Persistent memory — ratatoskr -_Last updated: 2026-05-21_ +_Last updated: 2026-05-22_ This file captures durable intent and supporting evidence (goals, decisions, foot-gun warnings, in-flight state) across context resets. Read it at session @@ -34,39 +34,48 @@ separate dev team rather than an in-tree Worldtree tool. _As of 2026-05-21:_ -**Status: v0 milestone landed.** Four issues complete end-to-end against -their contracts: `ratatoskr.sse_client` (#1), `ratatoskr.sessions` (#2), -`ratatoskr.cli --send` non-interactive presenter (#3), `ratatoskr.tui` shell -(#4). 164/164 tests GREEN (42 sse_client + 19 sessions + 58 cli + 43 tui + -2 boundary/metadata); ruff clean; all four issue contracts drift-check -clean. `ratatoskr` console-script entry point dispatches `--send` → cli -(stdout) and bare invocation → TUI (lazy textual import preserves issue #3 -INV-001). All work pushed to `origin/main`. +**Status: v0 milestone + issue #7 (mid-stream robustness) landed.** Four +core issues complete end-to-end (`sse_client` #1, `sessions` #2, `cli` +#3, `tui` #4) + one robustness fix on top: `MalformedSseData` exception ++ empty-data skip in `_iter_events` (issue #7, fixes mid-stream +JSONDecodeError crash discovered during long TUI conversations). +172/172 tests GREEN (46 sse_client + 19 sessions + 60 cli + 44 tui + +2 boundary/metadata + 1 from #7 last-seen probe); ruff clean; all five +issue contracts (#1, #3, #4, #5, #7) drift-check clean. `--send` validated end-to-end against personal Worldtree -(`http://10.250.50.152:8081`, mimir agent on qwen3.6-35-a3b). Smoke key -lives in gitignored `env.sh` (delivered by infra-ops 2026-05-21; see -[[personal-worldtree-smoke-target]] in auto-memory). +(`http://10.250.50.152:8081`, mimir on qwen3.6-35-a3b). Smoke key in +gitignored `env.sh` (delivered by infra-ops 2026-05-21; see +[[personal-worldtree-smoke-target]] in auto-memory). Long-conversation +smoke 2026-05-22 (3193-token completion, 374980-token context, 50s +streaming) confirmed empty-data frames are silently skipped — the +original 2026-05-22 crash unreproducible after fix. -**Nothing in flight.** Branch: `main` (clean). Remote: `origin → -git@gitea.phasefinal.com:vh/ratatoskr.git`. +**In-flight: issue #5 (`--end-user-id` flag for per-user agents like +lofn).** Contract drafted + parked (untracked at +`docs/contracts/issues/5.contract.md`); Volva paraphrase + TDD + smoke +pending. Issue #6 (TUI startup error visibility) also filed but +unscaffolded. Branch: `main` (clean apart from #5 contract). Remote: +`origin → git@gitea.phasefinal.com:vh/ratatoskr.git`. **Next natural moves:** -1. **TUI smoke (operator-side)** — `source env.sh && uv run ratatoskr --new - --agent mimir` from an interactive terminal. Validates the Textual app - lifecycle + post-Done markdown re-render end-to-end. Needs a TTY which - CC sessions don't have. -2. **Side-pane issues** — design-brief §5 lists 5 side panes (Persona, - Tools, AdminEvents, BifrostState, ServerLog). Each gets its own issue - + contract + TDD pass on top of the shell. Persona is the natural first - (file-tail of `persona.log` — cheap; no new Worldtree wire). -3. **Recorded SSE snapshot fixtures** from a running Worldtree. `--send - --new` redirected to a fixture file IS the recording probe — capture - outputs to `tests/snapshots/` for replay-based regression coverage. - The smoke proved the recording probe works. -4. **Startup session picker** — design-brief §4 `DataTable` of `GET - /sessions`. Modest scope; pairs naturally with the TUI shell. +1. **Resume issue #5 cadence** — `--end-user-id` flag for per-user agents + (lofn). Contract drafted + parked; next steps: Volva paraphrase, TDD, + smoke against lofn. Unblocks the Persona/Vili agent family. +2. **Issue #6 (TUI startup error visibility)** — filed, unscaffolded. + Restructure: move session-create out of `on_mount` into `run_tui` + pre-App.run() so errors print to real stderr (not the alt-screen that + tears down before user can read them). Independent of #5. +3. **TUI smoke (operator-side)** — `source env.sh && uv run ratatoskr + --new --agent mimir` from an interactive terminal. Validates Textual + app lifecycle + post-Done markdown re-render end-to-end. Needs a TTY + which CC sessions don't have. +4. **Side-pane issues** — design-brief §5 lists 5 side panes (Persona, + Tools, AdminEvents, BifrostState, ServerLog). Persona is the natural + first (file-tail of `persona.log`). +5. **Recorded SSE snapshot fixtures** from a running Worldtree. + `--send --new > fixture` IS the recording probe. ## Recent decisions @@ -89,6 +98,8 @@ decision. Captures rationale that won't be obvious from code alone. - `[2026-05-21]` **Default issue-tracker labels seeded** (17 total). Sleipnir gating, triage, type, resolution, Ratatoskr-specific area labels (sse-client, tui, cli, observability). - `[2026-05-21]` **Volva paraphrase + code-review across all 4 issues — calibration consistent.** Paraphrase rounds flag 3-5 contract ambiguities per issue; code-review rounds flag 3-8 code-vs-contract drifts after TDD-passing implementation. Hit rates: #1 paraphrase 3-of-5 amended / code-review 4 findings; #2 3-of-5 / 3 findings; #3 5-of-5 / 5 findings; #4 5-of-5 / 8 findings. The post-TDD code-review consistently catches three classes of gap the test-author's hypotheses don't cover: PRE-assertion boundary drift, exception-payload truncation / never-rendered-to-user observability misses, and "tested the state but not whether the user can see it" gaps (issue #4's primary finding: TUI footer state stored but never rendered to a visible widget — same-model TDD would systematically miss this). - `[2026-05-21]` **Manual smoke is load-bearing — found a real defect tests couldn't.** First wire-level smoke against personal Worldtree (post-TDD, post-Volva-code-review on #4) revealed httpx's default 5s read timeout killed the SSE connection mid-stream during mimir's thinking phase (~30s LLM latency >> 5s read timeout). The unit/contract test infrastructure (respx-mocked SSE wire) doesn't model real LLM latency, so the gap was invisible at the test layer. Fix: caller-owned `httpx.AsyncClient` constructed with `timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)`; defense in depth: `sse_client.stream_turn` ERROR_ROUTING catches `httpx.ReadTimeout` → `SseConnectionDropped`. Three contracts amended in-place to document the timeout policy. **Lesson: keep manual-smoke step in the per-issue cadence; mock-only validation is insufficient for streaming-against-real-server code.** Re-smoke succeeded: `[done] turn_id=88 model=qwen3.6-35-a3b duration_ms=2351`. Wire-compat envelope (personal v0.16.2 vs ratatoskr's v0.19.0 pin) confirmed end-to-end. +- `[2026-05-22]` **Issues #5/#6/#7 filed: per-user-agent support + TUI-startup-visibility + mid-stream-robustness.** Discovered during 2026-05-22 mimir TUI conversation: long completion (turn 93, 1077 events consumed) crashed with `JSONDecodeError("Expecting value: line 1 column 1 (char 0)")` from `json.loads('')` on an empty-`data:` SSE frame. Diagnosis surfaced #7 (the crash). Earlier same day, `ratatoskr --new --agent lofn` failed with 422 `end_user_id_required` — surfacing #5 (`--end-user-id` flag needed for per-user agents). #6 (TUI alt-screen masks the diagnostic before user can read it) was a corollary observation. All three filed; user reordered to #7 first (highest-impact for daily TUI use). +- `[2026-05-22]` **Issue #7 (`MalformedSseData` + empty-skip) implemented via TDD + Volva-code-reviewed + smoked.** Contract → Volva paraphrase (4 ambiguities, all amended; INV-001 wording tightened around exact `sse.data == ''` rule, ordering-before-id-parse made explicit, test-description bug fixed) → TDD (6 tests, full vertical-slice ordering) → Volva code-review (3 findings — F1 test-gap probing internal `last_sse_id` non-advancement via post-skip drop, F2 contract precision around log-vs-propagate responsibility, F3 cli test tightening for `raw='X'` shape + truncation coverage; all amended) → smoke (3193-token completion against personal Worldtree confirmed clean termination; original crash unreproducible). **Calibration milestone: issue #7 is the first issue with zero drift findings from Volva code-review** — TDD caught all runtime behavior cleanly. The 3 findings were assertion-precision and architectural-correctness-of-wording, not behavioral. Hypothesis: the tighter the contract spec + the smaller the code surface, the more Volva's role shifts from "catch behavioral drift" to "tighten observability + wording". Calibration table now: #1 (4 findings, 3 drift + 1 test-gap), #2 (3, 1+1+1 precision), #3 (5, 3+1+1), #4 (8, 5+2+1), #7 (3, 0 drift + 2 test-gap + 1 precision). _For per-issue TDD implementation notes, Volva findings, and contract amendments, see the git log (commits `9703eb2..61c3941` carry the full per-issue trail with structured commit messages)._ @@ -104,3 +115,4 @@ defense against re-attempting the same cul-de-sac. - `[2026-05-21]` **RichLog widget with `markup=True`.** Default impulse, but Rich interprets `[xxx]` spans as style markup and silently strips them. Every labeled stderr-style line — `[cancel_failed]`, `[done]`, `[error]`, `[busy]`, `[worker_phase]` — would render as just the content after the bracketed label, breaking the user-visible observability surface. Fix: `markup=False`. The post-Done Markdown rendering still works because `rich.markdown.Markdown` is a Renderable that ignores widget-level markup setting. Don't flip back to `markup=True` without first renaming every labeled-line format away from `[bracket]` notation. - `[2026-05-21]` **Querying `self.query_one("#transcript", RichLog)` from inside a Textual `run_worker` coroutine.** Failed initially with `NoMatches` because the worker fires before the test's `pilot.pause()` allows the Input.Submitted handler to fully dispatch (and thus the widget tree to settle). Initial reactive fix: widen worker signature to take `log` as a parameter (passed from the handler). Volva code-review flagged this as contract drift (signature didn't match spec). Reverted to single-param signature. The real fix was test-side: add `await pilot.pause()` between `inp.action_submit()` and the polling loop in `_submit_and_wait` so the handler finishes dispatching before the worker reads the widget tree. Don't widen worker signatures to dodge test timing. - `[2026-05-21]` **TUI session-identity rendering via `self.sub_title` + `self.hint` plain attributes.** Stored state but never rendered to a visible widget. The contract's "session-identity-always-visible" invariant was satisfied at the state-attribute level but not the user-visible-widget level. Tests asserted the attributes (which passed); Volva code-review flagged the gap. Fix: dedicated `Static(id="identity")` + `Static(id="hint")` widgets in compose; `_set_hint()` helper mirrors state → widget. Calibration evidence for the "TDD catches state, code-review catches whether the user can see it" pattern. +- `[2026-05-22]` **`json.loads(sse.data)` unguarded against empty data.** `_iter_events` unconditionally called `json.loads` on every dispatched `ServerSentEvent`. When `httpx_sse` surfaced a frame with `id:` present but `data:` empty (a known library-vs-spec divergence — RFC says don't dispatch; httpx_sse is permissive), `json.loads('')` raised `JSONDecodeError` → propagated through Textual's worker → app crash. Crashed mimir conversation at turn 93/seq 1078 after 1077 successful events. Fix: `if sse.data == '': continue` BEFORE `_parse_sse_id` (empty-data event with a malformed id is still a keepalive — don't reorder). Non-empty malformed data raises new `MalformedSseData(raw[:200])`. Don't reintroduce unconditional `json.loads(sse.data)`; always pre-check for the empty case. diff --git a/src/ratatoskr/cli.py b/src/ratatoskr/cli.py index 1c4366c..5d15e60 100644 --- a/src/ratatoskr/cli.py +++ b/src/ratatoskr/cli.py @@ -24,6 +24,7 @@ from ratatoskr.sse_client import ( Done, Error, Event, + MalformedSseData, MalformedSseId, SseConnectFailed, SseConnectionDropped, @@ -230,6 +231,9 @@ async def _run_turn( except MalformedSseId as exc: stderr.write(f"[malformed_sse_id] raw={exc.raw!r}\n") return 22 + except MalformedSseData as exc: + stderr.write(f"[malformed_sse_data] raw={exc.raw!r}\n") + return 22 except TurnIdFlip as exc: stderr.write(f"[turn_id_flip] expected={exc.established} got={exc.got}\n") return 22 diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index 0d654c0..0c99495 100644 --- a/src/ratatoskr/sse_client.py +++ b/src/ratatoskr/sse_client.py @@ -132,6 +132,19 @@ class MalformedSseId(Exception): self.raw = raw[:64] +class MalformedSseData(Exception): + """Raised when an SSE event's `data:` payload is non-empty but fails JSON parsing. + + Sibling to MalformedSseId — both are wire-level protocol violations from the + server. Empty-data frames are skipped silently per issue #7 INV-001; this + exception is for non-empty payloads that fail json.loads. + """ + + def __init__(self, raw: str) -> None: + super().__init__(f"malformed SSE data: {raw[:200]!r}") + self.raw = raw[:200] + + class TurnIdFlip(Exception): """Raised when an SSE event's turn_id doesn't match the established/expected turn.""" @@ -290,6 +303,12 @@ async def _iter_events( terminal_seen = False try: async for sse in event_source.aiter_sse(): + # Issue #7 INV-001: empty-data frames are keepalives — skip silently. + # ORDERING: this branch fires BEFORE _parse_sse_id; an empty-data event + # with a malformed id is silently swallowed (intentional — a keepalive + # with a bad id is still a keepalive). Don't reorder. + if sse.data == "": + continue try: sse_id = _parse_sse_id(sse.id) except ValueError as exc: @@ -298,7 +317,11 @@ async def _iter_events( established = sse_id.turn_id elif sse_id.turn_id != established: raise TurnIdFlip(established=established, got=sse_id.turn_id) - event = _envelope_for_type(json.loads(sse.data), sse_id=sse_id) + try: + body = json.loads(sse.data) + except json.JSONDecodeError as exc: + raise MalformedSseData(raw=sse.data) from exc + event = _envelope_for_type(body, sse_id=sse_id) yield event last_sse_id = sse_id if isinstance(event, (Done, Error, Cancelled)): diff --git a/src/ratatoskr/tui.py b/src/ratatoskr/tui.py index 9733c77..3c478a7 100644 --- a/src/ratatoskr/tui.py +++ b/src/ratatoskr/tui.py @@ -22,6 +22,7 @@ from ratatoskr.sse_client import ( Done, Error, Event, + MalformedSseData, MalformedSseId, SseConnectFailed, SseConnectionDropped, @@ -211,6 +212,8 @@ class RatatoskrApp(App[int]): log.write(f"[connection_dropped] last_seen={exc.last_seen_sse_id}") except MalformedSseId as exc: log.write(f"[malformed_sse_id] raw={exc.raw!r}") + except MalformedSseData as exc: + log.write(f"[malformed_sse_data] raw={exc.raw!r}") except TurnIdFlip as exc: log.write(f"[turn_id_flip] expected={exc.established} got={exc.got}") finally: diff --git a/tests/test_cli.py b/tests/test_cli.py index 73e9c6b..25b9ed2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -531,6 +531,49 @@ class TestRunTurn: assert exit_code == 22 assert "[malformed_sse_id]" in stderr.getvalue() + @respx.mock + async def test_malformed_sse_data(self) -> None: + """malformed_sse_data [error]: text + non-JSON → exit 22; [malformed_sse_data] raw='...'.""" + stream = ( + _sse_chunk("42:1", {"type": "text", "content": "x"}) + + b"id: 42:2\ndata: not-json\n\n" + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 22 + out = stderr.getvalue() + # Exact label + `raw='X'` shape — the stderr format the contract specifies + assert "[malformed_sse_data] raw='not-json'" in out + + @respx.mock + async def test_malformed_sse_data_truncation(self) -> None: + """malformed_sse_data_truncation [security]: 5000-char bad data → raw truncated.""" + huge_bad = "x" * 5000 + stream = ( + _sse_chunk("42:1", {"type": "text", "content": "x"}) + + f"id: 42:2\ndata: {huge_bad}\n\n".encode() + ) + respx.post("https://w.example/sessions/s-1/messages").mock( + return_value=_sse_resp(stream) + ) + sigint = asyncio.Event() + stdout, stderr = io.StringIO(), io.StringIO() + async with httpx.AsyncClient(base_url="https://w.example") as client: + exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) + assert exit_code == 22 + out = stderr.getvalue() + assert "[malformed_sse_data]" in out + # MalformedSseData.raw was truncated to 200 chars at the exception layer; + # presenter's `repr()` rendering of that 200-char string carries through. + # Full 5000-char payload MUST NOT appear in stderr. + assert "x" * 5000 not in out + assert "x" * 200 in out # the truncated form IS in the rendered raw='...' + @respx.mock async def test_turn_id_flip(self) -> None: """turn_id_flip [error]: …""" diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py index 6d7148e..e93359f 100644 --- a/tests/test_sse_client.py +++ b/tests/test_sse_client.py @@ -694,3 +694,145 @@ class TestCancelTurn: assert first.cancelled is True with pytest.raises(CancelAlreadyCompleted): await cancel_turn(client, "s1", 42) + + +# ============================================================================ +# Issue #7: empty-data skip + MalformedSseData raise +# ============================================================================ + + +def _sse_empty_chunk(sse_id: str) -> bytes: + """SSE frame with id but empty data (server-emitted keepalive shape).""" + return f"id: {sse_id}\ndata:\n\n".encode() + + +def _sse_raw_chunk(sse_id: str, raw_data: str) -> bytes: + """SSE frame with id + arbitrary raw data (for testing malformed JSON).""" + return f"id: {sse_id}\ndata: {raw_data}\n\n".encode() + + +class TestEmptyDataSkipped: + @respx.mock + async def test_empty_data_skipped(self) -> None: + """empty_data_skipped [trace]: 4 frames in, 3 events out; skip preserves last_sse_id.""" + from ratatoskr.sse_client import Done as _Done + from ratatoskr.sse_client import Text as _Text + + stream = ( + _sse_chunk("42:1", {"type": "text", "content": "first"}) + + _sse_empty_chunk("42:2") # ← skipped silently + + _sse_chunk("42:3", {"type": "text", "content": "second"}) + + _sse_chunk("42:4", _DONE_42_6) + ) + respx.post("https://w.example/sessions/s1/messages").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=stream + ) + ) + async with httpx.AsyncClient(base_url="https://w.example") as client: + events = [e async for e in stream_turn(client, "s1", "hi")] + # Exactly 3 events: Text, Text, Done — empty-data frame at 42:2 is invisible + assert len(events) == 3 + assert isinstance(events[0], _Text) and events[0].sse_id == SseId(42, 1) + assert isinstance(events[1], _Text) and events[1].sse_id == SseId(42, 3) + assert isinstance(events[2], _Done) and events[2].sse_id == SseId(42, 4) + # Per INV-003: skip MUST NOT advance through 42:2. The second Text's sse_id + # is (42, 3) — directly verifies the skip didn't bookkeep 42:2. + assert events[1].sse_id.seq == 3, "skip advanced through 42:2" + + @respx.mock + async def test_empty_data_skip_preserves_last_seen_sse_id(self) -> None: + """empty_skip_does_not_advance [trace]: drop-after-empty → last_seen is last real event.""" + from ratatoskr.sse_client import SseConnectionDropped + + # Stream: text(42:1), empty(42:2), then drop. Per INV-003, the skipped + # 42:2 must NOT advance internal last_sse_id. If the consumer caught a + # drop, SseConnectionDropped.last_seen_sse_id should be (42, 1) — the + # last *real* event — NOT (42, 2). + first = _sse_chunk("42:1", {"type": "text", "content": "x"}) + empty = _sse_empty_chunk("42:2") + + class _DropAfterEmpty(httpx.AsyncByteStream): + async def __aiter__(self): # type: ignore[no-untyped-def] + yield first + yield empty + raise httpx.RemoteProtocolError("drop after skip") + + async def aclose(self) -> None: + return None + + respx.post("https://w.example/sessions/s1/messages").mock( + return_value=httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + stream=_DropAfterEmpty(), + ) + ) + async with httpx.AsyncClient(base_url="https://w.example") as client: + with pytest.raises(SseConnectionDropped) as exc_info: + async for _ in stream_turn(client, "s1", "hi"): + pass + assert exc_info.value.last_seen_sse_id == SseId(42, 1), ( + f"skip advanced last_sse_id through 42:2; got {exc_info.value.last_seen_sse_id}" + ) + + @respx.mock + async def test_malformed_data_raises(self) -> None: + """malformed_data_raises [error]: text + bad-JSON → yields Text then MalformedSseData.""" + from ratatoskr.sse_client import MalformedSseData + from ratatoskr.sse_client import Text as _Text + + stream = ( + _sse_chunk("42:1", {"type": "text", "content": "hi"}) + + _sse_raw_chunk("42:2", "not-json") + ) + respx.post("https://w.example/sessions/s1/messages").mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=stream + ) + ) + yielded: list = [] + async with httpx.AsyncClient(base_url="https://w.example") as client: + with pytest.raises(MalformedSseData) as exc_info: + async for e in stream_turn(client, "s1", "hi"): + yielded.append(e) + assert len(yielded) == 1 + assert isinstance(yielded[0], _Text) + assert exc_info.value.raw == "not-json" + + @respx.mock + async def test_whitespace_data_raises(self) -> None: + """whitespace_data_raises [adv]: single-space data → MalformedSseData (NOT skipped).""" + from ratatoskr.sse_client import MalformedSseData + + stream = _sse_raw_chunk("42:1", " ") # single space — non-empty, JSON-invalid + 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: + with pytest.raises(MalformedSseData): + async for _ in stream_turn(client, "s1", "hi"): + pass + + @respx.mock + async def test_malformed_data_truncation(self) -> None: + """malformed_data_truncation [security]: 5000-char bad data → raw truncated to 200.""" + from ratatoskr.sse_client import MalformedSseData + + huge_bad = "x" * 5000 # not JSON; very long + stream = _sse_raw_chunk("42:1", huge_bad) + 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: + with pytest.raises(MalformedSseData) as exc_info: + async for _ in stream_turn(client, "s1", "hi"): + pass + assert len(exc_info.value.raw) == 200 + assert exc_info.value.raw == "x" * 200 + # Exception message also only contains the truncated form + assert "x" * 5000 not in str(exc_info.value) diff --git a/tests/test_tui.py b/tests/test_tui.py index a566604..7580665 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -706,6 +706,29 @@ class TestStreamTurnWorker: assert app.state == "idle" assert any("[connection_dropped]" in str(w) for w in writes) + @respx.mock + async def test_malformed_sse_data_returns_to_idle( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """malformed_sse_data_returns_to_idle [error]: bad-JSON → [malformed_sse_data]; idle.""" + stream = ( + _sse_chunk("42:1", {"type": "text", "content": "x"}) + + b"id: 42:2\ndata: not-json\n\n" + ) + respx.post( + "https://w.example/sessions/s-1existing/messages" + ).mock(return_value=_sse_resp(stream)) + writes = _spy_writes(monkeypatch) + app = RatatoskrApp(_args_existing()) + async with app.run_test() as pilot: + await pilot.pause() + await _submit_and_wait(app, pilot, "x") + assert app.state == "idle" + assert any("[malformed_sse_data]" in str(w) for w in writes) + assert any("not-json" in str(w) for w in writes) + # INV-008: mid-session error does NOT exit the app + assert app.return_value is None + @respx.mock async def test_rendered_event_per_event( self, monkeypatch: pytest.MonkeyPatch