From b9acdc11db1777f3d5055005bca242c25acade3b Mon Sep 17 00:00:00 2001 From: Simon Qi Date: Wed, 26 Aug 2026 20:22:44 +1200 Subject: [PATCH] Read every JSONL record, name the call the dialog holds, and forget a Session that ended (#98) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects in #75's merged Claude-lane code (`37cf18e`), all inside the adapters and none touching a seam or the locked types. **`_parse` split the JSONL on Unicode line separators.** `str.splitlines` also breaks on U+2028, U+2029 and U+0085, and `JSON.stringify` — which writes these files — leaves all three raw inside a string. A record carrying one was cut in two, neither half parsed, and the record was dropped whole; a dropped `tool_result` leaves the call it closes outstanding, which reads as a permanent false `PERMISSION`. Re-measured on 2026-08-26 over the 200 most recent transcripts on this machine: 4 files losing 31 records between them. JSONL is newline-delimited, so the split is on `"\n"` and nothing else. **`stopped_on` paired the transcript's tool with the dialog's `approval_id` without checking they were the same call.** One assistant message may carry several `tool_use` blocks: the transcript reading is held up on the newest outstanding one, the dialog is parked on whichever the far side stopped to ask about, and the handle belongs to the dialog's. Announcing the record's pick beside it would name one tool and carry the user's verdict to another. The hook payload is now the authority on what is parked (advisor, 2026-08-26): its `tool_name` wins, and the record's `detail` is kept only while the record named the *same* tool — naming none is not agreeing, because a record that could not read a tool's name could still summarise its input. #75's precedence is otherwise unchanged: a question wins outright, and a dialog that named no tool contradicts nothing and leaves fill-a-gap-never-overwrite standing. **`forget_session` had no caller**, so nothing ever evicted a dead Session's channel address, window or parsed transcript — records of files measured at 186 MB on this machine, held for the life of a login-start daemon. No seam method and no timer: the adapter that emits `SessionEnded` forgets the target itself, on both lanes, before the sink sees the event. On the Codex side both emitters now go through one `_drop_and_report_ended`, so a third cannot be written that forgets to. ## Against legacy (ADR 0010) - **Splitting on `"\n"` and nothing else: ported.** Legacy split on byte newlines and only those, in both of its readers — the forward one iterates a file object opened `newline=""` (`legacy@1d32845:bridge/transcript.py: 2795-2812`), the reverse one searches for `b"\n"` (`:1936-1961`) — so neither can break on U+2028. The behaviour is legacy's; v2 lost it when the read was re-expressed with `splitlines()`, and this restores it. What stays **dropped** is the error taxonomy around it, with the rest of #74's durable state: legacy raised on a line that would not parse, and v2 skips it. - **Pairing the announcement with the dialog: no legacy counterpart.** Legacy had no second source to rank a transcript against. The `approval_id` and the parked dialog are v2's, from a hook legacy did not have — it scraped the tool name out of an English Notification sentence (`legacy@1d32845:bridge/daemon.py:143-145,213-223,2049-2051`) and had no handle to mis-pair. #75 recorded that whole route as **adapted**; this repairs the adaptation. - **Evicting a Session that ended: no legacy counterpart.** Legacy kept no per-Session transcript cache on this lane at all — it read a Claude transcript only when a stop asked it to, on demand (`legacy@1d32845:bridge/daemon.py:2115-2167`) — so it had nothing to evict and no eviction to port. The cache is v2's, and so was the leak. Co-Authored-By: Claude Opus 5 (1M context) --- .../adapters/agent/claude/adapter.py | 71 ++++++- .../adapters/agent/claude/transcript.py | 11 +- .../adapters/agent/codex/adapter.py | 50 ++++- tests/test_claude_stop_wiring.py | 200 +++++++++++++++++- tests/test_codex_agent.py | 21 ++ 5 files changed, 330 insertions(+), 23 deletions(-) diff --git a/src/gpt_voicecoding/adapters/agent/claude/adapter.py b/src/gpt_voicecoding/adapters/agent/claude/adapter.py index 4b1f28a..4610a35 100644 --- a/src/gpt_voicecoding/adapters/agent/claude/adapter.py +++ b/src/gpt_voicecoding/adapters/agent/claude/adapter.py @@ -90,6 +90,7 @@ RelayReceipt, RelayRoute, ReplyWindow, + SessionEnded, SessionInspection, SessionLifecycle, SessionState, @@ -343,7 +344,12 @@ def reported(self, target: SessionTarget) -> SessionReport | None: return self._reported.get(target) def forget_session(self, target: SessionTarget) -> None: - """Stop holding a route to one Session. The Session itself is untouched.""" + """Stop holding a route to one Session. The Session itself is untouched. + + Called by `_emit` whenever this adapter reports a Session ended, which is + what keeps the caches below bounded by the machine's live Sessions rather + than by everything that has ever registered (#98). + """ self._channels.pop(target, None) self._windows.forget(target) report = self._reported.pop(target, None) @@ -478,10 +484,10 @@ def _overlay( precedence — a decision only the user can supply outranks a permission call beside it (`legacy@1d32845:bridge/transcript.py:1691-1692`) — and a hook holding a dialog open never overrides one. - 2. **A permission read from the transcript keeps its own fields** and - takes the `approval_id` from the dialog, which is the only place a - handle exists at all. It fills a tool name the record did not carry - and never overwrites one it did. + 2. **A permission read from the transcript takes the `approval_id` from + the dialog**, which is the only place a handle exists at all — and + the dialog is then also the authority on which call that handle + belongs to (`_announced_as`). 3. **A dialog with nothing readable behind it is still a stop.** This is the first-turn case: `PermissionRequest` fires when the dialog opens, before the `tool_use` record is flushed. The reference implementation @@ -505,11 +511,9 @@ def _overlay( if dialog is None or found.kind is WaitingKind.QUESTION: return found if found.kind is WaitingKind.PERMISSION: + tool_name, detail = _announced_as(found, dialog) return replace( - found, - tool_name=found.tool_name or dialog.tool_name or None, - detail=found.detail or dialog.detail or None, - approval_id=dialog.approval_id, + found, tool_name=tool_name, detail=detail, approval_id=dialog.approval_id ) _log.info( "the transcript for %s has not flushed the call behind the dialog parked here; " @@ -821,10 +825,59 @@ async def _dial(self, socket_path: Path) -> ChannelConnection: ) def _emit(self, event: AgentEvent) -> None: + """Raise one event upward, and let go of a Session that has ended (#98). + + **The adapter that says a Session ended is the one that forgets it.** + `forget_session` had no caller: it is not on the `AgentAdapter` seam, and + Bridge Core's `_session_ended` only marks state — so on an engine that + starts at login, every Session that ever registered kept its channel + address, its window and its parsed transcript for the life of the + process, and those records are of files measured at 186 MB on this + machine. It is done here rather than behind a new seam method because + the knowledge is already here, and rather than behind a timer because a + death is an observation and not an age. + + Forgetting happens *before* the sink sees the event, so this adapter is + never holding a route to a Session it has already declared ended: a + Relay aimed at one has to fail on the way out rather than be written + into a channel still listed here. + """ + if isinstance(event, SessionEnded): + self.forget_session(event.target) if self._sink is not None: self._sink.emit(event) +def _announced_as(found: WaitingFor, dialog: ApprovalRequest) -> tuple[str | None, str | None]: + """The tool name and detail the announcement uses, when a dialog is parked here. + + **The hook payload is the authority on what is parked** (advisor, 2026-08-26, + #98). One assistant message may carry several `tool_use` blocks; the + transcript reading is held up on the newest outstanding one, while the dialog + is parked on whichever call the far side actually stopped to ask about. The + two can name different tools, and the `approval_id` beside them belongs to + the dialog's — so a reading that kept the transcript's name would announce + one tool and carry the user's verdict to another, which is an Approval + delivered to a call they were never shown. + + So the dialog's `tool_name` wins whenever it has one, and `detail` follows + the name it describes: **the record's summary is kept only while the two + agree on the call.** "Agree" means the record named the same tool, not + merely that it named none — a record that could not read the tool's name + could still summarise its input, and that summary describes whichever call + the record was held up on rather than the one on screen. + """ + if not dialog.tool_name: + # The hook payload named no tool, so it contradicts nothing the record + # said, and #75's fill-a-gap-never-overwrite rule stands whole. + return found.tool_name or None, found.detail or dialog.detail or None + if found.tool_name == dialog.tool_name: + # One call, read twice. The record read the call's whole input, so its + # summary is the fuller of the two and the dialog's fills a gap. + return dialog.tool_name, found.detail or dialog.detail or None + return dialog.tool_name, dialog.detail or None + + def _classify(reply: dict[str, object], request_id: RequestId) -> DeliveryReceipt | None: """What one channel reply says about this attempt, or `None` for "keep waiting".""" kind = reply.get("type") diff --git a/src/gpt_voicecoding/adapters/agent/claude/transcript.py b/src/gpt_voicecoding/adapters/agent/claude/transcript.py index e6aeab5..8bb5f67 100644 --- a/src/gpt_voicecoding/adapters/agent/claude/transcript.py +++ b/src/gpt_voicecoding/adapters/agent/claude/transcript.py @@ -115,9 +115,18 @@ def _parse(text: str) -> tuple[Record, ...]: A line that does not parse is the record being written right now far more often than it is a broken file, so it costs itself and nothing else. Anything that parses to something other than an object is not a record either. + + **Split on `"\n"` and on nothing else.** JSONL is newline-delimited, while + `str.splitlines` also breaks on U+2028, U+2029 and U+0085 — and + `JSON.stringify`, which writes these files, leaves all three raw inside a + string. A record carrying one would be cut in two, neither half would parse, + and "costs itself and nothing else" would quietly cost the whole record: a + dropped `tool_result` leaves the call it closes outstanding, which reads as a + permanent false `PERMISSION`. Measured on 2026-08-26 over the 200 most recent + transcripts on this machine: 4 files, 31 records (#98). """ records: list[Record] = [] - for line in text.splitlines(): + for line in text.split("\n"): if not line.strip(): continue try: diff --git a/src/gpt_voicecoding/adapters/agent/codex/adapter.py b/src/gpt_voicecoding/adapters/agent/codex/adapter.py index 3a942a9..71d165e 100644 --- a/src/gpt_voicecoding/adapters/agent/codex/adapter.py +++ b/src/gpt_voicecoding/adapters/agent/codex/adapter.py @@ -125,8 +125,10 @@ def __init__( #: Prompts already resolved by somebody else, so a late verdict is #: refused rather than answered into a closed request. self._resolved: set[Any] = set() - #: Subscription retries in flight, so none outlives this adapter. - self._retrying: set[asyncio.Task[None]] = set() + #: Work this adapter started off a callback that could not await it — a + #: subscription retry, a connection being given back — held so that none + #: outlives the adapter. + self._background: set[asyncio.Task[None]] = set() self._own = OwnedAppServer( settings=self._settings, socket_path=own_socket_path or default_own_socket_path(self._settings), @@ -180,9 +182,9 @@ async def aclose(self) -> None: a machine with two and not on a machine with nine, which is the kind of bound that passes every test and fails on somebody's Tuesday (#96). """ - for task in list(self._retrying): + for task in list(self._background): task.cancel() - self._retrying.clear() + self._background.clear() watching = list(self._threads.values()) self._threads.clear() await asyncio.gather(*(self._detached(watched) for watched in watching)) @@ -743,7 +745,12 @@ def _heard(self, message: Message) -> None: if isinstance(settings, dict): watched.read_routing(settings) case "thread/closed" | "thread/deleted": - self._emit(SessionEnded(target=watched.target, detail=str(method))) + self._drop_and_report_ended(watched, str(method)) + # And the connection with it: it was opened to watch this + # thread, and there is no thread left to watch. The other + # emitter has no socket to give back — its far side is what + # went away. + self._spawn(self._detached(watched)) def _asked(self, message: Message) -> None: """One server request. Only the permission prompts are ours to hold.""" @@ -819,12 +826,30 @@ def _connection_lost(self, target: SessionTarget, reason: str) -> None: restart, and dialling a dead one in a loop would be this adapter inventing a recovery it cannot perform. """ - watched = self._threads.pop(target, None) + watched = self._threads.get(target) if watched is None: return + self._drop_and_report_ended(watched, reason) + + def _drop_and_report_ended(self, watched: WatchedThread, detail: str) -> None: + """Stop holding one Session, and say that it ended — in that order (#98). + + **The adapter that says a Session ended is the one that forgets it.** + `forget_session` had no caller: it is not on the `AgentAdapter` seam, and + Bridge Core's `_session_ended` only marks state — so anything held for a + Session that ended was held for the life of the process. Both emitters of + `SessionEnded` go through here, so a third one cannot be written that + forgets to. + + The dropping comes first because a Relay aimed at a Session this adapter + has just declared ended must fail on the way out, rather than be written + down a connection still listed here. What each emitter does about the + *connection* differs, and stays with the emitter. + """ + self._threads.pop(watched.target, None) watched.subscribed = False watched.pending.clear() - self._emit(SessionEnded(target=target, detail=reason)) + self._emit(SessionEnded(target=watched.target, detail=detail)) async def _resubscribe(self, watched: WatchedThread) -> None: """Try again to resume a thread that had nothing to resume.""" @@ -834,10 +859,15 @@ async def _resubscribe(self, watched: WatchedThread) -> None: _log.info("could not subscribe to %s: %s", watched.thread_id, unreachable) def _spawn(self, work: Any) -> None: - """Run a retry without letting it outlive the adapter that started it.""" + """Run work off a callback, without letting it outlive this adapter. + + The callers are the notification handlers, which are synchronous and + cannot await: a subscription retry, and the connection given back when + the thread it was opened for is closed. + """ task = asyncio.ensure_future(work) - self._retrying.add(task) - task.add_done_callback(self._retrying.discard) + self._background.add(task) + task.add_done_callback(self._background.discard) def _thread_for(self, thread_id: Any) -> WatchedThread | None: if not isinstance(thread_id, str): diff --git a/tests/test_claude_stop_wiring.py b/tests/test_claude_stop_wiring.py index aabffd6..f690935 100644 --- a/tests/test_claude_stop_wiring.py +++ b/tests/test_claude_stop_wiring.py @@ -14,6 +14,7 @@ import asyncio import json +import os from dataclasses import replace from pathlib import Path from typing import Any @@ -22,18 +23,24 @@ from gpt_voicecoding.adapters.agent.claude import adapter as claude_adapter from gpt_voicecoding.adapters.agent.claude.adapter import ClaudeAgentAdapter, SessionReport +from gpt_voicecoding.adapters.agent.claude.settings import ClaudeSettings from gpt_voicecoding.adapters.agent.claude.transcript import TranscriptReader from gpt_voicecoding.seams.agent import ( + AgentEvent, ApprovalRequest, LaneDiscovery, LaneUnavailable, + ReplyWindowChanged, + SessionEnded, SessionInspection, SessionState, + SessionStopped, WaitingFor, WaitingKind, ) from gpt_voicecoding.seams.identity import AgentKind, SessionTarget -from test_claude_stop_analysis import asked, called, turn +from test_claude_reply_window import Child, say +from test_claude_stop_analysis import asked, called, said, turn SESSION = "d3a776ae-3b60-437d-bc70-ba57a2b280c6" TARGET = SessionTarget(agent=AgentKind.CLAUDE, session_id=SESSION, pid=3538) @@ -102,6 +109,26 @@ def dialog(tool_name: str = "Bash", detail: str = "push the branch") -> Approval return ApprovalRequest(approval_id="a-1", target=TARGET, tool_name=tool_name, detail=detail) +def parallel(*calls: tuple[str, str, Any]) -> dict[str, Any]: + """One assistant record holding several `tool_use` blocks, as parallel calls arrive. + + `called` writes one block, which is the shape the ranking cases used until a + dialog had to be matched against the right one of several. + """ + return { + "type": "assistant", + "isSidechain": False, + "userType": "external", + "message": { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": identifier, "name": tool, "input": tool_input} + for tool, identifier, tool_input in calls + ], + }, + } + + class TestTheRanking: """Four rows. The transcript and the parked dialog are ranked, never merged.""" @@ -124,10 +151,18 @@ def test_a_readable_question_wins_outright(self, tmp_path: Path) -> None: def test_a_permission_read_from_the_record_keeps_its_fields_and_gains_the_handle( self, tmp_path: Path ) -> None: - """The handle is the one thing a transcript can never carry.""" + """The handle is the one thing a transcript can never carry. + + The dialog here names the same call, which is the ordinary case, and + while it does the record's own reading stands: it read the call's input + and the hook payload carries a summary built from a subset of it. This + parked a dialog naming a *different* tool until #98 — the record won + then, and that was the defect, because the `approval_id` beside it was + always the dialog's. + """ adapter = adapter_holding( transcript(tmp_path, [*turn(), called("Edit", "e1", {"file_path": "/tmp/notes.md"})]), - parked=(dialog(tool_name="Bash", detail="something else"),), + parked=(dialog(tool_name="Edit", detail="notes.md"),), ) waiting = adapter.stopped_on(TARGET, ROSTER_WAITING) assert waiting.kind is WaitingKind.PERMISSION @@ -137,6 +172,59 @@ def test_a_permission_read_from_the_record_keeps_its_fields_and_gains_the_handle assert waiting.approval_id == "a-1" assert waiting.caught_up is True + def test_the_dialog_names_the_call_the_verdict_will_reach(self, tmp_path: Path) -> None: + """Parallel calls in one message: the record's pick and the dialog can differ. + + **The hook payload is the authority on what is parked** (advisor, + 2026-08-26, #98). The transcript is held up on the newest outstanding + call, which for several calls written in one message is the last-listed + one; the dialog is parked on whichever the far side actually stopped to + ask about. Announcing the record's pick beside the dialog's + `approval_id` would name one tool and send the user's verdict to + another — an Approval carrying their authority to a call they were never + shown. + """ + adapter = adapter_holding( + transcript( + tmp_path, + [ + *turn(), + parallel( + ("Read", "r1", {"file_path": "/tmp/first"}), + ("Bash", "b1", {"description": "push the branch"}), + ), + ], + ), + parked=(dialog(tool_name="Read", detail="/tmp/first"),), + ) + waiting = adapter.stopped_on(TARGET, ROSTER_WAITING) + assert waiting.kind is WaitingKind.PERMISSION + assert waiting.tool_name == "Read" + # The record's `detail` describes the record's call, so it goes with it. + assert waiting.detail == "/tmp/first" + assert waiting.approval_id == "a-1" + + def test_a_record_that_named_no_tool_does_not_lend_the_dialog_its_summary( + self, tmp_path: Path + ) -> None: + """Naming no tool is not agreeing with the dialog about which call it is. + + A `tool_use` whose `name` this reader could not read is still summarised + from its input, and that summary describes the call the *record* was held + up on. Beside a dialog naming a tool, the two are not known to be one + call, so the announcement takes the dialog's own words whole. + """ + unnamed = called("Bash", "x1", {"description": "delete the release tag"}) + del unnamed["message"]["content"][0]["name"] + adapter = adapter_holding( + transcript(tmp_path, [*turn(), unnamed]), + parked=(dialog(tool_name="Bash", detail="push the branch"),), + ) + waiting = adapter.stopped_on(TARGET, ROSTER_WAITING) + assert waiting.tool_name == "Bash" + assert waiting.detail == "push the branch" + assert waiting.approval_id == "a-1" + def test_a_call_the_record_could_not_describe_takes_the_dialog_s_words( self, tmp_path: Path ) -> None: @@ -373,6 +461,34 @@ def test_a_line_that_is_not_a_record_is_skipped(self, tmp_path: Path, line: str) assert records is not None assert len(records) == 2 + @pytest.mark.parametrize("separator", ["\u2028", "\u2029", "\u0085"]) + def test_a_unicode_line_separator_inside_a_record_does_not_split_it( + self, tmp_path: Path, separator: str + ) -> None: + """JSONL is `\n`-delimited, and `str.splitlines` is not. + + `JSON.stringify` leaves U+2028, U+2029 and U+0085 raw inside a string, so + they reach the file as themselves — which is why this case writes its + records with `ensure_ascii=False` rather than through `transcript`. + Splitting on them cuts a record in two, neither half parses, and the + record is dropped whole. Measured on 2026-08-26 over the 200 most recent + transcripts on this machine: 4 files losing 31 records between them + (1, 2, 13 and 15). A dropped `tool_result` leaves the call it closes + outstanding, which reads as a permanent false `PERMISSION`. + """ + path = tmp_path / "session.jsonl" + path.write_text( + "".join( + json.dumps(record, ensure_ascii=False) + "\n" + for record in [*turn(), said(f"before{separator}after")] + ), + encoding="utf-8", + ) + records = TranscriptReader().records(path) + assert records is not None + assert len(records) == 3 + assert records[2]["message"]["content"][0]["text"] == f"before{separator}after" + def test_no_path_and_no_file_are_both_none_rather_than_empty(self, tmp_path: Path) -> None: reader = TranscriptReader() assert reader.records(None) is None @@ -395,3 +511,81 @@ def test_forgetting_one_session_leaves_the_others(self, tmp_path: Path) -> None: reader.forget(one) reader.forget(None) assert reader.records(other) is kept + + +class TestWhenTheSessionEnds: + """What the adapter lets go of when it reports a Session ended (#98). + + `forget_session` existed and nothing called it: it is not on the + `AgentAdapter` seam, and Bridge Core's `_session_ended` only marks state. On + an engine that starts at login, that made every Session that ever registered + keep its parsed transcript for the life of the process — records of files + measured at 186 MB on this machine. + + The fix is neither a timer nor a seam method: the adapter that emits + `SessionEnded` is the one that knows, and the cache is its own. So these + drive the real sweep and assert on what is left behind. + """ + + def watching( + self, tmp_path: Path, pid: int + ) -> tuple[ClaudeAgentAdapter, SessionTarget, Path, list[Any]]: + """An adapter registered on that pid, with the transcript already parsed. + + The registry stands in for `~/.claude/sessions` the way + `test_claude_reply_window` writes it, so the sweep below is the real one + rather than a stub of it. + """ + (tmp_path / "sessions").mkdir() + say(tmp_path, "busy", pid=pid, session_id=SESSION) + target = replace(TARGET, pid=pid) + raised: list[Any] = [] + + class Sink: + def emit(self, event: AgentEvent) -> None: + raised.append(event) + + adapter = ClaudeAgentAdapter( + sink=Sink(), + settings=ClaudeSettings( + registry_directory=tmp_path / "sessions", reply_window_poll_seconds=0.02 + ), + ) + path = transcript(tmp_path, [*turn(), called("Bash", "b1", {"description": "push"})]) + adapter._reported[target] = SessionReport( # noqa: SLF001 - seeding one registration + session_id=SESSION, pid=pid, transcript_path=path + ) + adapter.register_session(target, tmp_path / "channel.sock") + assert adapter.stopped_on(target).kind is WaitingKind.PERMISSION # warms the cache + return adapter, target, path, raised + + def test_a_death_reported_is_a_session_forgotten(self, tmp_path: Path) -> None: + corpse = Child() + corpse.kill() + adapter, target, path, raised = self.watching(tmp_path, corpse.pid) + + adapter._windows.poll_once() # noqa: SLF001 - one sweep, without the timer + + assert [type(event) for event in raised] == [SessionEnded] + assert adapter.reported(target) is None + assert adapter.reachable() == () + # The parsed records go with it, which is the megabytes this is about. + assert path not in adapter._transcripts._cache # noqa: SLF001 - the leak itself + + def test_a_session_that_is_still_alive_keeps_everything(self, tmp_path: Path) -> None: + """The sweep runs on every watched Session; only a death may forget one. + + The pid is this test's own process, which is alive by definition. Its + record finishes its turn between registration and the sweep, so the + sweep has something to report — which is what proves it looked at this + Session at all and still buried nothing. + """ + adapter, target, path, raised = self.watching(tmp_path, os.getpid()) + say(tmp_path, "idle", pid=os.getpid(), session_id=SESSION) + + adapter._windows.poll_once() # noqa: SLF001 + + assert [type(event) for event in raised] == [ReplyWindowChanged, SessionStopped] + assert adapter.reported(target) is not None + assert adapter.reachable() == (target,) + assert path in adapter._transcripts._cache # noqa: SLF001 diff --git a/tests/test_codex_agent.py b/tests/test_codex_agent.py index 3dcd33c..f136f13 100644 --- a/tests/test_codex_agent.py +++ b/tests/test_codex_agent.py @@ -751,6 +751,27 @@ async def scenario(): asyncio.run(scenario()) assert [event.target for event in sink.of(SessionEnded)] == [TARGET] + def test_a_closed_thread_is_dropped_as_well_as_reported(self, socket_path: Path) -> None: + """The Session ended, so nothing here goes on holding it (#98). + + `forget_session` had no caller on either lane. The adapter that emits + `SessionEnded` is the one that knows, so it lets go at the emission + site — the other emitter, `_connection_lost`, already did. + """ + sink = Sink() + + async def scenario(): + async with Codex(socket_path).script() as server: + adapter = await watching(server, sink) + try: + await server.notify_all("thread/closed", {"threadId": THREAD}) + await _settled() + return adapter.watching() + finally: + await adapter.aclose() + + assert asyncio.run(scenario()) == () + def test_a_notification_about_a_thread_nobody_watches_is_ignored( self, socket_path: Path ) -> None: