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: