diff --git a/docs/control-plane.md b/docs/control-plane.md index 27ebaff..e050052 100644 --- a/docs/control-plane.md +++ b/docs/control-plane.md @@ -87,7 +87,15 @@ the user is told. ## The actions -Seven, and the set is closed. Adding one is a contract change. +Eight, and the set is closed. Adding one is a contract change. + +`progress` is the eighth. It did **not** move the protocol version, and the +reason is what the version is for: a surface built against 4 can tell an engine +that disagrees from one too old to have been asked, and the failure it guards +against is a surface sending something the engine does not have. Adding an action +cannot cause that — a surface built against 4 never sends `progress`, and one +that does is talking to an engine that has it. *Removing* an action can, which is +what 4 was for. `launch` and `close` were the eighth and ninth until protocol 4. They are parked with the code behind them ([#72](https://github.com/okqixiaobao727-design/GPT-VoiceCoding/issues/72)): @@ -113,27 +121,98 @@ Payload: none. Data: } ``` -A session: +A session — every field of one roster row, because a surface that had to ask a +second question to render one line would be a second reader of the same Session: ```json {"target": {"agent": "claude", "session_id": "abc", "pid": 1234}, "label": "gpt-voicecoding · build the control plane", - "workspace": "/Users/…", "registered_at": 1787222000.0, - "state": "live", "reply_window": "closed"} + "name": "workspace-claude-ed", + "workspace": "/Users/…", "first_seen": 1787222000.0, + "lifecycle": "live", "state": "idle", + "waiting_for": {"kind": "question", "caught_up": true, "prompt": "Which base?", + "options": [{"text": "main", "recommended": true}], + "recommendation": "main", "tool_name": null, "detail": null, + "approval_id": null}, + "progress": {"recent": [{"role": "user", "text": "do the thing"}, + {"role": "assistant", "text": "done"}], + "truncated": false, "read_at": "2026-08-26T02:44:39+00:00"}, + "last_activity": "2026-08-26T02:44:39+00:00", + "child": {"kind": "main", "parent": null}, + "reply_window": "open"} ``` `target` is the **address**; `label` is for speech and for matching. A label never crosses the wire as an address — resolving one to a target is Bridge Core's router, on the way in from the Companion Channel. -`registered_at` is wall-clock seconds: it is written to disk and read back by the -next engine. +`first_seen` is wall-clock seconds, and it is when *this engine* first saw the +Session — no agent knows it. + +`waiting_for` is what a stopped Session stopped on, as structure rather than as a +rendered sentence. `kind` is one of `none`, `question`, `permission`, `unknown`; +`unknown` always comes with `caught_up: false`, and means *ask again*, never +*nothing is happening*. + +`progress` is `null` when nobody read it, and an object with an **empty** +`recent` when it was read and the Session had said nothing — a surface that +collapses the two reports a working Session as an idle one. Each entry says which +side spoke it, so `"make it blue"` and `"I made it blue"` cannot read the same. +It is bounded to the newest 3 entries and 3 KB per row: `truncated` says +something older was dropped, and entries are dropped whole, never cut. + +`last_activity` is separate from `progress` on purpose. A Session can have moved +without saying anything a reader would show, and this is the field that says so. +It is `null` when nothing read one. + +It is deliberately **wider than `progress`**: it counts any work in the Session, +its own subagents included, where `progress` carries only what the user would be +shown. On the Claude lane that means a sidechain record advances it even though +such a record never becomes an entry; on the Codex lane it is the thread's own +`updatedAt`, which moves for any work at all. One meaning on both lanes — and a +Session four minutes into a subagent is not a Session nobody has heard from. A +subagent is not a Child Process: that is a separate agent process, and it has a +row of its own. + +`reply_window` is derived on the row and rendered here, so no surface re-derives +it and no two surfaces can disagree about one Session. ### `sessions` Payload: none. Data: `{"sessions": [...]}` — the same rows `status` carries, for a surface that renders only the roster. +### `progress` + +Payload: `{"target": {"agent": "codex", "session_id": "abc", "pid": null}}`. Data: +`{"session": { … }}` — one roster row, in exactly the shape above. + +It is a **read**: it resolves one exact identity, asks that lane and no other, +and never starts a turn. What it adds over the same row in `sessions` is *when* — +the Session is read at the moment it was asked about, rather than at the last +discovery. A Session mid-turn is answered here and is deliberately not carried on +the roster: reading it is the expensive half, and "how far along is it" is the +question a user asks precisely while it works. + +**It refuses rather than answering emptily**, and the four refusals are four +different facts: + +| Code | When | +| --- | --- | +| `unknown_session` | No Session by that identity was ever registered here. | +| `stale_session` | The identity names a different process now, **or** the Session has ended. The row is not ended by this action: `sessions` is the whole-lane reading and is the only thing that ends one. | +| `refused` | The lane could not be read at all. The message is the lane's own words, and the row stands exactly as the roster last saw it: not being able to look is not a sighting. | +| `refused` | Nothing could read how far it has got — a Codex Session the shared daemon does not hold, or one whose first turn has written no record yet. | + +The last one is the line this whole action is drawn around. A Session that *was* +read and had said nothing answers normally, with an empty `recent`; a Session +nobody could read is a refusal. A surface handed the second as the first would +render a working Session as an idle one. + +A Session the Codex daemon does not hold therefore never gets an invented +reading. Its rollout is on disk and reading it would be a second source answering +the same question with worse evidence. + ### `switch` Payload: `{"name": "duty", "on": true}`. `on` must be a JSON boolean; a string is @@ -219,6 +298,7 @@ by one parser, so neither can grow a command the other lacks. status switch on|off sessions +progress :[:] live relay :[:] [--supplement] approve allow|deny|ask diff --git a/shell/Sources/ShellCore/Wire.swift b/shell/Sources/ShellCore/Wire.swift index cdeb966..a4ab9bc 100644 --- a/shell/Sources/ShellCore/Wire.swift +++ b/shell/Sources/ShellCore/Wire.swift @@ -9,8 +9,10 @@ public let maxRequestBytes = 65536 /// declaration by the cross-language agreement test in `tests/test_app_bundle.py`. public let controlPlaneProtocolVersion = 4 -/// Every action this engine has. Seven, and the set is closed — adding one is a -/// contract change, so the shell names them rather than composing strings. +/// Every action this engine has. Eight, and the set is closed — adding one is a +/// contract change, so the shell names them rather than composing strings. The +/// two spellings are held to each other by `tests/test_app_bundle.py`, which +/// reads this enum and compares it with the engine's own. /// /// `launch` and `close` were here until protocol 4, and are parked with the code /// behind them: the engine answers `unknown_action` for either now, so naming @@ -18,6 +20,8 @@ public let controlPlaneProtocolVersion = 4 public enum Action: String, Sendable, CaseIterable { case status case sessions + /// How far along one exact Session is, read now. A question, never a turn. + case progress case `switch` /// The Live Toggle. One action: it ends the call the system owns, or starts /// one if none is up. Bridge Core owns that policy; no surface holds call state. diff --git a/src/gpt_voicecoding/adapters/agent/_progress.py b/src/gpt_voicecoding/adapters/agent/_progress.py new file mode 100644 index 0000000..5f6155c --- /dev/null +++ b/src/gpt_voicecoding/adapters/agent/_progress.py @@ -0,0 +1,77 @@ +"""How much of what a Session said travels. One bound, both lanes, no exceptions. + +Each lane finds its entries its own way — Claude walks a transcript it owns, +Codex asks the shared daemon for a thread's turns — but *how many* of them a +`Progress` carries is a property of the type and of the wire it crosses, not of +either lane. Two lanes bounding themselves would be two answers to one question, +and a Control Panel row whose length depended on which agent wrote it. + +**Why these two numbers.** `Progress` rides on **every** roster row of a reply +with a 64 KB ceiling (`seams/control_plane.py:44`, enforced on the reply too at +`control_plane/client.py:88-94`). The reference implementation's 12 entries and +32 KB were a per-Session verb's, asked about one Session at a time +(`legacy@1d32845:config.plist:449-452`), and one such answer would spend half +this reply on one row. Three entries is what a spoken line or a menu-bar row can +use — the last thing said, and enough before it to place it — and 3 KB is the +smallest budget that still holds an ordinary paragraph, so a row does not go +silent over a sentence somebody wrote at length. + +The **rule** is ported whole from `legacy@1d32845:bridge/transcript.py:2828-2860`: +keep the newest, widen down until it fits, and say so when anything was dropped. +Legacy's `drop_oversize_tail` is the roster case here rather than an option — a +single entry over the budget yields no entries and `truncated`, because a reply +nobody can read is a worse failure than a row that admits it is empty. +""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from typing import Final + +from gpt_voicecoding.seams.agent import ProgressEntry + +#: How many entries one reading carries. +RECENT_LIMIT: Final = 3 + +#: And how many bytes they may encode to, on the document that travels. +RECENT_MAX_BYTES: Final = 3 * 1024 + + +def bounded( + entries: Sequence[ProgressEntry], + limit: int = RECENT_LIMIT, + *, + max_bytes: int = RECENT_MAX_BYTES, +) -> tuple[tuple[ProgressEntry, ...], bool]: + """The newest of these that fit, and whether anything older was dropped. + + An entry goes whole or stays whole. Half a sentence read aloud says less + than nothing, and a cut lands mid-secret as readily as mid-word — the same + rule `stop_analysis.SUMMARY_MAX_CHARS` follows for a tool summary. + """ + if limit <= 0 or max_bytes <= 0: + raise ValueError("a bound that bounds nothing is a bound nobody set") + kept = list(entries[-limit:]) + truncated = len(kept) < len(entries) + while kept and encoded_size(kept) > max_bytes: + kept.pop(0) + truncated = True + return tuple(kept), truncated + + +def encoded_size(entries: Sequence[ProgressEntry]) -> int: + """How many bytes these entries take on the wire. + + Measured on the document `control_plane.payloads.progress_document` renders, + rather than on the text alone, because the budget exists to keep a roster + reply under the protocol's ceiling and the role travels inside it. + `tests/test_control_plane_actions.py` holds the two spellings together, so + this cannot quietly start measuring a shape nobody sends. + """ + return len( + json.dumps( + [{"role": str(entry.role), "text": entry.text} for entry in entries], + ensure_ascii=False, + ).encode("utf-8") + ) diff --git a/src/gpt_voicecoding/adapters/agent/claude/adapter.py b/src/gpt_voicecoding/adapters/agent/claude/adapter.py index 95b31a4..4b1f28a 100644 --- a/src/gpt_voicecoding/adapters/agent/claude/adapter.py +++ b/src/gpt_voicecoding/adapters/agent/claude/adapter.py @@ -44,10 +44,11 @@ import logging from contextlib import suppress from dataclasses import dataclass, replace +from datetime import UTC, datetime from pathlib import Path from gpt_voicecoding.adapters.agent.claude import discovery as claude_discovery -from gpt_voicecoding.adapters.agent.claude import stop_analysis +from gpt_voicecoding.adapters.agent.claude import stop_analysis, transcript_tail from gpt_voicecoding.adapters.agent.claude.approval import ( CWD_FIELD, MESSAGING_SOCKET_FIELD, @@ -73,7 +74,7 @@ channel_kind_for, ) from gpt_voicecoding.adapters.agent.claude.settings import ClaudeSettings -from gpt_voicecoding.adapters.agent.claude.transcript import TranscriptReader +from gpt_voicecoding.adapters.agent.claude.transcript import Record, TranscriptReader from gpt_voicecoding.adapters.agent.claude.window import ReplyWindowWatcher from gpt_voicecoding.adapters.agent.claude.wire import ( ChannelConnection, @@ -85,6 +86,7 @@ ApprovalVerdict, LaneDiscovery, LaneUnavailable, + Progress, RelayReceipt, RelayRoute, ReplyWindow, @@ -393,21 +395,82 @@ async def discover(self) -> LaneDiscovery: return replace(lane, rows=tuple(self._row_with_stop(row) for row in lane.rows)) def _row_with_stop(self, row: SessionInspection) -> SessionInspection: - """One roster row, with what it stopped on read in. + """One roster row, with everything its own transcript says about it. **A Session mid-turn is not stopped on anything**, so a `RUNNING` row is returned untouched and its transcript is never opened. That is what keeps this off the hot path: on a machine of busy Sessions, the five-second - cadence costs one roster command and no file reads at all. + cadence costs one roster command and no file reads at all. #76 rides on + the same gate rather than a wider one of its own — a roster row is the + cheap projection, and the `progress` verb is where a running Session can + still be asked (`inspect`). """ if row.state is SessionState.RUNNING: return row - waiting = self.stopped_on(row.target, row.waiting_for) - return row if waiting == row.waiting_for else replace(row, waiting_for=waiting) + return self._read_into(row) + + def _read_into(self, row: SessionInspection) -> SessionInspection: + """One row, filled from one read of the transcript its Session named. + + **One read answers both questions.** What the Session stopped on (#75) + and how far along it is (#76) come from the same records at the same + moment; two reads would be two whole-file parses per tick describing two + moments of a file the Session is still appending to. + + **Nothing here bypasses the reader's cache, and nothing needs to.** That + cache is keyed on the file's own identity as the filesystem reports it + (`transcript.py:36-40`), so a hit means the Session has not written a + byte since — and a transcript that has not changed cannot have changed + what its tail says. Forgetting the parse first would buy no freshness and + cost a second pass over a file measured at 186 MB on this machine. + + **A `RUNNING` row is read for progress and never for a stop.** The two + questions do not have the same answer mid-turn: an outstanding `tool_use` + with no result yet is a tool *running*, and reading it as a stop would + report a working Session as one waiting on the user's permission. Only + `inspect` gets here with a `RUNNING` row at all — the cadence returns one + untouched — and that is the row the gate exists for. + """ + records = self._transcripts.records(self._transcript_path(row.target)) + waiting = ( + row.waiting_for + if row.state is SessionState.RUNNING + else self._overlay(row.target, row.waiting_for, records) + ) + if records is None: + # No path, no file yet, or a read that failed. The roster's own word + # stands, and `progress` stays `None` — "not read", never "read and + # found nothing". + return row if waiting == row.waiting_for else replace(row, waiting_for=waiting) + entries, truncated, moved = transcript_tail.recent(records) + return replace( + row, + waiting_for=waiting, + progress=Progress(recent=entries, truncated=truncated, read_at=datetime.now(UTC)), + last_activity=moved, + ) + + def _transcript_path(self, target: SessionTarget) -> Path | None: + """Where this Session's own record is, as its registration named it.""" + report = self._reported.get(target) + return report.transcript_path if report else None def stopped_on(self, target: SessionTarget, roster: WaitingFor | None = None) -> WaitingFor: """What one Session stopped on, from its transcript and any parked dialog. + The Reply Window watcher's route to the same answer (`window.py:308-311`), + and the reason the overlay below is a method rather than inline: a stop + raised by the watcher and a stop read off the roster must be the one + reading, not two that agree most of the time. + """ + records = self._transcripts.records(self._transcript_path(target)) + return self._overlay(target, roster if roster is not None else WaitingFor(), records) + + def _overlay( + self, target: SessionTarget, base: WaitingFor, records: tuple[Record, ...] | None + ) -> WaitingFor: + """What these records and any parked dialog say this Session stopped on. + Two sources, and they are ranked rather than merged, because they can disagree and the ranking is the behaviour (#75): @@ -433,9 +496,6 @@ def stopped_on(self, target: SessionTarget, roster: WaitingFor | None = None) -> Session the roster calls `waiting`, which is the seam's way of saying *ask again, never guess*. """ - base = roster if roster is not None else WaitingFor() - report = self._reported.get(target) - records = self._transcripts.records(report.transcript_path if report else None) found = base if records is None else stop_analysis.analyse(records) if found.kind is WaitingKind.NONE: # The transcript is not held up on anything, which the roster may @@ -473,6 +533,17 @@ async def inspect(self, target: SessionTarget) -> SessionInspection: A Session the roster no longer lists reads as `ENDED`, which is the honest answer to "what is it doing" for a Session that is not there. + **This one is asked about a running Session too, and `discover` is not.** + The cadence skips a `RUNNING` row because a Session mid-turn has not + stopped on anything; but "how far along is it" is the question a user + asks *precisely* while it works, so the per-target read has no such gate + (#76, the `progress` verb), and every row that reaches it is read here + rather than taken from the cadence's pass. What that costs is one pure + walk of records already in memory: the file itself is opened again only + if the Session has written to it since, because the reader's cache is + keyed on the file's own identity. Bypassing that cache is deliberately + *not* done — a hit is the proof of freshness, not a stale answer. + **A roster that could not be read is not a roster that lists nothing.** The two are one value apart here and a whole verdict apart for the caller, so a failed enumeration raises rather than answering `ENDED`. @@ -483,7 +554,7 @@ async def inspect(self, target: SessionTarget) -> SessionInspection: raise LaneUnavailable(AgentKind.CLAUDE, lane.error) for row in lane.rows: if row.target == target: - return row + return self._read_into(row) return SessionInspection( target=target, workspace=Path(), diff --git a/src/gpt_voicecoding/adapters/agent/claude/stop_analysis.py b/src/gpt_voicecoding/adapters/agent/claude/stop_analysis.py index b593b7f..e36b806 100644 --- a/src/gpt_voicecoding/adapters/agent/claude/stop_analysis.py +++ b/src/gpt_voicecoding/adapters/agent/claude/stop_analysis.py @@ -110,7 +110,7 @@ def analyse(records: Sequence[Mapping[str, Any]]) -> WaitingFor: # closes the call it names whether or not it is worth reading aloud, # and a call is followed whether or not its record is visible. _follow(content, open_calls, ordinal) - if _is_visible(record) and not _is_pipeline_noise(record, content) and _speaks(content): + if is_visible(record) and not is_pipeline_noise(record, content) and visible_text(content): last_spoken_at = ordinal return _tail_wait(open_calls, last_spoken_at=last_spoken_at) @@ -297,7 +297,7 @@ def _split_recommendation(label: str) -> tuple[str, bool]: return label, False -def _is_visible(record: Mapping[str, Any]) -> bool: +def is_visible(record: Mapping[str, Any]) -> bool: """Whether this record is part of the conversation the user can see. Three semantic exclusions, not format checks: a sidechain record is a @@ -317,7 +317,7 @@ def _is_visible(record: Mapping[str, Any]) -> bool: ) -def _is_pipeline_noise(record: Mapping[str, Any], content: Any) -> bool: +def is_pipeline_noise(record: Mapping[str, Any], content: Any) -> bool: """Whether this record is slash-command plumbing rather than conversation. Running a slash command writes machinery into the transcript beside the @@ -349,34 +349,70 @@ def _is_pipeline_noise(record: Mapping[str, Any], content: Any) -> bool: return False -def _speaks(content: Any) -> bool: - """Whether this message put anything in front of the user. - - The tail boundary, and nothing more — which is why it answers a boolean - rather than returning the text. What the Session *said* is `Progress`, read - from these same records by its own reader (#76); what this needs to know is - only where the last thing the user would hear sits, because that is what - makes "the tail" mean this stop rather than any older moment. - - `tool_use` counts for exactly one tool name. The Session's own words say what - decision it is waiting on, but the choices live in `AskUserQuestion`'s input, - so this call *is* something the user is shown. Every other `tool_use` carries - commands, code and file contents (`legacy@1d32845:bridge/transcript.py: - 1568-1605`) — and, more to the point here, counting one would put the tail - boundary after the very call the Session is waiting on. +def visible_text(content: Any) -> str: + """What this message put in front of the user, or nothing at all. + + **Two callers, one rule, and that is the point.** `analyse` needs only + whether a message spoke, because that is what gives "the tail" a boundary; + `transcript_tail` needs the words themselves, because that is what `Progress` + is made of (#76). Answering the boolean from a second scan of the same + content is how the reference implementation ended up with readers that + disagreed about the same record — so there is one extractor and the boundary + is `bool(visible_text(...))`. + + Every content shape Claude Code writes is *collected from* rather than + enumerated: real transcripts carry at least six user shapes and five + assistant ones and the product adds more, so an unknown item contributes + nothing instead of making the message unreadable + (`legacy@1d32845:bridge/transcript.py:1568-1607`). + + `tool_use` is readable for exactly one tool name. The Session's own words say + what decision it is waiting on, but the choices live in `AskUserQuestion`'s + input, so this call *is* something the user is shown — without it the user + hears the question and never hears the options. Every other `tool_use` + carries commands, code and file contents, which must not be read aloud; and, + for the tail boundary, counting one would put the boundary after the very + call the Session is waiting on. """ if isinstance(content, str): - return bool(content.strip()) + return content if content.strip() else "" if not isinstance(content, list): - return False + return "" + readable = "" for item in content: if not isinstance(item, Mapping): continue if item.get("type") == "text": text = item.get("text") if isinstance(text, str) and text.strip(): - return True + readable += text elif item.get("type") == "tool_use" and item.get("name") == QUESTION_TOOL: - if _question(item.get("input")) is not None: - return True - return False + asked = _question_text(item.get("input")) + if asked: + readable += f"\n{asked}" if readable else asked + return readable + + +def _question_text(tool_input: Any) -> str: + """What one `AskUserQuestion` call asks, and what it offers as answers. + + Rendered off the same `_groups` the seam's `Option` values are built from, so + the words the user hears and the choices they may say back can never come + from two different readings of one call + (`legacy@1d32845:bridge/transcript.py:1861-1874`). + + **A call with no readable option renders as nothing**, which is the same + test `_question` fails on. An input still being streamed arrives with no + options at all, and a prompt without choices read aloud is a question the + user cannot answer — and, for `analyse`'s tail boundary, would move the + boundary past the very call the Session is waiting on. + """ + groups = _groups(tool_input) + if not any(labels for _, labels in groups): + return "" + lines: list[str] = [] + for prompt, labels in groups: + if prompt: + lines.append(f"Question: {prompt}") + lines.extend(f"Option: {label}" for label in labels) + return "\n".join(lines) diff --git a/src/gpt_voicecoding/adapters/agent/claude/transcript_tail.py b/src/gpt_voicecoding/adapters/agent/claude/transcript_tail.py new file mode 100644 index 0000000..ea8fc2f --- /dev/null +++ b/src/gpt_voicecoding/adapters/agent/claude/transcript_tail.py @@ -0,0 +1,144 @@ +"""How far along one Claude Session is, read out of its own transcript records. + +One function, no I/O. It is handed records — already parsed, in the order the +Session wrote them, by the lane's one opener of a transcript file +(`transcript.py`) — and answers the two fields #74 locked beside each other: +`Progress` and `last_activity`. + +**They are two fields because they are two facts.** A Session that spent a +minute running tools has moved without saying anything a reader would show, so a +`last_activity` derived from the newest *visible* entry would report it as +stalled. The time therefore comes off any record in the file, and the entries +come only off the conversation. + +**Including a sidechain record, and that is a departure from #76's own +sentence.** The ticket says the helper excludes "child/sidechain, system and +command-plumbing records as legacy does", and `recent`'s *entries* do exactly +that. `last_activity` deliberately does not, ruled on #76 by the Advisor, for +three reasons worth keeping together: + +* The field means *this Session moved* (`seams/agent.py:314-316`). A Task + subagent writes into **this** transcript and has no roster row of its own, so + no other row could own that time; excluding it would report a Session four + minutes into a subagent as one nobody has heard from, which is the exact + failure the field was split off from `progress` to prevent. +* Two lanes, one meaning. The Codex lane's `last_activity` is the thread's own + `updatedAt`, which moves for any work at all (`codex/thread_tail.py`), and the + Claude lane must not say something narrower under the same name. +* A sidechain record is **not** #68's Child Process. That rule is about a + separate `claude` *process*, which gets a row of its own and is never spoken + to; a sidechain record is one Session's own subagent, in one Session's own + file. Conflating them is what makes this look like an oversight. + +**The visibility rules are not restated here.** `is_visible`, +`is_pipeline_noise` and `visible_text` live in `stop_analysis` and are consumed, +not re-implemented — the same records, the same exclusions, one reading. Legacy +ran two scans of one file for these two questions and they could describe two +different moments of a file the Session is still appending to +(`legacy@1d32845:bridge/transcript.py:1184-1208`). + +**Against legacy** (ADR 0010, `CLAUDE.md`). The read is **ported** from +`legacy@1d32845:bridge/transcript.py:1184-1246` (walk the records, keep what the +user can see, bound the tail) with its text extraction at `:1568-1607` and its +bounding walk at `:2828-2860`. **Adapted**: legacy's sidechain exclusion +(`_top_level_records`, `:1125-1142`) is a **visibility** rule — what to show, and +what counts as a stop — and it is ported whole into `recent`'s entries and into +`stop_analysis.analyse`. `last_activity` is not a visibility fact, and legacy is +not a citation for it either way: gen 1 had no such field at all. **Adapted**: +legacy's entry was +`TranscriptMessage(role, text)` and this is the seam's `ProgressEntry`, which +carries the same two facts; legacy's bounds were a per-Session verb's (12 +entries / 32 KB, `config.plist:449-452`) and these are a roster row's, for the +reason beside them. **Dropped**: the read deadline, the whole-file identity +check and the `TranscriptReadError` taxonomy — all three left with the reader +(`transcript.py`), which answers `None` and is asked again on the next tick. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import datetime +from typing import Any, Final + +from gpt_voicecoding.adapters.agent._progress import RECENT_LIMIT, RECENT_MAX_BYTES, bounded +from gpt_voicecoding.adapters.agent.claude.stop_analysis import ( + is_pipeline_noise, + is_visible, + visible_text, +) +from gpt_voicecoding.seams.agent import ProgressEntry, ProgressRole + +#: What Claude Code writes a record's time as. Read with `fromisoformat`, which +#: takes the trailing `Z` and answers a timezone-aware value, so nothing here +#: has to decide which zone an unmarked time was in. +_TIMESTAMP = "timestamp" + +#: The two record types that are the conversation. `ProgressRole` is spelled the +#: same way, so the record says which side spoke and nothing infers it. +_ROLES: Final = {"user": ProgressRole.USER, "assistant": ProgressRole.ASSISTANT} + + +def recent( + records: Sequence[Mapping[str, Any]], + limit: int = RECENT_LIMIT, + *, + max_bytes: int = RECENT_MAX_BYTES, +) -> tuple[tuple[ProgressEntry, ...], bool, datetime | None]: + """The newest of what was said, whether anything older was dropped, and when. + + An empty tuple with `truncated=False` is a Session that has genuinely said + nothing yet; with `truncated=True` it is one whose newest entry alone is over + the budget. The caller reports either as `Progress`; what it may never do is + read the pair as "nothing is happening", which is `None` from the reader + above and a different fact. + """ + entries: list[ProgressEntry] = [] + moved: datetime | None = None + for record in records: + if not isinstance(record, Mapping): + continue + when = _moment(record.get(_TIMESTAMP)) + if when is not None: + # Any record at all, this Session's own subagents included: a + # subagent writing here is this Session running, the field exists + # to say so when nothing was said, and no other row could own that + # time. Deliberately wider than the entries below — the module + # docstring says why, and #76's Advisor ruling is what settled it. + moved = when + entry = _entry(record) + if entry is not None: + entries.append(entry) + + kept, truncated = bounded(entries, limit, max_bytes=max_bytes) + return kept, truncated, moved + + +def _entry(record: Mapping[str, Any]) -> ProgressEntry | None: + """One record as something that was said, or `None` if it was not.""" + role = _ROLES.get(str(record.get("type"))) + if role is None: + return None + message = record.get("message") + if not isinstance(message, Mapping) or message.get("role") != str(record.get("type")): + return None + content = message.get("content") + if not is_visible(record) or is_pipeline_noise(record, content): + return None + text = visible_text(content) + return ProgressEntry(role=role, text=text) if text.strip() else None + + +def _moment(stamp: Any) -> datetime | None: + """When a record says it was written, or nothing. + + A record being appended right now routinely has no readable time, and the + answer to that is the same one the reader gives a half-written line: it + costs itself. Blanking the time the rest of the file proves would report a + busy Session as one nobody has heard from. + """ + if not isinstance(stamp, str) or not stamp.strip(): + return None + try: + return datetime.fromisoformat(stamp.strip()) + except ValueError: + return None diff --git a/src/gpt_voicecoding/adapters/agent/codex/adapter.py b/src/gpt_voicecoding/adapters/agent/codex/adapter.py index 617c3f9..3a942a9 100644 --- a/src/gpt_voicecoding/adapters/agent/codex/adapter.py +++ b/src/gpt_voicecoding/adapters/agent/codex/adapter.py @@ -37,12 +37,15 @@ import asyncio import logging import os +from dataclasses import replace from pathlib import Path from typing import Any from gpt_voicecoding import __version__ from gpt_voicecoding.adapters.agent.codex import approvals as approval_wire from gpt_voicecoding.adapters.agent.codex import discovery as codex_discovery +from gpt_voicecoding.adapters.agent.codex import thread_tail +from gpt_voicecoding.adapters.agent.codex.shared_daemon import SharedDaemon from gpt_voicecoding.adapters.agent.codex.threads import ( PINNED_POLICY, USER_REVIEWER, @@ -113,6 +116,8 @@ def __init__( settings: CodexSettings | None = None, own_socket_path: Path | None = None, own_log_path: Path | None = None, + daemon: SharedDaemon | None = None, + processes: codex_discovery.ProcessLister | None = None, ) -> None: self._sink = sink self._settings = settings or CodexSettings() @@ -128,6 +133,18 @@ def __init__( log_path=own_log_path, version=__version__, ) + #: This engine's client of the daemon somebody else owns. Joined lazily, + #: on the first discovery that needs it, because an engine must start on + #: a machine whose daemon is not up yet. Injectable for the same reason + #: the app-server's socket path is: a test that dialled the real one + #: would be a test talking to the Sessions of whoever ran it. + self._daemon = daemon or SharedDaemon(settings=self._settings, version=__version__) + #: What each loaded thread last said, read at most once per change. + self._turns = codex_discovery.TurnCache() + #: How the machine's own process table is read. Injected for the reason + #: the daemon is: a test that shelled out to `ps` would be a test whose + #: answer depends on what the person running it happens to have open. + self._processes = processes or codex_discovery.enumerate_sessions self._opened = False # -- the connection this engine owns ---------------------------------- @@ -169,6 +186,13 @@ async def aclose(self) -> None: watching = list(self._threads.values()) self._threads.clear() await asyncio.gather(*(self._detached(watched) for watched in watching)) + # Only this engine's end of the shared daemon: the daemon itself carries + # on, because the user's `codex` TUIs are its clients and stopping it + # would end their Sessions (#83's written rule, ADR 0012). It never waits + # on a dial in flight — #96's shutdown budget is derived from the phases + # it bounds, and a phase that could sit out a ten-second lookup is not + # one of them (`shared_daemon.SharedDaemon.aclose`). + await self._daemon.aclose() await self._own.aclose() self._opened = False @@ -254,13 +278,19 @@ def watching(self) -> tuple[SessionTarget, ...]: async def discover(self) -> LaneDiscovery: """Every Codex Session on this machine, from the daemon and from the machine. - Delegated whole to `discovery.py`. The client handed over is the shared - daemon's when there is one — `None` while the LaunchAgent that starts it - is #83's to install, which is not a gap in the roster: those Sessions are - listed from the process table, and a Relay into one fails at the wire - with its reason (#82). + Delegated whole to `discovery.py`. The client handed over is this + engine's connection to the shared daemon — `None` when the daemon is not + answering, which is not a gap in the roster: those Sessions are listed + from the process table, the lane says so on `degraded`, and a Relay into + one fails at the wire with its reason (#82). """ - return await codex_discovery.discover(self._shared_daemon()) + client = await self._shared_daemon() + return await codex_discovery.discover( + client, + processes=self._processes, + turns=self._turns, + daemon_note=self._daemon.note, + ) async def inspect(self, target: SessionTarget) -> SessionInspection: """One Session, freshly read from the same sources `discover` reads. @@ -269,17 +299,26 @@ async def inspect(self, target: SessionTarget) -> SessionInspection: a Codex Session gains its thread id at its first turn (#73), so the two readings of one process may name it differently. + **One live read, and exactly one** (#76). Dropping this thread's + remembered turns *before* enumerating is what makes that true: the + enumeration below then takes one fresh deep read for it, where leaving + the cache alone would have the cadence answer from memory and this verb + read a second time — two 558,875-byte reads for one question. A thread + mid-turn is gated out of that read and is picked up by `_turns_into` + instead, which is the other half of the same one-read rule. + **A lane that could not look raises**, rather than reporting a Session it never saw as `ENDED` — the same rule `SessionRegistry.observe` follows for `LaneDiscovery.error`, for the same reason. """ + self._turns.forget(target.session_id) lane = await self.discover() if not lane.enumerated: assert lane.error is not None # `enumerated` is exactly this test raise LaneUnavailable(AgentKind.CODEX, lane.error) for row in lane.rows: if row.target == target or (target.pid is not None and row.target.pid == target.pid): - return row + return await self._turns_into(row) return SessionInspection( target=target, workspace=Path(), @@ -287,22 +326,66 @@ async def inspect(self, target: SessionTarget) -> SessionInspection: state=SessionState.IDLE, ) - def _shared_daemon(self) -> codex_discovery.DaemonClient | None: - """A connection to the shared app-server daemon, when this engine holds one. - - There is none yet, and deliberately: the daemon's lifecycle is a login - `LaunchAgent` that #83 installs, and #74 does not start daemons. Until - then this returns `None` and the lane reads the machine instead, which - is exactly what it does for a TUI the daemon never adopted anyway. - - **`None` here means "nothing was dialled", and the roster says exactly - that** (`discovery.NO_CLIENT`). It used to say the daemon had not - answered, which is a claim about the daemon that this build is in no - position to make — and on a machine where `bridge-install status` really - does dial one and really does get an answer, the two readings look like - a contradiction rather than like two builds (#96). + async def _shared_daemon(self) -> codex_discovery.DaemonClient | None: + """This engine's connection to the shared app-server daemon, if it has one. + + The lane's one door to the daemon, and the reason it is a method rather + than a field: #77 takes its Relay and Approval routes through this same + connection, and its pre-wire `FAILED` is what `None` here means. A second + caller opening its own client would be a second thing for the daemon to + drop and a second thing to notice it had. + + `None` is honest and ordinary: the daemon may not be up yet, and the lane + then reads the machine instead — which is exactly what it does for a TUI + the daemon never adopted anyway (#82). + + **`None` no longer means "nothing was dialled", and the roster had to + stop saying it did.** #96 split that sentence out precisely because this + method returned `None` without a byte being sent, so a roster reading + "the daemon did not answer" contradicted `bridge-install status` dialling + the same daemon on the same machine and getting an answer. #76 builds the + client, so something is now always attempted — and what the roster says + is the dial's own reason (`SharedDaemon.note`), which is more precise + than either sentence #96 had to choose between. """ - return None + return await self._daemon.client() + + async def _turns_into(self, row: SessionInspection) -> SessionInspection: + """One row with its turns read live, for the verb that asks about one Session. + + **The cadence's gate is not this verb's** (#76). A thread mid-turn carries + no progress on the roster, because reading turns costs 558,875 bytes per + thread per read (measured, `discovery.TurnCache`) and a working Session is + not stopped on anything — but "how far along is it" is the question a user + asks *precisely* while it works, so here it is read. + + **A row that already carries a reading keeps it, and that is the one-read + rule rather than a shortcut.** `inspect` dropped this thread's cached + turns before enumerating, so a reading on the row is one taken moments + ago in this same call. Reading again would be the second of two + half-megabyte reads for one question. + + **An unattached row keeps `progress=None`, and never a guessed one.** Its + rollout is on disk and reading it would be a second source answering the + same question with worse evidence — the port table left exactly that + behind (P6, P13). Bridge Core turns that `None` into the honest error #76 + asks for; it is not this adapter's to invent one. + """ + if row.progress is not None or row.target.session_id is None: + return row + client = await self._shared_daemon() + if client is None: + return row + described = await codex_discovery.read_thread( + client, row.target.session_id, with_turns=True + ) + if described is None: + return row + return replace( + row, + progress=codex_discovery.progress_from(described), + last_activity=thread_tail.last_activity(described) or row.last_activity, + ) def supported_routes(self) -> frozenset[RelayRoute]: """Both. `turn/steer` is stable in the codex this adapter is built against.""" diff --git a/src/gpt_voicecoding/adapters/agent/codex/discovery.py b/src/gpt_voicecoding/adapters/agent/codex/discovery.py index 385f1c8..4d100a1 100644 --- a/src/gpt_voicecoding/adapters/agent/codex/discovery.py +++ b/src/gpt_voicecoding/adapters/agent/codex/discovery.py @@ -30,14 +30,16 @@ import logging import os -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Mapping +from datetime import UTC, datetime from pathlib import Path from typing import Any, Final, Protocol -from gpt_voicecoding.adapters.agent.codex import rollouts +from gpt_voicecoding.adapters.agent.codex import rollouts, thread_tail from gpt_voicecoding.adapters.agent.codex.processes import Candidate, enumerate_sessions from gpt_voicecoding.seams.agent import ( LaneDiscovery, + Progress, SessionInspection, SessionLifecycle, SessionState, @@ -48,18 +50,22 @@ _log = logging.getLogger(__name__) -#: The daemon's roster, and the per-thread read. `includeTurns` is off: this is -#: discovery, and a turn list is #76's business and a much larger answer. +#: The daemon's roster, and the per-thread read. ROSTER_METHOD: Final = "thread/loaded/list" READ_METHOD: Final = "thread/read" +#: What a thread's status is called while a turn is running. Named because two +#: rules turn on it: the row reads as `RUNNING`, and the cadence does not read +#: its turns. +ACTIVE_STATUS: Final = "active" + #: What a thread's `status.type` can be, as the Codex spoke has observed it. #: `systemError` is a thread whose turn ended badly — still reachable, still #: able to take the next Relay, which is why it reads as idle rather than as #: something that must be waited out. STATUS_TYPES: Final = { "idle": SessionState.IDLE, - "active": SessionState.RUNNING, + ACTIVE_STATUS: SessionState.RUNNING, "systemError": SessionState.IDLE, } @@ -68,18 +74,31 @@ #: are not the same fact. FROM_THE_MACHINE = "so these rows come from the process table and the rollouts on disk" -#: This build never dials. `CodexAdapter._shared_daemon` returns `None` until -#: #77 builds the client, so there is no connection to try and nothing answers -#: or fails to answer. +#: This engine holds no connection, and nothing recorded why. **The fallback +#: sentence, and it should be the rarest of the three** — `SharedDaemon.client()` +#: sets a reason on `note` on every path that answers `None`, and `_degraded` +#: prefers that reason over this one because the dial's own words are always more +#: precise than this is. +#: +#: **It says only what this process can see**, which is the reading the Advisor +#: fixed on #96 ("consequence for part B", 2026-08-26): a client `None` means +#: there is no connection *here*, and it may not be spelled as a claim about +#: whether the daemon was dialled, answered, or exists. That is `NO_DAEMON`'s +#: sentence, and it is only ever said after a request was really made. #: -#: **The distinction is not pedantry; it cost a session** (#96). This used to be -#: the same sentence as `NO_DAEMON` below, so a roster reading "the daemon did -#: not answer" was produced without a single byte being sent — while -#: `bridge-install status`, on the same machine at the same moment, really did -#: dial the daemon and really did get an answer. The two readings looked like a -#: contradiction to be investigated, and a session went and investigated it. +#: **What this used to say, and why it may not say it again** (#96). Until #76, +#: `CodexAgentAdapter._shared_daemon()` returned `None` unconditionally, and this +#: sentence was the same one as `NO_DAEMON` below — so a roster reading "the +#: daemon did not answer" was produced without a single byte being sent, while +#: `bridge-install status` on the same machine at the same moment really did dial +#: the daemon and really did get an answer. The two readings looked like a +#: contradiction to be investigated, and a session went and investigated it. #76 +#: builds the client, so "nothing was dialled" is no longer true either; what +#: survives from #96 is its rule, which is the one that mattered: **never claim +#: the daemon was silent, and never claim anything about it this build did not +#: observe.** NO_CLIENT = ( - f"this build does not connect to the shared Codex app-server daemon yet, {FROM_THE_MACHINE}" + f"this engine holds no connection to the shared Codex app-server daemon, {FROM_THE_MACHINE}" ) #: The daemon was dialled and did not answer. Only ever said after a request was @@ -109,11 +128,86 @@ async def request(self, method: str, params: dict[str, Any] | None = None) -> di ProcessLister = Callable[[], Awaitable[tuple[Candidate, ...]]] +class TurnCache: + """Every loaded thread's `Progress`, read at most once per change (#76). + + **The cache is the whole reason this class exists, and it was measured.** A + `thread/read` with `includeTurns: true` answered **558,875 bytes** for a + thread of two turns against the real daemon on codex 0.149.1 (2026-08-26), + where the same read without turns answered 3,600. `numTurns` is not a + parameter of this method — passing it changed nothing — so there is no way + to ask for a smaller answer. On a five-second cadence over a machine of + stopped Sessions, reading turns every tick would be megabytes a minute for + an answer that had not changed. + + **Keyed on the thread's own `updatedAt`,** which is the same argument the + Claude lane's transcript cache makes from `(size, mtime)`: a thread that has + not been touched cannot have changed what its tail says, and the key is the + far side's own account rather than a clock this module would have to guess + the right length for. A thread that names no `updatedAt` is never cached — + without a time there is nothing to say it has not moved. + + **`updatedAt` is epoch seconds, so its resolution is one second**, and two + changes to a thread inside the same second are one change to this cache. That + is written down rather than defended against: the rows it keys are *stopped* + threads, so the next thing that moves one moves the second too, and a row + somebody asks about through the `progress` verb is read live regardless. + + `read_at` is kept from the reading that was taken, not refreshed on a hit. + That is what it means: when this was true. + """ + + def __init__(self) -> None: + #: thread id → (the `updatedAt` it was read at, what was read). + self._cache: dict[str, tuple[Any, Progress]] = {} + + async def progress_for(self, client: DaemonClient, thread: dict[str, Any]) -> Progress | None: + """This thread's progress, read or remembered — or `None` for a live turn. + + A thread mid-turn is not read on the cadence, for the reason the Claude + lane does not open a `RUNNING` Session's transcript: it is the expensive + read, and the roster row is the cheap projection beside the per-target + verb (#76, advisor ruling Q3). + """ + thread_id = thread.get("id") + if not isinstance(thread_id, str) or _status_of(thread) == ACTIVE_STATUS: + return None + stamp = thread.get(thread_tail.UPDATED_AT) + cached = self._cache.get(thread_id) + if stamp is not None and cached is not None and cached[0] == stamp: + return cached[1] + described = await read_thread(client, thread_id, with_turns=True) + if described is None: + return None + progress = progress_from(described) + if stamp is not None: + self._cache[thread_id] = (stamp, progress) + return progress + + def forget(self, thread_id: str | None) -> None: + """Drop one thread's remembered reading, so the next look is a fresh one. + + The per-target verb's doing (#76). It is how "read this Session now" + stays **one** read: the enumeration that follows takes the fresh deep + read for this thread, instead of the cadence answering from the cache + and the verb then reading a second time. + """ + if thread_id is not None: + self._cache.pop(thread_id, None) + + def retain(self, thread_ids: set[str]) -> None: + """Forget every thread the daemon no longer holds, so this stays roster-sized.""" + for gone in set(self._cache) - thread_ids: + del self._cache[gone] + + async def discover( client: DaemonClient | None = None, *, processes: ProcessLister = enumerate_sessions, home: Path | None = None, + turns: TurnCache | None = None, + daemon_note: str = "", ) -> LaneDiscovery: """Every Codex Session on this machine, however well it can be described. @@ -134,14 +228,53 @@ async def discover( pid = _pid_for(thread, candidates, claimed) if pid is not None: claimed.add(pid) - rows.append(_from_thread(thread, pid)) + progress = ( + await turns.progress_for(client, thread) + if turns is not None and client is not None + else None + ) + rows.append(_from_thread(thread, pid, progress)) + if turns is not None: + turns.retain({str(thread.get("id")) for thread in threads}) rows.extend( _from_process(candidate, home=home) for candidate in candidates if candidate.pid not in claimed ) - return LaneDiscovery(rows=tuple(rows), degraded=daemon_error) + return LaneDiscovery(rows=tuple(rows), degraded=_degraded(daemon_error, daemon_note)) + + +def progress_from(thread: Mapping[str, Any]) -> Progress: + """One `thread/read` answer, as the seam holds it. + + The one place a thread document becomes a `Progress`, so the cadence's cached + read and the verb's live one cannot come back describing it two ways. + `read_at` is stamped here because it belongs to the *reading*: it is when this + was true, and a value carried forward from a cache hit keeps its own moment. + """ + entries, truncated = thread_tail.recent(thread) + return Progress(recent=entries, truncated=truncated, read_at=datetime.now(UTC)) + + +def _degraded(daemon_error: str | None, note: str) -> str | None: + """Why these rows are thinner than usual — from both things that can say so. + + A lane can be reading from the process table *and* joined to a daemon whose + version disagrees with the CLI's, and neither fact is allowed to hide the + other. There is deliberately no `error` path here: a missing daemon has + never been a reason to report no Sessions (#74). + + **A dial that failed says why in its own words, once.** When there is no + client and the dial left a reason, that reason replaces `NO_CLIENT` rather + than following it: "holds no connection; codex did not answer within 10 + seconds" is two sentences making one claim, and #96 is the record of what a + roster that makes more claims than it observed costs to read. + """ + if daemon_error == NO_CLIENT and note: + return f"{note}, {FROM_THE_MACHINE}" + reasons = [reason for reason in (daemon_error, note) if reason] + return "; ".join(reasons) or None async def _threads(client: DaemonClient | None) -> tuple[list[dict[str, Any]], str | None]: @@ -174,16 +307,24 @@ async def _threads(client: DaemonClient | None) -> tuple[list[dict[str, Any]], s for thread_id in ids: if not isinstance(thread_id, str) or not thread_id.strip(): continue - described = await _read(client, thread_id.strip()) + described = await read_thread(client, thread_id.strip()) if described is not None: found.append(described) return found, None -async def _read(client: DaemonClient, thread_id: str) -> dict[str, Any] | None: - """One thread as the daemon describes it, or `None` if it cannot describe it.""" +async def read_thread( + client: DaemonClient, thread_id: str, *, with_turns: bool = False +) -> dict[str, Any] | None: + """One thread as the daemon describes it, or `None` if it cannot describe it. + + `with_turns` is the expensive half and is asked for only by `TurnCache`, + which is where the measurement justifying that word lives. + """ try: - answer = await client.request(READ_METHOD, {"threadId": thread_id, "includeTurns": False}) + answer = await client.request( + READ_METHOD, {"threadId": thread_id, "includeTurns": with_turns} + ) except Exception as unreadable: # noqa: BLE001 - one bad thread is not a bad roster _log.info("the daemon could not describe thread %s: %s", thread_id, unreadable) return None @@ -217,10 +358,18 @@ def _pid_for( return matches[0] if len(matches) == 1 else None -def _from_thread(thread: dict[str, Any], pid: int | None) -> SessionInspection: - """One daemon-held thread as the seam holds it.""" +def _status_of(thread: Mapping[str, Any]) -> str | None: + """What this thread says it is doing, in its own word.""" status = thread.get("status") - kind = status.get("type") if isinstance(status, dict) else None + kind = status.get("type") if isinstance(status, Mapping) else None + return kind if isinstance(kind, str) else None + + +def _from_thread( + thread: dict[str, Any], pid: int | None, progress: Progress | None = None +) -> SessionInspection: + """One daemon-held thread as the seam holds it.""" + kind = _status_of(thread) state = STATUS_TYPES.get(str(kind), SessionState.RUNNING) cwd = thread.get("cwd") name = thread.get("name") @@ -236,6 +385,11 @@ def _from_thread(thread: dict[str, Any], pid: int | None) -> SessionInspection: if kind == "systemError" else WaitingFor() ), + progress=progress, + # Free on the cheap read, and honest for a thread mid-turn too: it is + # the thread's own account of when it last moved, which is exactly the + # case `last_activity` exists to answer when nothing was said (#76). + last_activity=thread_tail.last_activity(thread), name=name.strip() if isinstance(name, str) and name.strip() else None, ) diff --git a/src/gpt_voicecoding/adapters/agent/codex/shared_daemon.py b/src/gpt_voicecoding/adapters/agent/codex/shared_daemon.py new file mode 100644 index 0000000..d31ec71 --- /dev/null +++ b/src/gpt_voicecoding/adapters/agent/codex/shared_daemon.py @@ -0,0 +1,323 @@ +"""This engine's client of the shared Codex app-server daemon. Join-only. + +**Why this exists at all.** #82 proved that the shared daemon is the only source +that knows a thread's id, its name and what it has been doing, and #83 installed +the login job that starts it — but nothing in this engine ever dialled it, so +every Codex row came off the process table and no thread could be read. Progress +on the Codex lane was `None` by construction (#76, advisor ruling Q1). This is +the dial, and only the dial. + +**Join-only, and that is a rule rather than a scope note.** This product starts +a daemon the user's Sessions will join and never stops one they are attached to +(#83, ADR 0012): by the time this engine shuts down, the user's `codex` TUIs are +thin clients of that daemon, and closing it would end their Sessions. So nothing +here spawns, bootstraps or boots out anything — it opens a connection, keeps it, +and lets go of its own end. + +**Where the socket is comes from the daemon, never from `CODEX_HOME`.** +`codex app-server daemon version` answers a document naming the socket it is +listening on, which is the one address that cannot go stale under a moved home +or an updated managed binary. Ported from #82's prototype +(`scripts/prototype_codex_daemon.py:51-68,442-454`), which is where this shape +was proved against a real daemon. `installation/codex_launch_agent.py` runs the +same command for a different question — whether the daemon a person just +installed is answering, as one sentence for a status run — and the two are +deliberately not one function: installation may import no part of the engine +(ADR 0012's one-way rule), and this is the engine. + +**No version pin** (Simon's ruling on #67). A CLI and a daemon whose versions +disagree still get dialled; the disagreement rides out as a note on +`LaneDiscovery.degraded`, so the user can see the lane is running on a +disagreement instead of the lane simply going quiet. + +**Locating is re-tried, not remembered.** The address is looked up only when +there is no live connection, so a healthy engine spawns no subprocess per tick; +an engine whose daemon is down probes once per discovery instead, which is the +only honest way to notice it came back. + +**Against legacy** (ADR 0010, `CLAUDE.md`): **dropped, because gen 1 had no +shared daemon to join.** It drove a launched, wrapped, per-Session app-server it +owned — `legacy@1d32845:bridge/codex.py:1319-1347` opened a client per read +against a socket its own runtime had spawned — and #82 recorded that whole route +as dropped from porting. What survives from it is the *shape* of the read side: +locate, connect, ask, and never fall back to another source when this one cannot +answer. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Final + +from gpt_voicecoding.adapters.codex_app_server.process import AppServerError, attach +from gpt_voicecoding.adapters.codex_app_server.settings import CodexSettings +from gpt_voicecoding.adapters.codex_app_server.wire import AppServerConnection, WireError + +_log = logging.getLogger(__name__) + +#: What is asked of the `codex` binary to find the daemon. `version` because it +#: is the one daemon subcommand that only reports — `start` would make this the +#: owner of a lifecycle it is forbidden to own. +DAEMON_VERSION_ARGUMENTS: Final = ("app-server", "daemon", "version") + +#: How long that one command gets before it is given up on. Measured at 139 ms +#: against a daemon that is up, so this is not a budget the ordinary case spends +#: — it is the bound on the case that has no ordinary one, a `codex` that never +#: answers. Without it, `communicate()` waits forever and the five-second +#: discovery cadence waits with it: `core/bridge.py:505` awaits `discover()` with +#: no deadline of its own, and it is one loop over the adapters, so one hung +#: subprocess here stops the **Claude** lane's roster too. The number is the +#: lane's other subprocess reader's (`codex/processes.py:138`), because it is the +#: same risk read off the same machine. +COMMAND_TIMEOUT_SECONDS: Final = 10.0 + +#: How much of an unreadable answer is quoted back in the reason. Long enough to +#: recognise what came out — a shell banner, an HTML error page, a stack trace's +#: first line — and short enough that a daemon answering megabytes cannot put +#: them into a roster reply with a 64 KB ceiling on it. +UNREADABLE_ANSWER_QUOTED_CHARS: Final = 120 + +SOCKET_PATH_FIELD: Final = "socketPath" +CLI_VERSION_FIELD: Final = "cliVersion" +APP_SERVER_VERSION_FIELD: Final = "appServerVersion" + +#: How the command is run. Injected so a test never reaches the real machine — +#: the same rule `tests/conftest.py` holds the whole suite to after two drafts of +#: the installation boundary started a real daemon from a test run. +Runner = Callable[[list[str]], Awaitable[tuple[int, str]]] + + +@dataclass(frozen=True, slots=True) +class DaemonAddress: + """Where the shared daemon listens, and what it said about itself.""" + + socket_path: Path + cli_version: str + app_server_version: str + + @property + def note(self) -> str: + """Why rows read through this connection deserve a caveat, if they do. + + Both versions are checked for being *said* before they are compared: a + document carrying neither field makes them both empty, and `"" == ""` + would report a daemon that said nothing at all as one whose versions + agree (`installation/codex_launch_agent.py:301-306`, the same trap). + """ + if not self.cli_version or not self.app_server_version: + return ( + f"the shared Codex daemon did not say its versions " + f"({CLI_VERSION_FIELD}={self.cli_version!r}, " + f"{APP_SERVER_VERSION_FIELD}={self.app_server_version!r})" + ) + if self.cli_version != self.app_server_version: + return ( + f"the Codex CLI is {self.cli_version!r} and the running app-server is " + f"{self.app_server_version!r} — a Session started by this CLI will not " + "join that daemon" + ) + return "" + + +async def locate(executable: str, *, run: Runner | None = None) -> tuple[DaemonAddress | None, str]: + """Where the shared daemon is, or the reason nothing could be found. + + Never raises. This runs inside a five-second discovery tick, and a lane that + threw because a binary moved would take the roster down with it — the honest + answer is a reason the rows can carry (`LaneDiscovery.degraded`, #74). + """ + try: + status, said = await (run or _run)([executable, *DAEMON_VERSION_ARGUMENTS]) + except TimeoutError: + # Before `OSError`, and that ordering is the behaviour rather than a + # style: `TimeoutError` **is** an `OSError`, so caught second a daemon + # that hung would be reported as a `codex` that is not installed — the + # one reason that sends somebody looking in the wrong place. + return None, ( + f"{executable} did not answer where the shared Codex daemon is within " + f"{COMMAND_TIMEOUT_SECONDS:.0f} seconds" + ) + except OSError as unrunnable: + return None, f"{executable} could not be run: {unrunnable}" + if status != 0: + # The last line, because a `codex` refusal is a short reason under a + # longer "Error:" banner and the reason is the part worth carrying. + reason = said.splitlines()[-1].strip() if said.strip() else "it gave no reason" + return None, f"the shared Codex daemon is not answering: {reason}" + try: + reported: Any = json.loads(said) + except json.JSONDecodeError: + return None, ( + "the shared Codex daemon answered, and not with JSON: " + f"{said[:UNREADABLE_ANSWER_QUOTED_CHARS]}" + ) + if not isinstance(reported, dict): + return None, "the shared Codex daemon answered with something that is not a document" + socket_path = reported.get(SOCKET_PATH_FIELD) + if not isinstance(socket_path, str) or not socket_path.strip(): + return None, ( + f"the shared Codex daemon answered without a {SOCKET_PATH_FIELD}, and this " + "engine will not guess one" + ) + return ( + DaemonAddress( + socket_path=Path(socket_path.strip()), + cli_version=_said(reported.get(CLI_VERSION_FIELD)), + app_server_version=_said(reported.get(APP_SERVER_VERSION_FIELD)), + ), + "", + ) + + +class SharedDaemon: + """One connection to the daemon somebody else owns, held for as long as it lives.""" + + def __init__( + self, + *, + settings: CodexSettings, + version: str, + locate: Callable[[str], Awaitable[tuple[DaemonAddress | None, str]]] = locate, + attach: Callable[..., Awaitable[Any]] = attach, + ) -> None: + self._settings = settings + self._version = version + self._locate = locate + self._attach = attach + self._connection: AppServerConnection | None = None + self._note = "" + #: Held across the whole dial, because the dial is where the race is. + #: The engine has two callers that arrive independently — the five-second + #: discovery cadence and a control-plane `progress` ask — and the check + #: for a live connection is separated from writing the new one by two + #: awaits, which is room enough for both to find none and both to attach. + self._dialling = asyncio.Lock() + #: Bumped by every `aclose`. A dial that finishes on an older number is + #: one the engine has already said goodbye to, so it closes what it made + #: rather than writing it back. This is what lets `aclose` skip the lock + #: instead of waiting behind a dial — see its docstring for why it must. + self._generation = 0 + + @property + def note(self) -> str: + """What the lane should say about rows read through this, if anything. + + Empty when there is nothing to say. It is deliberately not the same as + "there is no connection": a daemon that answers with a mismatched version + is joined *and* worth a caveat, and one that is simply absent is a + caveat with no connection behind it. + """ + return self._note + + async def client(self) -> AppServerConnection | None: + """A live connection to the daemon, or `None` with `note` saying why not. + + A connection the far side dropped is not reused: the daemon can be + restarted under a running engine — its own updater does exactly that — + and an engine that kept a dead handle would report an empty roster for + the rest of its life. + + **One dial at a time, and the answer is asked for twice.** A live + connection is answered without taking the lock, which is the ordinary + case and stays free; a caller that finds none waits, and then asks again + — because what it was waiting on is very likely the dial that answers its + own question. Without this, two callers arriving together both attached, + the daemon held two clients of an engine that is meant to be one of them, + and the loser was dropped with nothing left to close it. + """ + held = self._connection + if held is not None and held.is_open: + return held + + async with self._dialling: + held = self._connection + if held is not None and held.is_open: + return held + self._connection = None + began = self._generation + + address, reason = await self._locate(self._settings.executable) + if address is None: + self._note = reason + return None + try: + connection = await self._attach( + address.socket_path, version=self._version, settings=self._settings + ) + except (WireError, AppServerError, OSError) as unreachable: + self._note = ( + f"the shared Codex daemon at {address.socket_path} did not accept a " + f"connection: {unreachable}" + ) + return None + if self._generation != began: + # Let go of while this was in flight. A dial does not get to + # resurrect a connection the engine has already said goodbye to, + # and what it made is closed here rather than left for nobody. + await connection.aclose() + return None + _log.info("joined the shared Codex daemon at %s", address.socket_path) + self._connection = connection + self._note = address.note + return connection + + async def aclose(self) -> None: + """Let go of this engine's end. The daemon and its Sessions carry on. + + **It never waits for a dial in flight, and that is #96's arithmetic + rather than a preference.** `runner.SHUTDOWN_SECONDS` is sixteen seconds + because the phases it bounds sum to 13.2 and it must exceed them, and a + test computes that sum from the constants themselves. A shutdown phase + that could sit out a ten-second daemon lookup is not one that sum has + room for — and an overrun does not buy a tidier stop, it buys a SIGKILL + with nothing written down, which is the failure #96 exists to end. + + So the dial is **invalidated instead of waited on**: the generation + moves, and a dial that finishes afterwards closes what it made rather + than writing it back. Clearing the field alone would have left exactly + that client behind, with nothing holding it and nothing to close it. + """ + self._generation += 1 + connection, self._connection = self._connection, None + self._note = "" + if connection is not None: + await connection.aclose() + + +def _said(value: Any) -> str: + """One version field as a string, and an unsaid one as nothing.""" + return value.strip() if isinstance(value, str) else "" + + +async def _run(arguments: list[str]) -> tuple[int, str]: + """Run one short command and take what it said, however it ended. + + `stderr` is folded into `stdout` because a `codex` that refuses writes its + reason to one and its document to the other, and the caller wants whichever + turned up. + + **The bound lives here, in the runner, rather than at the call site**, so a + caller injecting a `Runner` of its own is the only thing that can be + unbounded — and the only callers that do are tests. Raises `TimeoutError`, + which `locate` turns into a reason like every other way this can fail. + """ + process = await asyncio.create_subprocess_exec( + *arguments, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + try: + said, _ = await asyncio.wait_for(process.communicate(), COMMAND_TIMEOUT_SECONDS) + except TimeoutError: + # Killed rather than left, because the wait ending is not the command + # ending: a `codex` abandoned here would go on holding whatever it was + # stuck on, and the next tick would start another one beside it. + process.kill() + await process.wait() + raise + return process.returncode or 0, said.decode("utf-8", errors="replace") diff --git a/src/gpt_voicecoding/adapters/agent/codex/thread_tail.py b/src/gpt_voicecoding/adapters/agent/codex/thread_tail.py new file mode 100644 index 0000000..be1f365 --- /dev/null +++ b/src/gpt_voicecoding/adapters/agent/codex/thread_tail.py @@ -0,0 +1,140 @@ +"""How far along one Codex thread is, read out of what the daemon says about it. + +No I/O. It is handed the `thread` document a `thread/read` answered and returns +the same two facts the Claude lane's `transcript_tail` returns, bounded by the +same shared rule (`adapters/agent/_progress.py`) — one bound, one type, whichever +lane the row came from. + +**Attached threads only, and that is a rule rather than a limitation.** A Codex +Session the shared daemon does not hold has no turns to read: its rollout is on +disk, but reading it would be a second source answering the same question with +worse evidence, and the port table left that behind explicitly ("no rollout +reading for daemon-attached threads" — and for unattached ones, no manufactured +progress at all). Such a row carries `progress=None`, which is "not read", not +"read and found nothing". + +**Against legacy** (ADR 0010, `CLAUDE.md`). The item selection is **ported** +whole from `legacy@1d32845:bridge/codex.py:1465-1520`: `agentMessage` is what +the agent said, a `userMessage`'s `text` content is what it was told, and every +other item type is the machinery of doing the work rather than a report of it. +**Adapted** in one place, deliberately: legacy *raised* on an item type it did +not recognise (`:1519`), which was right for a verb answering about one Session +and is wrong here, because the same reading now rides on a roster row and one +unknown item would blank a row the user is looking at. So an unreadable item +costs itself and nothing else — which is the rule legacy already used on its +Claude side (`bridge/transcript.py:1568-1580`), applied to both lanes. +**Dropped**: `turn_id` and `turn_status` beside each entry (`:1484-1492, +1516-1520`), because no v1.0 consumer reads them — the Live Call, the Companion +Channel and the Control Panel ask what a Session last said and what it was last +told, never which turn that was — and `ProgressEntry` can gain a field later +without `Progress` widening twice. + +**Times are epoch seconds, measured not assumed.** `thread/read` on codex +0.149.1 answers `updatedAt`, `recencyAt` and `createdAt` as integers — +`1787712279` for a thread last touched at 2026-08-26T02:44:39Z — read off the +live daemon on 2026-08-26. `updatedAt` is `last_activity`: it moves when the +thread does, including while a turn is running, which is exactly the case the +field exists for. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import UTC, datetime +from typing import Any, Final + +from gpt_voicecoding.adapters.agent._progress import RECENT_LIMIT, RECENT_MAX_BYTES, bounded +from gpt_voicecoding.seams.agent import ProgressEntry, ProgressRole + +#: What the agent said, and what it was told. Everything else a turn holds — +#: `reasoning`, `commandExecution`, `fileChange`, `plan`, the tool calls, the +#: review-mode markers, `contextCompaction` — is the machinery of doing the work +#: and never something the user was shown +#: (`legacy@1d32845:bridge/codex.py:71-86,1479-1520`). +AGENT_ITEM: Final = "agentMessage" +USER_ITEM: Final = "userMessage" + +#: Which parts of a user message are words. `image` and `localImage` are the +#: other two shapes a `userMessage` carries and neither has any to contribute. +USER_TEXT: Final = "text" + +#: When the thread was last touched, as codex spells it. +UPDATED_AT: Final = "updatedAt" + + +def recent( + thread: Mapping[str, Any], + limit: int = RECENT_LIMIT, + *, + max_bytes: int = RECENT_MAX_BYTES, +) -> tuple[tuple[ProgressEntry, ...], bool]: + """The newest of what this thread said, and whether anything older was dropped. + + A document with no `turns` list yields nothing rather than raising: it is + what a `thread/read` answers when turns were not asked for, and a caller that + took that as an error would turn the cheap roster read into a failure. + """ + turns = thread.get("turns") + if not isinstance(turns, list): + return (), False + entries: list[ProgressEntry] = [] + for turn in turns: + if not isinstance(turn, Mapping): + continue + items = turn.get("items") + if not isinstance(items, list): + continue + entries.extend(entry for entry in map(_entry, items) if entry is not None) + return bounded(entries, limit, max_bytes=max_bytes) + + +def moment(value: Any) -> datetime | None: + """One of the thread's own times as a moment, or nothing. + + Epoch seconds, timezone-aware in UTC. A `bool` is refused before the number + check because `True` is an `int` in Python and would read as one second past + the epoch — a Session last active in 1970 is the kind of nonsense a surface + renders without blinking. + """ + if isinstance(value, bool) or not isinstance(value, int | float): + return None + try: + return datetime.fromtimestamp(value, UTC) + except (OSError, OverflowError, ValueError): + return None + + +def last_activity(thread: Mapping[str, Any]) -> datetime | None: + """When this thread last moved, by its own account.""" + return moment(thread.get(UPDATED_AT)) + + +def _entry(item: Any) -> ProgressEntry | None: + """One turn item as something that was said, or `None` if it was not.""" + if not isinstance(item, Mapping): + return None + kind = item.get("type") + if kind == AGENT_ITEM: + text = item.get("text") + return _said(ProgressRole.ASSISTANT, text if isinstance(text, str) else "") + if kind == USER_ITEM: + return _said(ProgressRole.USER, _user_text(item.get("content"))) + return None + + +def _user_text(content: Any) -> str: + """The words in one `userMessage`, and nothing else it carried.""" + if not isinstance(content, list): + return "" + parts = [ + part.get("text") + for part in content + if isinstance(part, Mapping) + and part.get("type") == USER_TEXT + and isinstance(part.get("text"), str) + ] + return "".join(part for part in parts if part) + + +def _said(role: ProgressRole, text: str) -> ProgressEntry | None: + return ProgressEntry(role=role, text=text) if text.strip() else None diff --git a/src/gpt_voicecoding/control_plane/actions.py b/src/gpt_voicecoding/control_plane/actions.py index 15eafad..8b0ccb1 100644 --- a/src/gpt_voicecoding/control_plane/actions.py +++ b/src/gpt_voicecoding/control_plane/actions.py @@ -72,6 +72,7 @@ def __init__(self, core: BridgeCore) -> None: Action.STATUS: self._status, Action.SWITCH: self._switch, Action.SESSIONS: self._sessions, + Action.PROGRESS: self._progress, Action.LIVE: self._live, Action.RELAY: self._relay, Action.APPROVE: self._approve, @@ -116,6 +117,20 @@ async def _status(self, payload: Mapping[str, Any]) -> dict[str, Any]: async def _sessions(self, payload: Mapping[str, Any]) -> dict[str, Any]: return {"sessions": payloads.status_document(self._core.status())["sessions"]} + async def _progress(self, payload: Mapping[str, Any]) -> dict[str, Any]: + """How far along one exact Session is, rendered as one roster row. + + The same document `sessions` renders, deliberately: a surface that had to + learn a second shape to show one Session's progress would be a second + reader of the same facts. What this action adds is *when* — the row comes + back read at the moment it was asked for, rather than at the last tick. + """ + return { + "session": payloads.session_document( + await self._core.progress(payloads.read_target(payload)) + ) + } + async def _switch(self, payload: Mapping[str, Any]) -> dict[str, Any]: name = payloads.read_text(payload, "name") on = payloads.read_flag(payload, "on") diff --git a/src/gpt_voicecoding/control_plane/commands.py b/src/gpt_voicecoding/control_plane/commands.py index 8374acb..6db6d81 100644 --- a/src/gpt_voicecoding/control_plane/commands.py +++ b/src/gpt_voicecoding/control_plane/commands.py @@ -29,6 +29,7 @@ Action.STATUS: "status", Action.SWITCH: "switch on|off", Action.SESSIONS: "sessions", + Action.PROGRESS: "progress :[:]", Action.LIVE: "live", Action.RELAY: "relay :[:] [--supplement] ", Action.APPROVE: "approve allow|deny|ask", @@ -69,6 +70,9 @@ def _payload(action: Action, arguments: list[str]) -> dict[str, object]: case Action.APPROVE: approval_id, verdict = _exactly(action, arguments, 2) return {"approval_id": approval_id, "verdict": verdict} + case Action.PROGRESS: + (address,) = _exactly(action, arguments, 1) + return {"target": parse_address(address)} raise CommandError(f"no command called {action!r}") # unreachable: the set is closed @@ -145,6 +149,8 @@ def render(reply: Reply) -> str: return "\n".join(_status_lines(data)) case Action.SESSIONS: return "\n".join(_roster_lines(data["sessions"])) + case Action.PROGRESS: + return "\n".join(_progress_lines(data["session"])) case Action.SWITCH: was = "on" if data["previous"] else "off" return f"{data['name']} is {'on' if data['on'] else 'off'} (was {was})" @@ -192,6 +198,31 @@ def _roster_lines(sessions: object) -> list[str]: ] +def _progress_lines(session: object) -> list[str]: + """One Session's own words, newest last, and when they were read. + + `read_at` is said out loud rather than left implicit: a progress line's whole + meaning is when it was true, and a surface that printed it bare would let a + reading taken before a five-minute silence read as one taken just now. + """ + assert isinstance(session, dict) + #: The same one-line summary the roster prints for this Session, without the + #: `sessions:` heading a list of them carries — so the two surfaces cannot + #: describe one Session two ways. + lines = _roster_lines([session])[1:] + progress = session["progress"] + if progress is None: + return [*lines, " progress: not read"] + lines.append(f" last activity: {session['last_activity'] or 'not read'}") + if progress["truncated"]: + lines.append(" (older entries dropped)") + lines.extend(f" {entry['role']}: {entry['text']}" for entry in progress["recent"]) + if not progress["recent"]: + lines.append(" nothing said yet") + lines.append(f" read at {progress['read_at']}") + return lines + + def _lane_lines(lanes: object) -> list[str]: """Said only when a lane is down, because silence is what "fine" looks like.""" if not isinstance(lanes, dict) or not lanes: diff --git a/src/gpt_voicecoding/control_plane/payloads.py b/src/gpt_voicecoding/control_plane/payloads.py index cf2bb66..12542ac 100644 --- a/src/gpt_voicecoding/control_plane/payloads.py +++ b/src/gpt_voicecoding/control_plane/payloads.py @@ -185,7 +185,10 @@ def progress_document(progress: Progress | None) -> dict[str, Any] | None: if progress is None: return None return { - "recent": list(progress.recent), + # Each entry says which side spoke it. Carried rather than inferred: a + # roster of bare strings reads "make it blue" and "I made it blue" the + # same way, and every surface would have to guess (#76). + "recent": [{"role": str(entry.role), "text": entry.text} for entry in progress.recent], "truncated": progress.truncated, "read_at": progress.read_at.isoformat() if progress.read_at is not None else None, } diff --git a/src/gpt_voicecoding/core/bridge.py b/src/gpt_voicecoding/core/bridge.py index e1fd210..1dbc71a 100644 --- a/src/gpt_voicecoding/core/bridge.py +++ b/src/gpt_voicecoding/core/bridge.py @@ -35,7 +35,13 @@ from gpt_voicecoding.core.adjudication import SwitchAdjudicator from gpt_voicecoding.core.approvals import ApprovalOutcome, ApprovalPipeline, PendingApproval from gpt_voicecoding.core.clock import Clock, default_clock, wall_clock -from gpt_voicecoding.core.errors import BridgeCoreError, UnknownRelayError +from gpt_voicecoding.core.errors import ( + BridgeCoreError, + LaneUnreadable, + ProgressUnavailable, + StaleSessionError, + UnknownRelayError, +) from gpt_voicecoding.core.escalation import EscalationPipeline, Notice from gpt_voicecoding.core.events import EventQueue from gpt_voicecoding.core.instructions import InstructionContext, Instructions, generate @@ -60,11 +66,13 @@ AgentAdapter, ApprovalVerdict, AwaitingApproval, + LaneUnavailable, RelayReceipt, RelayRoute, ReplyWindow, ReplyWindowChanged, SessionEnded, + SessionLifecycle, SessionState, SessionStopped, WaitingFor, @@ -287,6 +295,61 @@ def status(self) -> Status: pending_approvals=self.approvals.pending(), ) + async def progress(self, target: SessionTarget) -> Session: + """How far along one exact Session is, read now. Never starts a turn. + + A hub verb, and a *read*: it resolves one identity, asks that lane and + no other, and returns the same `Session` row `status` renders. The + reference implementation's own rule, ported — + `legacy@1d32845:bridge/daemon.py:2202-2271` resolved one exact registered + identity, asked only that agent's own authority, and never fell back to + another lane, a terminal or a screen when its source could not answer. + + **It is not a Relay and it costs no turn.** The router says so of the + whole class (`core/router.py:31-32`) and the seam keeps it true: + `inspect` reads what the agent has already written down. + + **Three more refusals, and each is a different fact** — #76 asks for an + honest error rather than an answer that says nothing: + + - *The lane could not be read.* `LaneUnreadable`, carrying the lane's own + words. The roster's row is left exactly as it stood: not being able to + look is not a sighting. + - *The Session has ended.* `StaleSessionError`. The row is **not** ended + here: `SessionRegistry.observe` is the one component that ends rows, + and the value an `inspect` returns for a Session it could not find + carries no workspace and no name — folding it in would strip the very + fields a surface needs to say what happened to it. The next discovery + ends it properly, within one cadence. + - *Nothing could read how far it has got.* `ProgressUnavailable` — an + unattached Codex Session, or one whose first turn has written no record + yet. A Session that *was* read and had said nothing is not this: it + answers normally, with an empty reading. + + Two further refusals are the resolver's and are not restated here — an + identity nobody registered, and a Child Process, which is seen and never + spoken to (#68). + + Whatever is read is folded back into the roster before it is answered, so + a surface that asks for progress and then asks for `status` cannot be + told two different things about one Session. + """ + session = self._state.sessions.resolve(target) + adapter = self._agents.get(target.agent) + if adapter is None: + raise LaneUnreadable(str(target.agent), "this engine has no adapter for that agent") + try: + row = await adapter.inspect(session.target) + except LaneUnavailable as unread: + raise LaneUnreadable(str(unread.agent), unread.reason) from None + + if row.lifecycle is not SessionLifecycle.LIVE: + raise StaleSessionError(target, reason=f"that Session is {row.lifecycle}") + read = self._state.sessions.observed_one(row, now=self._stamp()) + if read.progress is None: + raise ProgressUnavailable(target) + return read + async def flip_switch(self, name: str, on: bool) -> bool: """Flip a switch and report the state it held before. diff --git a/src/gpt_voicecoding/core/errors.py b/src/gpt_voicecoding/core/errors.py index 086178f..e26ed9b 100644 --- a/src/gpt_voicecoding/core/errors.py +++ b/src/gpt_voicecoding/core/errors.py @@ -77,6 +77,34 @@ def __init__(self, target: SessionTarget, parent: SessionTarget | None = None) - self.parent = parent +class ProgressUnavailable(SessionError): + """Nothing on this machine can say how far that Session has got. + + The honest error #76 asks the `progress` verb for, and it is a refusal rather + than an answer carrying no progress for the reason the whole ticket turns on: + "nobody could read it" and "it has said nothing" are different facts, and a + surface handed the first as the second reports a working Session as an idle + one. A Session that *was* read and had said nothing comes back as an ordinary + answer with an empty reading. + + Two ways to arrive here, and neither may be papered over: a Codex Session the + shared daemon does not hold — its rollout is on disk and reading it would be + a second source answering with worse evidence (port table P6, P13) — and a + Session whose first turn has not written a record yet (#73). + + **Ported** from the reference implementation, which rejected the same verb + with the source's own reason rather than falling back to a terminal, a screen + or the other agent (`legacy@1d32845:bridge/daemon.py:2222-2246`). + """ + + def __init__(self, target: SessionTarget) -> None: + super().__init__( + f"nothing has read how far {target} has got: this engine reads a Session's " + "own record and never infers one" + ) + self.target = target + + class DuplicateSessionError(SessionError): """That identity is already registered. Registering it again would split truth.""" @@ -167,3 +195,30 @@ class VoiceInstructionsMissing(BridgeCoreError): def __init__(self) -> None: super().__init__("this engine generated no voice instructions, so it cannot start a call") + + +class LaneUnreadable(BridgeCoreError): + """A lane could not be read at all, so nothing can be said about its Sessions. + + The hub's translation of the Agent seam's `LaneUnavailable`, which is the one + thing on that seam that raises. It is deliberately *not* an answer with empty + progress: "I could not look" and "it has said nothing" are different facts, + and a surface that rendered the first as the second would report a working + Session as an idle one every time `claude` fell off the PATH. + + It carries the lane's own words rather than a rephrasing, and the row it was + asked about is left exactly as the roster last saw it (`seams/agent.py`, + `LaneUnavailable`). + + **Ported** from the reference implementation's own rule for the same verb: + a progress source that could not answer produced a rejection carrying its + reason, and never a fallback to a terminal, a screen or the other agent + (`legacy@1d32845:bridge/daemon.py:2222-2246`). **Adapted**: legacy's + rejection was a reply document built by the daemon, and this is a refusal the + control plane maps onto the closed error set. + """ + + def __init__(self, agent: str, reason: str) -> None: + super().__init__(f"the {agent} lane could not be read: {reason}") + self.agent = agent + self.reason = reason diff --git a/src/gpt_voicecoding/core/instructions/delegated.py b/src/gpt_voicecoding/core/instructions/delegated.py index 2722346..f82645a 100644 --- a/src/gpt_voicecoding/core/instructions/delegated.py +++ b/src/gpt_voicecoding/core/instructions/delegated.py @@ -36,6 +36,7 @@ Action.STATUS: "everything the engine knows right now: switches, sessions, call, pending work", Action.SWITCH: "flip one switch on or off", Action.SESSIONS: "the session roster on its own", + Action.PROGRESS: "how far along one exact session is, read now — never a turn", Action.LIVE: "the Live Toggle — end the call that is up, or start one when none is", Action.RELAY: "carry the user's own words into one exact session", Action.APPROVE: "carry the user's verdict on one pending permission request", diff --git a/src/gpt_voicecoding/core/sessions.py b/src/gpt_voicecoding/core/sessions.py index 902c8ed..6fc78e3 100644 --- a/src/gpt_voicecoding/core/sessions.py +++ b/src/gpt_voicecoding/core/sessions.py @@ -246,7 +246,7 @@ def observe( row.target, ) continue - seen.add(self._admit(row, now=now).target) + seen.add(self.observed_one(row, now=now).target) ended: list[SessionTarget] = [] for held in [row for row in self._of(agent) if row.target not in seen]: @@ -274,8 +274,16 @@ def lane_degradations(self) -> dict[AgentKind, str]: """ return dict(self._lane_degradations) - def _admit(self, row: SessionInspection, *, now: float) -> Session: - """Fold one freshly-read row into the roster, in place where it belongs.""" + def observed_one(self, row: SessionInspection, *, now: float) -> Session: + """Fold one freshly-read row into the roster, in place where it belongs. + + Public because the per-target read is a caller too (#76's `progress` + verb): it looked at one Session and at no others, so it may correct that + Session's entry and may conclude nothing about the rest. **`observe` is + the whole-lane verb and stays the only one that ends a row** — a verb + that ended rows because it was asked about one Session would end the + whole lane every time somebody asked about one of its members. + """ held = self._same_row(row) if held is None: fresh = session_from(row, first_seen=now) diff --git a/src/gpt_voicecoding/seams/agent.py b/src/gpt_voicecoding/seams/agent.py index c3dfef7..dcb6664 100644 --- a/src/gpt_voicecoding/seams/agent.py +++ b/src/gpt_voicecoding/seams/agent.py @@ -210,6 +210,41 @@ def as_approval_request(self, target: SessionTarget) -> ApprovalRequest | None: ) +class ProgressRole(StrEnum): + """Which side said one progress entry. + + **Carried rather than inferred.** A roster of bare strings reads "make it + blue" and "I made it blue" the same way, so every surface would have to + guess — which is the per-consumer parsing #74 exists to end. The reference + implementation carried the role all the way to the wire + (`legacy@1d32845:bridge/transcript.py:40-46`), and this is that fact, named. + """ + + USER = "user" + ASSISTANT = "assistant" + + +@dataclass(frozen=True, slots=True) +class ProgressEntry: + """One thing that was said in a Session, and by which side. + + Deliberately just the two fields. Legacy carried `turn_id` and `turn_status` + beside them on the Codex lane (`legacy@1d32845:bridge/codex.py:1484-1492, + 1516-1520`); **dropped, because** no v1.0 consumer reads them — the Live + Call, the Companion Channel and the Control Panel ask what a Session last + said and what it was last told, never which turn that was or how the turn + ended — and adding a field here later is additive rather than a second + widening of `Progress`. + """ + + role: ProgressRole + text: str + + def __post_init__(self) -> None: + if not self.text.strip(): + raise ValueError("an entry with nothing said in it is not progress") + + @dataclass(frozen=True, slots=True) class Progress: """How far along a Session is, in its own words, as of one reading. @@ -220,7 +255,7 @@ class Progress: that renders "3 steps" when there were thirty is worse than one that says so. """ - recent: tuple[str, ...] = () + recent: tuple[ProgressEntry, ...] = () truncated: bool = False read_at: datetime | None = None diff --git a/src/gpt_voicecoding/seams/control_plane.py b/src/gpt_voicecoding/seams/control_plane.py index 72ae329..8bff656 100644 --- a/src/gpt_voicecoding/seams/control_plane.py +++ b/src/gpt_voicecoding/seams/control_plane.py @@ -38,6 +38,13 @@ #: once every seam a hub verb reaches became one configuration requires. A #: surface built against 3 would send an action this engine no longer has, so #: the disagreement has to be visible before the request is made. +#: +#: **Not bumped for `progress` (#76).** A version exists so a surface can tell an +#: engine that disagrees from one too old to have been asked, and the failure it +#: guards against is a surface sending something the engine does not have. +#: Adding an action cannot cause that: a surface built against 4 never sends it, +#: and one that does is talking to an engine that has it. Removing one can, which +#: is what 4 was for. PROTOCOL_VERSION = 4 #: The longest line either side will read. Generous for a roster, small enough @@ -59,6 +66,9 @@ class Action(StrEnum): SWITCH = "switch" #: The Session roster on its own, for a surface that only renders that. SESSIONS = "sessions" + #: How far along one exact Session is, read now. A status query like every + #: other one here: it never starts a turn and never touches the Session. + PROGRESS = "progress" #: The Live Toggle: end the call the system owns, or start one if none is up. LIVE = "live" #: An Answer Relay — the user's own words, for one exact Session. diff --git a/tests/acceptance/journey.py b/tests/acceptance/journey.py index 5df32b1..4fed75f 100644 --- a/tests/acceptance/journey.py +++ b/tests/acceptance/journey.py @@ -461,12 +461,22 @@ def progress(self) -> str: it does not make the Session work. The second is checked the only way it can be from outside — the agent's own record does not grow across the read, and the roster's state does not leave `idle`. + + **Both surfaces are exercised, because #76 built both and they are one + reading.** `bridgectl progress ` is the verb the ticket names, + and it reads the Session *now*; the roster row carries the same fields so + a Control Panel can render them without asking a second question. The + step fails if either is absent, and fails if they disagree. """ if self.address is None: raise LaneBlocked("no Session to read progress from") before_size = self._record_size() before_state = self._roster_field("state") + answer = self.bridgectl("progress", self.address) + if not answer.ok: + raise StepFailed(f"`bridgectl progress {self.address}` refused: {answer.text}") + row = self._roster_row() if row is None: raise StepFailed(f"{self.address} left the roster before progress could be read") @@ -479,6 +489,17 @@ def progress(self) -> str: reported = row["progress"] if not reported or not (reported.get("recent") if isinstance(reported, dict) else None): raise StepFailed(f"progress for {self.address} is {reported!r} after a turn that ran") + said = support.flatten( + f"{entry.get('role')}: {entry.get('text')}" + for entry in reported["recent"] + if isinstance(entry, dict) + ) + newest = reported["recent"][-1] + if isinstance(newest, dict) and str(newest.get("text", ""))[:40] not in answer.text: + raise StepFailed( + f"the verb and the roster describe {self.address} differently — " + f"`bridgectl progress` said {answer.text[:200]!r} and the row says {said[:200]!r}" + ) time.sleep(2.0) after_size = self._record_size() @@ -489,8 +510,9 @@ def progress(self) -> str: f"{after_size} bytes — that is a turn, and #76 forbids one" ) return ( - f"progress read without a turn: {support.flatten(reported.get('recent'))[:160]!r}; " - f"record steady at {after_size} bytes; state {before_state!r} → {after_state!r}" + f"progress read without a turn, through `bridgectl progress` and the roster row: " + f"{said[:160]!r}; record steady at {after_size} bytes; " + f"state {before_state!r} → {after_state!r}" ) # --- stop notice ------------------------------------------------------ diff --git a/tests/fakes.py b/tests/fakes.py index 6234771..af71a97 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -29,6 +29,8 @@ RelayRoute, ReplyWindow, SessionInspection, + SessionLifecycle, + SessionState, ) from gpt_voicecoding.seams.call import CallSnapshot, CallState, DelegatedReply from gpt_voicecoding.seams.delivery import Delivery, DeliveryReceipt @@ -91,6 +93,13 @@ def __init__( self.discovery = LaneDiscovery() #: How many times the hub asked. A cadence is a thing a test asserts on. self.discoveries = 0 + #: Every target `inspect` was asked about, in order. The per-target read + #: is a different verb from the cadence and is asserted separately (#76). + self.inspections: list[SessionTarget] = [] + #: What `inspect` raises instead of answering. `LaneUnavailable` is the + #: one thing on this seam that raises, and a lane that could not look + #: must not be answerable as a Session that said nothing. + self.inspect_raises: Exception | None = None def supported_routes(self) -> frozenset[RelayRoute]: return self._routes @@ -145,11 +154,25 @@ async def discover(self) -> LaneDiscovery: return self.discovery async def inspect(self, target: SessionTarget) -> SessionInspection: - """The row this fake already holds for that target, or a bare live one.""" + """The row this fake holds for that target, or the honest `ENDED`. + + A lane that looked and did not find the Session answers `ENDED`, which is + what both real adapters answer and what Bridge Core branches on. A fake + that returned a bare live row instead would let a caller pass a test the + product fails. + """ + self.inspections.append(target) + if self.inspect_raises is not None: + raise self.inspect_raises for row in self.discovery.rows: if row.target == target: return row - return SessionInspection(target=target, workspace=Path("/nowhere")) + return SessionInspection( + target=target, + workspace=Path("/nowhere"), + lifecycle=SessionLifecycle.ENDED, + state=SessionState.IDLE, + ) async def verify(self) -> VerifyResult: return self.verify_result diff --git a/tests/test_app_bundle.py b/tests/test_app_bundle.py index 2420130..7a1a18b 100644 --- a/tests/test_app_bundle.py +++ b/tests/test_app_bundle.py @@ -22,7 +22,7 @@ from app_bundle.plan import BuildPlan from gpt_voicecoding import config -from gpt_voicecoding.seams.control_plane import PROTOCOL_VERSION +from gpt_voicecoding.seams.control_plane import PROTOCOL_VERSION, Action #: What every binary in a current arm64 tree actually starts with. MACH_O = b"\xcf\xfa\xed\xfe" @@ -415,6 +415,21 @@ def test_the_shell_and_the_engine_speak_the_same_protocol_version(self) -> None: assert declaration is not None, "Wire.swift no longer declares the supported protocol" assert int(declaration["version"]) == PROTOCOL_VERSION + def test_the_shell_and_the_engine_have_the_same_action_set(self) -> None: + """A closed set spelled in two languages is two sets until something says so. + + The engine's set is the contract (`docs/control-plane.md`); the shell + names each one rather than composing strings, so it can offer an action + this engine does not have, or miss one it does, and nothing would fail + until a person clicked. That is `#47`'s shape, one file over. + """ + swift = (inputs.SHELL_PACKAGE / "Sources/ShellCore/Wire.swift").read_text() + body = re.search(r"public enum Action: String.*?\n\}", swift, re.DOTALL) + + assert body is not None, "Wire.swift no longer declares the action set as one enum" + named = set(re.findall(r"^\s*case `?(?P\w+)`?\s*$", body[0], re.MULTILINE)) + assert named == {str(action) for action in Action} + def test_the_shell_and_the_engine_name_the_same_default_socket_path(self) -> None: """The Swift shell and Python engine must meet at the same default address. diff --git a/tests/test_bridgectl.py b/tests/test_bridgectl.py index ccf7dcb..d66fdd3 100644 --- a/tests/test_bridgectl.py +++ b/tests/test_bridgectl.py @@ -24,9 +24,10 @@ from gpt_voicecoding.cli import main from gpt_voicecoding.config import load from gpt_voicecoding.control_plane.client import DEFAULT_TIMEOUT_SECONDS, EngineUnreachable +from gpt_voicecoding.control_plane.commands import render from gpt_voicecoding.engine.composition import Engine from gpt_voicecoding.seams.call import CallSnapshot -from gpt_voicecoding.seams.control_plane import Reply +from gpt_voicecoding.seams.control_plane import Action, Reply #: Longer than any deadline these tests hand the surface, and short enough that #: the engine's own shutdown does not wait on it. It stands in for the real @@ -290,3 +291,91 @@ async def record(request: object, *, path: Path, timeout: float) -> Reply: main([*socket, "status"]) assert waited == [DEFAULT_TIMEOUT_SECONDS, DEFAULT_TIMEOUT_SECONDS] + + +class TestAskingHowFarAlongOneSessionIs: + """#76's command. The rendering is here; the reading is the adapters'.""" + + def test_a_session_that_was_never_registered_cannot_be_asked( + self, engine_at: Path, capsys: pytest.CaptureFixture[str] + ) -> None: + code = main(["--config", str(engine_at), "progress", "codex:never-seen"]) + + assert code == 1 + assert "unknown Session" in capsys.readouterr().err + + def test_it_needs_an_address_and_says_how_to_write_one( + self, engine_at: Path, capsys: pytest.CaptureFixture[str] + ) -> None: + assert main(["--config", str(engine_at), "progress"]) == 2 + assert "progress :" in capsys.readouterr().err + + +class TestRenderingWhatProgressCameBack: + """One reply, several shapes, and none of them may read as another.""" + + def reply(self, progress: object, last_activity: object = None) -> Reply: + return Reply.answered( + Action.PROGRESS, + { + "session": { + "target": {"agent": "codex", "session_id": "abc", "pid": None}, + "label": "GPT-VoiceCoding · build it", + "name": None, + "workspace": "/tmp/workspace", + "state": "idle", + "reply_window": "open", + "progress": progress, + "last_activity": last_activity, + } + }, + ) + + def test_each_entry_says_which_side_spoke_it(self) -> None: + rendered = render( + self.reply( + { + "recent": [ + {"role": "user", "text": "do the thing"}, + {"role": "assistant", "text": "done"}, + ], + "truncated": False, + "read_at": "2026-08-26T02:44:39+00:00", + } + ) + ) + + assert "user: do the thing" in rendered + assert "assistant: done" in rendered + + def test_the_reading_says_when_it_was_taken(self) -> None: + """A progress line's whole meaning is when it was true.""" + rendered = render( + self.reply( + {"recent": [], "truncated": False, "read_at": "2026-08-26T02:44:39+00:00"}, + last_activity="2026-08-26T02:44:39+00:00", + ) + ) + + assert "read at 2026-08-26T02:44:39+00:00" in rendered + assert "nothing said yet" in rendered + + def test_a_dropped_tail_is_admitted_rather_than_implied(self) -> None: + rendered = render( + self.reply( + { + "recent": [{"role": "assistant", "text": "done"}], + "truncated": True, + "read_at": "2026-08-26T02:44:39+00:00", + } + ) + ) + + assert "older entries dropped" in rendered + + def test_not_read_is_never_rendered_as_said_nothing(self) -> None: + """`None` is "nobody looked", and a surface that collapsed the two lies.""" + rendered = render(self.reply(None)) + + assert "progress: not read" in rendered + assert "nothing said yet" not in rendered diff --git a/tests/test_claude_progress.py b/tests/test_claude_progress.py new file mode 100644 index 0000000..9b7e253 --- /dev/null +++ b/tests/test_claude_progress.py @@ -0,0 +1,220 @@ +"""Where the Claude lane's `Progress` and `last_activity` come from (#76). + +The reading itself is pure and tested against fragments in +`test_claude_transcript_tail.py`. This is the wiring: the roster row carries what +the same one transcript read already knows, and the per-target verb answers about +a Session the cadence deliberately skips. + +The two are deliberately different, and both are here: `discover` runs every five +seconds over the whole machine and reads nothing for a Session mid-turn, while +`inspect` is asked about one Session because somebody wants to know now — which +is most often exactly while it works. +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from gpt_voicecoding.adapters.agent.claude import transcript as claude_transcript +from gpt_voicecoding.seams.agent import ( + LaneDiscovery, + ProgressRole, + SessionInspection, + SessionState, + WaitingFor, + WaitingKind, +) +from test_claude_stop_analysis import called, said, turn +from test_claude_stop_wiring import ROSTER_WAITING, TARGET, adapter_holding, roster, transcript + +__all__ = ["roster"] # the fixture is imported, and ruff must see it used + + +def stamped(text: str, *, role: str = "assistant", at: str) -> dict[str, object]: + """One visible record with the `timestamp` Claude Code writes on every one.""" + record = said(text, role=role) + record["timestamp"] = at + return record + + +#: What the roster alone says about a Session it is not calling `waiting`. +NOT_WAITING = WaitingFor() + + +def row(state: SessionState, waiting: WaitingFor = NOT_WAITING) -> SessionInspection: + return SessionInspection( + target=TARGET, workspace=Path("/tmp/workspace"), state=state, waiting_for=waiting + ) + + +class TestTheRosterRow: + """The cheap projection: what the cadence already read, carried on the row.""" + + def test_a_stopped_row_says_what_the_session_has_been_saying( + self, tmp_path: Path, roster + ) -> None: + adapter = adapter_holding(transcript(tmp_path, list(turn()))) + roster(LaneDiscovery(rows=(row(SessionState.IDLE),))) + + found = asyncio.run(adapter.discover()).rows[0] + + assert found.progress is not None + assert [(entry.role, entry.text) for entry in found.progress.recent] == [ + (ProgressRole.USER, "do the thing"), + (ProgressRole.ASSISTANT, "done"), + ] + assert found.progress.truncated is False + + def test_the_reading_says_when_it_was_taken(self, tmp_path: Path, roster) -> None: + """`read_at` is on the value: a progress line's meaning is when it was true.""" + adapter = adapter_holding(transcript(tmp_path, list(turn()))) + roster(LaneDiscovery(rows=(row(SessionState.IDLE),))) + + before = datetime.now(UTC) + found = asyncio.run(adapter.discover()).rows[0] + + assert found.progress is not None + assert found.progress.read_at is not None + assert before <= found.progress.read_at <= datetime.now(UTC) + + def test_last_activity_comes_off_the_record_not_off_the_roster( + self, tmp_path: Path, roster + ) -> None: + """`startedAt` is on the roster row and is deliberately not read as this.""" + adapter = adapter_holding( + transcript(tmp_path, [stamped("done", at="2026-08-26T04:30:09.000Z")]) + ) + roster(LaneDiscovery(rows=(row(SessionState.IDLE),))) + + found = asyncio.run(adapter.discover()).rows[0] + + assert found.last_activity == datetime(2026, 8, 26, 4, 30, 9, tzinfo=UTC) + + def test_a_session_mid_turn_carries_no_progress_and_no_time( + self, tmp_path: Path, roster + ) -> None: + """The gate is #75's, and #76 rides on it rather than widening it. + + `None` is "not read" — a surface may not render it as a Session that has + said nothing, and it may certainly not render a time nobody read. + """ + adapter = adapter_holding(transcript(tmp_path, list(turn()))) + roster(LaneDiscovery(rows=(row(SessionState.RUNNING),))) + + found = asyncio.run(adapter.discover()).rows[0] + + assert found.progress is None + assert found.last_activity is None + + def test_a_session_with_no_transcript_yet_is_unread_rather_than_empty(self, roster) -> None: + """A Session has no transcript file at all until it takes a turn (#73).""" + adapter = adapter_holding(None) + roster(LaneDiscovery(rows=(row(SessionState.IDLE),))) + + found = asyncio.run(adapter.discover()).rows[0] + + assert found.progress is None + + def test_a_session_that_has_written_nothing_readable_is_read_and_empty( + self, tmp_path: Path, roster + ) -> None: + """The other side of that: a file that was read and held no conversation.""" + adapter = adapter_holding( + transcript(tmp_path, [called("Bash", "b1", {"description": "push"})]) + ) + roster(LaneDiscovery(rows=(row(SessionState.WAITING, ROSTER_WAITING),))) + + found = asyncio.run(adapter.discover()).rows[0] + + assert found.progress is not None + assert found.progress.recent == () + + +class TestThePerTargetRead: + """The verb beside it: asked about one Session, whatever it is doing.""" + + def test_a_running_session_can_still_be_asked_how_far_along_it_is( + self, tmp_path: Path, roster + ) -> None: + """The cadence skips it; the verb does not. That is the whole difference.""" + adapter = adapter_holding(transcript(tmp_path, list(turn()))) + roster(LaneDiscovery(rows=(row(SessionState.RUNNING),))) + + found = asyncio.run(adapter.inspect(TARGET)) + + assert found.progress is not None + assert [entry.text for entry in found.progress.recent] == ["do the thing", "done"] + + def test_the_verb_reads_again_without_reopening_an_unchanged_file( + self, tmp_path: Path, roster, monkeypatch: pytest.MonkeyPatch + ) -> None: + """It reads; it does not hand back the cadence's row — and that is free. + + The verb goes through the reader on every ask, so nothing it answers is + older than the moment it was asked. What that costs is a `stat`: the + cache is keyed on the file's own identity, so an unchanged transcript is + not parsed twice, and a hit is the proof of freshness rather than a stale + answer (the advisor's amendment to #76's Q3, 2026-08-26). + """ + adapter = adapter_holding(transcript(tmp_path, list(turn()))) + roster(LaneDiscovery(rows=(row(SessionState.IDLE),))) + parses = 0 + original = claude_transcript._parse # noqa: SLF001 + + def counted(text: str) -> object: + nonlocal parses + parses += 1 + return original(text) + + monkeypatch.setattr(claude_transcript, "_parse", counted) + found = asyncio.run(adapter.inspect(TARGET)) + + assert parses == 1 # `discover` parsed it; the verb re-read a file that had not moved + assert found.progress is not None + + def test_a_transcript_that_moved_between_the_two_is_read_again( + self, tmp_path: Path, roster + ) -> None: + """The other half of the same rule: a changed file is a changed answer.""" + path = transcript(tmp_path, list(turn())) + adapter = adapter_holding(path) + roster(LaneDiscovery(rows=(row(SessionState.IDLE),))) + + asyncio.run(adapter.discover()) + transcript(tmp_path, [*turn(), said("and one more thing", role="user")]) + found = asyncio.run(adapter.inspect(TARGET)) + + assert found.progress is not None + assert [entry.text for entry in found.progress.recent][-1] == "and one more thing" + + def test_a_session_that_is_gone_is_ended_and_says_nothing_about_progress(self, roster) -> None: + """`ENDED` carries no reading, because nobody read one.""" + adapter = adapter_holding(None) + roster(LaneDiscovery()) + + found = asyncio.run(adapter.inspect(TARGET)) + + assert found.progress is None + assert found.last_activity is None + + def test_a_working_session_is_read_for_progress_and_not_for_a_stop( + self, tmp_path: Path, roster + ) -> None: + """An outstanding tool call mid-turn is a tool running, not a dialog (#75). + + The two questions have different answers here, and reading the tail for + both would report a working Session as one waiting on the user. + """ + adapter = adapter_holding( + transcript(tmp_path, [*turn(), called("Bash", "b1", {"description": "push"})]) + ) + roster(LaneDiscovery(rows=(row(SessionState.RUNNING),))) + + found = asyncio.run(adapter.inspect(TARGET)) + + assert found.waiting_for.kind is WaitingKind.NONE + assert found.progress is not None diff --git a/tests/test_claude_transcript_tail.py b/tests/test_claude_transcript_tail.py new file mode 100644 index 0000000..3332dba --- /dev/null +++ b/tests/test_claude_transcript_tail.py @@ -0,0 +1,195 @@ +"""How far along a Claude Session is, read out of the records it wrote. + +The cases are **re-expressed** from what the reference implementation proved +about the same read (`legacy@1d32845:bridge/transcript.py:1184-1246,1568-1607, +2828-2860`), not copied: legacy asserted on a JSON-RPC progress reply built by a +daemon with a store behind it, and this asserts on the one value the pure module +returns. + +Records here are the shape Claude Code really writes — `isSidechain`, `userType`, +`promptSource`, the nested `message.role` and the ISO-8601 `timestamp` included — +because the exclusions under test *are* those fields, and a fixture that omitted +them would test a reader that never sees a real transcript. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from gpt_voicecoding.adapters.agent._progress import RECENT_LIMIT +from gpt_voicecoding.adapters.agent.claude.stop_analysis import QUESTION_TOOL +from gpt_voicecoding.adapters.agent.claude.transcript_tail import recent +from gpt_voicecoding.seams.agent import ProgressRole + +# --- the shapes Claude Code writes ------------------------------------------- + +#: Every fixture stamps from here, so "newest wins" is a fact about order rather +#: than about whichever clock ran the test. +FIRST = datetime(2026, 8, 26, 4, 30, 0, tzinfo=UTC) + + +def stamp(seconds: int) -> str: + """One record's `timestamp`, in the spelling Claude Code writes it.""" + moment = FIRST.replace(second=FIRST.second + seconds) + return moment.isoformat().replace("+00:00", "Z") + + +def said(text: str, *, role: str = "assistant", at: int = 0, **extra: Any) -> dict[str, Any]: + """One record of the visible conversation.""" + return { + "type": role, + "isSidechain": False, + "userType": "external", + "timestamp": stamp(at), + "message": {"role": role, "content": [{"type": "text", "text": text}]}, + **extra, + } + + +def worked(at: int) -> dict[str, Any]: + """A record that moves the Session without putting anything in front of anyone.""" + return { + "type": "user", + "isSidechain": False, + "userType": "external", + "timestamp": stamp(at), + "message": { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "ok"}], + }, + } + + +def texts(records: list[dict[str, Any]]) -> list[str]: + entries, _, _ = recent(records) + return [entry.text for entry in entries] + + +class TestWhatCountsAsProgress: + """P6's visible half: the conversation, and only the conversation.""" + + def test_both_sides_are_carried_and_each_says_which_it_is(self) -> None: + """Legacy carried the role to the wire; #74's type carries it as a field.""" + entries, truncated, _ = recent([said("do the thing", role="user"), said("done", at=1)]) + assert [(entry.role, entry.text) for entry in entries] == [ + (ProgressRole.USER, "do the thing"), + (ProgressRole.ASSISTANT, "done"), + ] + assert truncated is False + + def test_a_child_process_record_is_not_this_session_speaking(self) -> None: + """#68's rule, and 32% of real records (`transcript.py:1131-1133`).""" + assert texts([said("mine"), said("the crew's", at=1, isSidechain=True)]) == ["mine"] + + def test_product_injected_plumbing_is_not_the_user_speaking(self) -> None: + """A `system` prompt source would be reported back as the user's words.""" + assert texts( + [said("mine", role="user"), said("injected", role="user", promptSource="system")] + ) == ["mine"] + + def test_a_slash_command_caveat_is_not_the_user_speaking(self) -> None: + """Command plumbing rides beside the user's turn (`transcript.py:1515-1562`). + + Its content is a bare string and `isMeta` is set, which is how Claude + Code 2.1.246 really writes it — read off a live transcript, not assumed. + """ + caveat = said("mine", role="user") + caveat["message"]["content"] = ( + "Caveat: The messages below were generated by the user " + "while running local commands." + ) + caveat["isMeta"] = True + assert texts([caveat, said("mine", role="user", at=1)]) == ["mine"] + + def test_an_unreadable_item_costs_itself_and_nothing_else(self) -> None: + """Shapes are collected from, never enumerated (`transcript.py:1568-1580`).""" + mixed = said("kept") + mixed["message"]["content"] = [ + {"type": "thinking", "thinking": "not for the user"}, + {"type": "text", "text": "kept"}, + {"type": "image", "source": {}}, + ] + assert texts([mixed]) == ["kept"] + + def test_a_message_with_nothing_readable_is_not_an_entry(self) -> None: + """`tool_result` and `thinking` never had a say; an empty entry is refused.""" + entries, truncated, _ = recent([worked(at=1)]) + assert entries == () + assert truncated is False + + def test_the_options_of_a_question_are_part_of_what_was_said(self) -> None: + """#66: without them the user hears the question and never hears the choices.""" + asking = { + "type": "assistant", + "isSidechain": False, + "userType": "external", + "timestamp": stamp(1), + "message": { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "q1", + "name": QUESTION_TOOL, + "input": { + "questions": [ + { + "question": "Which base?", + "options": [{"label": "main"}, {"label": "feature"}], + } + ] + }, + } + ], + }, + } + assert texts([asking]) == ["Question: Which base?\nOption: main\nOption: feature"] + + +class TestTheBound: + """That this reader is bounded at all, and by the shared rule. + + The rule and its two numbers are `test_progress_bound.py`'s, because both + lanes share one bound (#76). What is here is that this lane goes through it. + """ + + def test_this_reader_goes_through_it(self) -> None: + """One wiring check, and the shared module owns the rule itself. + + Both lanes bound their reading with `_progress.bounded`, so asserting the + drop rule here as well would be one implementation with two sets of tests + that can disagree about it. + """ + entries, truncated, _ = recent( + [said(f"step {index}", at=index) for index in range(RECENT_LIMIT + 2)] + ) + + assert len(entries) == RECENT_LIMIT + assert truncated is True + + +class TestLastActivity: + """Separate from `progress` on purpose (`seams/agent.py:279-281`).""" + + def test_a_session_that_worked_without_speaking_still_moved(self) -> None: + """The whole reason the two are different fields.""" + entries, _, moved = recent([said("done", at=0), worked(at=9)]) + assert [entry.text for entry in entries] == ["done"] + assert moved == FIRST.replace(second=9) + + def test_a_child_process_working_is_the_session_working(self) -> None: + """A sidechain record is not this Session *speaking*; it is it running.""" + _, _, moved = recent([said("mine", at=0), said("the crew's", at=4, isSidechain=True)]) + assert moved == FIRST.replace(second=4) + + def test_records_with_no_readable_time_are_passed_over(self) -> None: + """A half-written tail must not blank the time the rest of the file proves.""" + untimed = said("later", at=0) + untimed["timestamp"] = "not a time" + _, _, moved = recent([said("earlier", at=3), untimed]) + assert moved == FIRST.replace(second=3) + + def test_nothing_read_is_no_time_at_all(self) -> None: + """`None` is "not read", which no consumer may render as "just now".""" + assert recent([]) == ((), False, None) diff --git a/tests/test_codex_discovery.py b/tests/test_codex_discovery.py index 422f66d..2461f48 100644 --- a/tests/test_codex_discovery.py +++ b/tests/test_codex_discovery.py @@ -86,9 +86,19 @@ async def processes() -> tuple[Candidate, ...]: return processes -def found(client: object | None, *candidates: Candidate, home: Path | None = None) -> object: +def found( + client: object | None, + *candidates: Candidate, + home: Path | None = None, + daemon_note: str = "", +) -> object: return asyncio.run( - discover(client, processes=listing(*candidates), home=home) # type: ignore[arg-type] + discover( + client, # type: ignore[arg-type] + processes=listing(*candidates), + home=home, + daemon_note=daemon_note, + ) ) @@ -170,17 +180,38 @@ def test_those_rows_say_where_they_came_from(self) -> None: def test_a_lane_with_no_client_says_so_rather_than_blaming_the_daemon(self) -> None: """#96: "did not answer" was said without a byte being sent. - `CodexAdapter._shared_daemon` returns `None` until #77 builds the - client, so this reading is produced without dialling anything — while - `bridge-install status`, on the same machine, dials the real daemon and - gets a real answer. Saying the daemon was silent made those two look - like a contradiction, and a session went looking for a fault that was - not there. The consequence is identical; the claim is not. + The wording moved when #76 built the client — this build does dial now, + so "this build does not connect yet" stopped being true. It says only + what this process can see, which is the reading the Advisor fixed on #96 + ("consequence for part B"). **#96's rule did not move**, and it is what + is asserted here rather than any spelling: a lane with no client may not + claim the daemon was silent, because being unable to reach it and it + having nothing to say are two facts and only one of them was observed. + The consequence is identical; the claim is not. """ lane = found(None, running(101, "/tmp/w")) assert lane.degraded == discovery.NO_CLIENT - assert "does not connect" in lane.degraded + assert "holds no connection" in lane.degraded assert "did not answer" not in lane.degraded + # The sentence #96 was told not to write, held so it cannot come back. + assert "does not connect" not in lane.degraded + + def test_a_dial_that_failed_says_why_once_and_in_its_own_words(self) -> None: + """The dial's reason replaces the fallback rather than following it (#76). + + `SharedDaemon` sets a reason on every path that answers `None`, and that + reason is always more precise than `NO_CLIENT`. Printed one after the + other, a roster would make two claims about one failure — the shape #96 + is the record of. + """ + lane = found( + None, running(101, "/tmp/w"), daemon_note="codex could not be run: no such file" + ) + + assert lane.degraded is not None + assert lane.degraded.startswith("codex could not be run: no such file") + assert discovery.FROM_THE_MACHINE in lane.degraded + assert "holds no connection" not in lane.degraded def test_a_daemon_that_refuses_is_still_a_daemon_that_did_not_answer(self) -> None: """The other sentence, and the only one that may claim the daemon was silent.""" diff --git a/tests/test_codex_progress.py b/tests/test_codex_progress.py new file mode 100644 index 0000000..53eec99 --- /dev/null +++ b/tests/test_codex_progress.py @@ -0,0 +1,346 @@ +"""Where the Codex lane's `Progress` and `last_activity` come from (#76). + +The reading is pure and tested in `test_codex_thread_tail.py`. This is the +wiring, and its whole subject is **cost**: a `thread/read` with turns answered +558,875 bytes against the real daemon for a thread of two turns, where the same +read without them answered 3,600 (measured on codex 0.149.1, 2026-08-26, and +`numTurns` is not a parameter of that method — passing it changed nothing). So +the roster row is a cached, gated projection and the per-target verb is the live +read, and the tests below are mostly about *which reads happen*. +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime +from pathlib import Path + +from gpt_voicecoding.adapters.agent.codex import CodexAgentAdapter +from gpt_voicecoding.adapters.agent.codex.discovery import TurnCache, discover +from gpt_voicecoding.seams.agent import ( + ProgressRole, + SessionLifecycle, + SessionState, +) +from gpt_voicecoding.seams.identity import AgentKind, SessionTarget +from test_codex_discovery import THREAD, FakeDaemon, running, thread +from test_codex_thread_tail import MEASURED, MEASURED_SECONDS, spoke, told, turn + +WORKSPACE = "/tmp/workspace" + + +class TurnedDaemon(FakeDaemon): + """A daemon that answers turns only when they were asked for, and counts it.""" + + def __init__(self, threads: dict[str, dict], turns: dict[str, list[dict]]) -> None: + super().__init__(threads) + self._turns = turns + #: Every thread id a turn list was asked for, in order. + self.deep: list[str] = [] + + async def request(self, method: str, params: dict | None = None) -> dict: + answer = await super().request(method, params) + if method != "thread/read": + return answer + thread_id = str((params or {}).get("threadId")) + if (params or {}).get("includeTurns"): + self.deep.append(thread_id) + return {"thread": {**answer["thread"], "turns": self._turns.get(thread_id, [])}} + return answer + + +def stopped(**extra: object) -> dict: + return {**thread(THREAD, cwd=WORKSPACE, status="idle"), "updatedAt": MEASURED_SECONDS, **extra} + + +def working(**extra: object) -> dict: + return { + **thread(THREAD, cwd=WORKSPACE, status="active"), + "updatedAt": MEASURED_SECONDS, + **extra, + } + + +def once() -> list[dict]: + return [turn(told("do the thing"), spoke("done"))] + + +async def processes() -> tuple: + return (running(6548, WORKSPACE),) + + +def found(daemon: FakeDaemon, cache: TurnCache | None = None): + return asyncio.run( + discover(daemon, processes=processes, home=Path("/nonexistent"), turns=cache) + ) + + +class TestTheRosterRow: + """The cheap projection: gated on the turn, cached on the thread's own clock.""" + + def test_a_stopped_thread_says_what_it_has_been_saying(self) -> None: + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + + row = found(daemon, TurnCache()).rows[0] + + assert row.progress is not None + assert [(entry.role, entry.text) for entry in row.progress.recent] == [ + (ProgressRole.USER, "do the thing"), + (ProgressRole.ASSISTANT, "done"), + ] + + def test_a_thread_mid_turn_is_not_read_deeply_at_all(self) -> None: + """559 KB per thread per tick is what this gate is worth.""" + daemon = TurnedDaemon({THREAD: working()}, {THREAD: once()}) + + row = found(daemon, TurnCache()).rows[0] + + assert daemon.deep == [] + assert row.state is SessionState.RUNNING + assert row.progress is None + + def test_a_thread_that_has_not_moved_is_not_read_twice(self) -> None: + """Keyed on `updatedAt`: an untouched thread cannot have a changed tail.""" + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + cache = TurnCache() + + first = found(daemon, cache).rows[0] + second = found(daemon, cache).rows[0] + + assert daemon.deep == [THREAD] + assert second.progress == first.progress + + def test_a_thread_that_moved_is_read_again(self) -> None: + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + cache = TurnCache() + + found(daemon, cache) + daemon.threads[THREAD] = stopped(updatedAt=MEASURED_SECONDS + 60) + daemon._turns[THREAD] = [turn(told("do the thing"), spoke("and again"))] # noqa: SLF001 + row = found(daemon, cache).rows[0] + + assert daemon.deep == [THREAD, THREAD] + assert row.progress is not None + assert [entry.text for entry in row.progress.recent][-1] == "and again" + + def test_a_thread_that_names_no_time_is_never_cached(self) -> None: + """Without a clock there is nothing to say it has not moved.""" + daemon = TurnedDaemon({THREAD: thread(THREAD, cwd=WORKSPACE)}, {THREAD: once()}) + cache = TurnCache() + + found(daemon, cache) + found(daemon, cache) + + assert daemon.deep == [THREAD, THREAD] + + def test_a_thread_the_daemon_let_go_is_forgotten(self) -> None: + """The cache is roster-sized, not machine-lifetime-sized.""" + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + cache = TurnCache() + + found(daemon, cache) + cache.retain(set()) + found(daemon, cache) + + assert daemon.deep == [THREAD, THREAD] + + def test_a_thread_the_daemon_could_not_describe_is_unread_rather_than_empty(self) -> None: + """One bad thread is not a bad roster, and it is certainly not an idle one.""" + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + daemon._turns.clear() # noqa: SLF001 + + async def refuse(method: str, params: dict | None = None) -> dict: + if method == "thread/read" and (params or {}).get("includeTurns"): + raise RuntimeError("the daemon dropped the read") + return await FakeDaemon.request(daemon, method, params) + + daemon.request = refuse # type: ignore[method-assign] + row = found(daemon, TurnCache()).rows[0] + + assert row.progress is None + assert row.last_activity == MEASURED + + def test_last_activity_is_free_and_is_there_even_mid_turn(self) -> None: + """It comes off the cheap read, so a working Session still has one.""" + daemon = TurnedDaemon({THREAD: working()}, {THREAD: once()}) + + row = found(daemon, TurnCache()).rows[0] + + assert row.last_activity == MEASURED + assert row.progress is None + + def test_a_lane_with_no_turn_cache_reads_no_turns(self) -> None: + """`discover` without one is still the roster it always was (#74).""" + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + + row = found(daemon).rows[0] + + assert daemon.deep == [] + assert row.progress is None + + +class TestAnUnattachedRow: + """Never inferred: a Session the daemon does not hold has nothing to read.""" + + def test_a_process_table_row_carries_no_progress_and_no_time(self) -> None: + daemon = TurnedDaemon({}, {}) + + rows = found(daemon, TurnCache()).rows + + assert [row.progress for row in rows] == [None] + assert [row.last_activity for row in rows] == [None] + + +class TestTheDaemonNote: + """A version disagreement rides out with the rows rather than hiding them.""" + + def test_the_note_reaches_the_lane_beside_whatever_else_degraded_it(self) -> None: + lane = asyncio.run( + discover( + None, + processes=processes, + home=Path("/nonexistent"), + daemon_note="the Codex CLI is '0.148.0' and the app-server is '0.149.1'", + ) + ) + + assert lane.enumerated + assert lane.degraded is not None + assert "0.148.0" in lane.degraded + assert "process table" in lane.degraded + + def test_a_note_on_its_own_still_degrades_a_lane_that_read_fine(self) -> None: + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + + lane = asyncio.run( + discover( + daemon, + processes=processes, + home=Path("/nonexistent"), + turns=TurnCache(), + daemon_note="the shared Codex daemon did not say its versions", + ) + ) + + assert lane.degraded == "the shared Codex daemon did not say its versions" + + def test_silence_is_what_a_healthy_lane_looks_like(self) -> None: + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + + assert found(daemon, TurnCache()).degraded is None + + +class TestTheReadingSaysWhenItWasTaken: + def test_read_at_is_the_moment_the_turns_were_read(self) -> None: + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + + before = datetime.now(UTC) + row = found(daemon, TurnCache()).rows[0] + + assert row.progress is not None + assert row.progress.read_at is not None + assert before <= row.progress.read_at <= datetime.now(UTC) + + +class _HeldDaemon: + """A `SharedDaemon` that hands back a stand-in, so no real socket is dialled.""" + + def __init__(self, client: object | None) -> None: + self._client = client + self.note = "" + + async def client(self) -> object | None: + return self._client + + async def aclose(self) -> None: + self._client = None + + +def adapter_over(daemon: object | None) -> CodexAgentAdapter: + """A whole adapter over this daemon, enumerating *for real*. + + **Nothing here stubs `discover`, and that is the point.** An earlier version + of this file did, and it hid a defect the reviewer found by hand: `inspect` + runs an enumeration first, so a stubbed one cannot see the read that + enumeration takes — and the verb was paying for two 558,875-byte reads where + the ticket asks for one. The process table is injected rather than stubbed + for the ordinary reason: a test that shelled out to `ps` would answer + differently depending on what the person running it has open. + """ + return CodexAgentAdapter( + daemon=_HeldDaemon(daemon), # type: ignore[arg-type] + processes=processes, + ) + + +class TestThePerTargetRead: + """The verb beside the roster: one Session, whatever it is doing.""" + + def target(self, session_id: str | None = THREAD, pid: int | None = 6548) -> SessionTarget: + """The exact identity the roster gives this thread: the TUI is joined on cwd.""" + return SessionTarget(agent=AgentKind.CODEX, session_id=session_id, pid=pid) + + def test_a_working_thread_can_still_be_asked_how_far_along_it_is(self) -> None: + """The cadence skips it; the verb does not. That is the whole difference.""" + daemon = TurnedDaemon({THREAD: working()}, {THREAD: once()}) + adapter = adapter_over(daemon) + assert asyncio.run(adapter.discover()).rows[0].progress is None # the cadence skipped it + daemon.deep.clear() + + row = asyncio.run(adapter.inspect(self.target())) + + assert daemon.deep == [THREAD] + assert row.progress is not None + assert [entry.text for entry in row.progress.recent] == ["do the thing", "done"] + + def test_one_ask_is_one_deep_read_however_warm_the_cache_is(self) -> None: + """The whole verb, counted end to end — enumeration included. + + `inspect` enumerates before it answers, so the roster's own read is part + of what one ask costs. At 558,875 bytes a read, the difference between + one and two is the finding this test exists to keep closed. + """ + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + adapter = adapter_over(daemon) + asyncio.run(adapter.discover()) # warm the cadence's cache + daemon.deep.clear() + + row = asyncio.run(adapter.inspect(self.target())) + + assert daemon.deep == [THREAD] + assert row.progress is not None + + def test_the_one_read_is_live_and_not_the_remembered_one(self) -> None: + """`updatedAt` is epoch *seconds*, so a warm entry is not proof of freshness. + + Two changes inside one second are one change to that key — tolerable for + a row nobody asked about, and not for one somebody just did. + """ + daemon = TurnedDaemon({THREAD: stopped()}, {THREAD: once()}) + adapter = adapter_over(daemon) + asyncio.run(adapter.discover()) + daemon._turns[THREAD] = [turn(told("do the thing"), spoke("and again"))] # noqa: SLF001 + + row = asyncio.run(adapter.inspect(self.target())) + + assert row.progress is not None + assert [entry.text for entry in row.progress.recent][-1] == "and again" + + def test_a_session_the_daemon_does_not_hold_is_never_guessed_at(self) -> None: + """An unattached row's rollout is on disk; reading it is what P6 left behind.""" + daemon = TurnedDaemon({}, {}) + adapter = adapter_over(daemon) + + row = asyncio.run(adapter.inspect(self.target(None, 6548))) + + assert daemon.deep == [] + assert row.progress is None + + def test_a_daemon_that_is_not_answering_leaves_the_row_as_it_stands(self) -> None: + """No connection, no reading — and certainly no invented one.""" + adapter = adapter_over(None) + + row = asyncio.run(adapter.inspect(self.target(None, 6548))) + + assert row.progress is None + assert row.lifecycle is SessionLifecycle.LIVE diff --git a/tests/test_codex_shared_daemon.py b/tests/test_codex_shared_daemon.py new file mode 100644 index 0000000..198f91f --- /dev/null +++ b/tests/test_codex_shared_daemon.py @@ -0,0 +1,390 @@ +"""Joining the shared Codex app-server daemon, and admitting when it cannot be. + +#82 established that the shared daemon is the only route to a thread's own +truth, and #83 installed the login job that starts it. Nothing in this engine +ever dialled it: `CodexAgentAdapter._shared_daemon()` returned `None` +unconditionally, so every Codex row came from the process table and no thread +could be read at all. This is the join (advisor ruling on #76, Q1). + +Join-only, and the tests say so: nothing here spawns a daemon, stops one, or +outlives one. By the time this engine shuts down the user's `codex` TUIs are +thin clients of that daemon, and a product that stopped it would end their +Sessions (ADR 0012, #83's written rule). +""" + +from __future__ import annotations + +import asyncio +import json +import time +from pathlib import Path + +import pytest + +from gpt_voicecoding.adapters.agent.codex import shared_daemon +from gpt_voicecoding.adapters.agent.codex.shared_daemon import ( + DAEMON_VERSION_ARGUMENTS, + DaemonAddress, + SharedDaemon, + locate, +) +from gpt_voicecoding.adapters.codex_app_server.settings import CodexSettings + +SOCKET = Path("/tmp/app-server-control.sock") + + +def answering(document: object, *, status: int = 0) -> object: + """A `codex app-server daemon version` that answers exactly this.""" + + async def run(arguments: list[str]) -> tuple[int, str]: + assert arguments[1:] == list(DAEMON_VERSION_ARGUMENTS) + return status, document if isinstance(document, str) else json.dumps(document) + + return run + + +def running(**extra: object) -> dict[str, object]: + return { + "status": "running", + "socketPath": str(SOCKET), + "cliVersion": "0.149.1", + "appServerVersion": "0.149.1", + **extra, + } + + +class TestFindingIt: + """One lookup, ported from #82's prototype rather than invented again.""" + + def test_the_socket_path_comes_from_the_daemon_itself(self) -> None: + """Never derived from `CODEX_HOME`: the daemon is the one that knows.""" + address, reason = asyncio.run(locate("codex", run=answering(running()))) + + assert address == DaemonAddress( + socket_path=SOCKET, cli_version="0.149.1", app_server_version="0.149.1" + ) + assert reason == "" + + def test_a_daemon_that_is_not_running_is_a_reason_not_an_exception(self) -> None: + """The lane keeps working from the process table; it just says why (#74).""" + address, reason = asyncio.run( + locate("codex", run=answering("Error:\nno daemon is running", status=1)) + ) + + assert address is None + assert "no daemon is running" in reason + + def test_a_codex_that_is_not_there_is_the_same_kind_of_reason(self) -> None: + """A missing binary must not raise out of a five-second discovery tick.""" + + async def missing(arguments: list[str]) -> tuple[int, str]: + raise FileNotFoundError(arguments[0]) + + address, reason = asyncio.run(locate("codex", run=missing)) + + assert address is None + assert "codex" in reason + + def test_a_refusal_that_said_nothing_still_says_something(self) -> None: + """A reason nobody can read is worse than the honest admission of none.""" + address, reason = asyncio.run(locate("codex", run=answering("", status=1))) + + assert address is None + assert "gave no reason" in reason + + @pytest.mark.parametrize( + "document", + [ + "not json at all", + ["a", "list"], + {"status": "running"}, + {"status": "running", "socketPath": " "}, + ], + ids=["unparseable", "not-a-document", "no-socket-path", "blank-socket-path"], + ) + def test_an_answer_without_a_usable_socket_names_nothing(self, document: object) -> None: + """A shape this build cannot read is a reason, never a guessed path.""" + address, reason = asyncio.run(locate("codex", run=answering(document))) + + assert address is None + assert reason + + +class TestTheVersionMismatch: + """No pin (#67, Simon's ruling): a mismatch is reported, never refused.""" + + def test_a_mismatch_is_something_the_rows_carry_rather_than_lose(self) -> None: + address = DaemonAddress( + socket_path=SOCKET, cli_version="0.148.0", app_server_version="0.149.1" + ) + + assert "0.148.0" in address.note + assert "0.149.1" in address.note + + def test_matching_versions_have_nothing_to_say(self) -> None: + assert ( + DaemonAddress( + socket_path=SOCKET, cli_version="0.149.1", app_server_version="0.149.1" + ).note + == "" + ) + + def test_a_daemon_that_named_no_versions_is_not_read_as_agreeing(self) -> None: + """`None == None` would report a daemon that said nothing as one that agrees.""" + assert DaemonAddress(socket_path=SOCKET, cli_version="", app_server_version="").note + + +class TestTheConnection: + """Attached once, kept, and re-attached when the far side goes away.""" + + def daemon( + self, + attached: list[Path], + *, + fails: Exception | None = None, + slow: bool = False, + made: list[object] | None = None, + order: list[str] | None = None, + ) -> SharedDaemon: + class _Connection: + def __init__(self) -> None: + self.is_open = True + self.closed = 0 + + async def aclose(self) -> None: + self.is_open = False + self.closed += 1 + + async def attach(path: Path, **_: object) -> object: + attached.append(path) + if slow: + # A real dial does I/O, so it yields. A fake that returns + # without ever awaiting hands the loop back to nobody, and a + # race between two callers cannot happen in a test that never + # lets the second one start. + await asyncio.sleep(0) + await asyncio.sleep(0) + if fails is not None: + raise fails + if order is not None: + order.append("dial finished") + one = _Connection() + if made is not None: + made.append(one) + return one + + return SharedDaemon( + settings=CodexSettings(), + version="test", + locate=lambda executable: locate(executable, run=answering(running())), + attach=attach, + ) + + def test_one_connection_serves_every_tick(self) -> None: + """Dialling per tick would open a client per five seconds, forever.""" + attached: list[Path] = [] + daemon = self.daemon(attached) + + first = asyncio.run(daemon.client()) + second = asyncio.run(daemon.client()) + + assert first is second + assert attached == [SOCKET] + + def test_a_connection_that_went_away_is_dialled_again(self) -> None: + """The daemon can be restarted under us; the next tick must find it.""" + attached: list[Path] = [] + daemon = self.daemon(attached) + + first = asyncio.run(daemon.client()) + first.is_open = False # type: ignore[union-attr] + second = asyncio.run(daemon.client()) + + assert second is not first + assert attached == [SOCKET, SOCKET] + + def test_a_daemon_that_refuses_the_dial_leaves_the_lane_working(self) -> None: + """`None` plus a note: the rows come from the process table, as before.""" + attached: list[Path] = [] + daemon = self.daemon(attached, fails=OSError("connection refused")) + + assert asyncio.run(daemon.client()) is None + assert "connection refused" in daemon.note + + def test_a_daemon_that_cannot_be_found_says_so_without_dialling(self) -> None: + attached: list[Path] = [] + daemon = SharedDaemon( + settings=CodexSettings(), + version="test", + locate=lambda executable: locate( + executable, run=answering("Error:\nno daemon", status=1) + ), + attach=lambda path, **_: attached.append(path), # type: ignore[misc,return-value] + ) + + assert asyncio.run(daemon.client()) is None + assert attached == [] + assert "no daemon" in daemon.note + + def test_two_callers_at_once_leave_this_engine_holding_one_client(self) -> None: + """The cadence and a `progress` ask really do arrive together. + + `client()` read the connection, awaited the dial, and only then wrote + what it had made — so two callers that arrived while there was none both + saw none and both attached. The daemon then held two clients of an engine + that is supposed to be one of its clients, and the loser was dropped on + the floor with nothing left to close it. + """ + attached: list[Path] = [] + daemon = self.daemon(attached, slow=True) + + first, second = asyncio.run(_both(daemon)) + + assert first is second + assert attached == [SOCKET] + + def test_letting_go_while_a_dial_is_in_flight_leaves_nothing_attached(self) -> None: + """Shutdown invalidates the dial rather than waiting for it. + + Waiting would be the obvious answer and it is the wrong one: #96 derives + `runner.SHUTDOWN_SECONDS` from the phases it bounds, and a phase that + could sit out a ten-second daemon lookup is not one that sum has room + for. So `aclose` returns at once and the dial finds its generation stale + — and closes what it made, because clearing the field alone would leave + a client nothing holds and nothing closes. + """ + attached: list[Path] = [] + made: list[object] = [] + daemon = self.daemon(attached, slow=True, made=made) + + async def dial_and_let_go() -> object: + dialling = asyncio.ensure_future(daemon.client()) + await asyncio.sleep(0) + await daemon.aclose() + return await dialling + + answered = asyncio.run(dial_and_let_go()) + + assert answered is None + assert daemon._connection is None # noqa: SLF001 - the field is the leak + assert [one.closed for one in made] == [1] # type: ignore[attr-defined] + + def test_letting_go_does_not_wait_behind_a_dial(self) -> None: + """The budget, asserted rather than described (#96, `runner.py:95-117`). + + A `aclose` that took the dial's lock would finish only after the dial + did. Held here as an ordering fact, so a later change that reaches for + the obvious lock fails rather than quietly spending the shutdown budget. + """ + attached: list[Path] = [] + order: list[str] = [] + daemon = self.daemon(attached, slow=True, order=order) + + async def dial_and_let_go() -> None: + dialling = asyncio.ensure_future(daemon.client()) + await asyncio.sleep(0) + await daemon.aclose() + order.append("let go") + await dialling + + asyncio.run(dial_and_let_go()) + + assert order.index("let go") < order.index("dial finished") + + def test_letting_go_closes_this_engine_s_client_and_nothing_else(self) -> None: + """The daemon lives on: the user's TUIs are attached to it (#83's rule).""" + attached: list[Path] = [] + daemon = self.daemon(attached) + connection = asyncio.run(daemon.client()) + + asyncio.run(daemon.aclose()) + + assert connection.closed == 1 # type: ignore[union-attr] + assert asyncio.run(daemon.client()) is not connection + + +class TestItNeverOwnsTheDaemon: + """ADR 0012 and #83: this product starts no daemon and stops none it joined. + + Asserted on what the module *runs*, not on what its source says: a test that + scanned the text would trip over the paragraph explaining the rule, and + would still pass for a module that spelled the forbidden verb differently. + """ + + def test_the_only_daemon_subcommand_asked_for_is_the_one_that_reports(self) -> None: + """`start` and `bootstrap` are lifecycle; `version` only tells.""" + assert DAEMON_VERSION_ARGUMENTS[-1] == "version" + + def test_nothing_else_is_ever_run(self) -> None: + """One command, exactly as spelled — and it is the whole of what runs.""" + ran: list[list[str]] = [] + + async def watched(arguments: list[str]) -> tuple[int, str]: + ran.append(arguments) + return 0, json.dumps(running()) + + asyncio.run(locate("codex", run=watched)) + + assert ran == [["codex", *DAEMON_VERSION_ARGUMENTS]] + + +async def _both(daemon: SharedDaemon) -> tuple[object, object]: + """Two asks for the client, started before either has answered.""" + first, second = await asyncio.gather(daemon.client(), daemon.client()) + return first, second + + +class TestACodexThatNeverAnswers: + """The bound that keeps one hung subprocess off the five-second cadence. + + `core/bridge.py:505` awaits `discover()` with no deadline of its own, so a + `communicate()` that never returns stops the roster for **both** lanes for as + long as that process lives — the Claude lane included, because the cadence is + one loop over the adapters. The lane's other subprocess reader has carried + this bound since it was written (`codex/processes.py:175-181`), and so does + the Claude lane's (`claude/discovery.py:94-100`); this is the same risk read + off the same machine. + """ + + def test_a_lookup_that_timed_out_is_a_reason_rather_than_an_exception(self) -> None: + """`locate` never raises, whatever happened: a lane that threw takes the roster.""" + + async def never(arguments: list[str]) -> tuple[int, str]: + raise TimeoutError + + address, reason = asyncio.run(locate("codex", run=never)) + + assert address is None + assert "did not answer" in reason + + def test_a_timeout_is_not_reported_as_a_binary_that_could_not_be_run(self) -> None: + """`TimeoutError` is an `OSError`, so the order of the two handlers is the behaviour. + + Caught after `OSError`, a daemon that hung would be reported to the user + as a `codex` that is not installed — the one reason that sends somebody + looking in the wrong place. + """ + + async def never(arguments: list[str]) -> tuple[int, str]: + raise TimeoutError + + _, reason = asyncio.run(locate("codex", run=never)) + + assert "could not be run" not in reason + + def test_the_command_is_given_up_on_and_the_process_killed( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Bounded in the default runner, so nothing that injects one has to remember to. + + Run against a real command that does not answer, because the bound is a + property of how the subprocess is awaited and a fake runner cannot hold + it. The wait is the patched one, not the shipped ten seconds. + """ + monkeypatch.setattr(shared_daemon, "COMMAND_TIMEOUT_SECONDS", 0.05) + started = time.monotonic() + + with pytest.raises(TimeoutError): + asyncio.run(shared_daemon._run(["/bin/sleep", "30"])) # noqa: SLF001 + + assert time.monotonic() - started < 5.0 diff --git a/tests/test_codex_thread_tail.py b/tests/test_codex_thread_tail.py new file mode 100644 index 0000000..244e58c --- /dev/null +++ b/tests/test_codex_thread_tail.py @@ -0,0 +1,150 @@ +"""How far along a Codex thread is, read out of the daemon's own answer (#76). + +Every shape here was **measured**, not assumed: `thread/read` was called against +the real shared daemon on codex 0.149.1 on 2026-08-26 and its answer copied down. +That matters twice. The item types below (`reasoning`, `commandExecution` beside +`agentMessage` and `userMessage`) are what a real turn actually holds, and the +times are integers — `updatedAt: 1787712279` for a thread last touched at +2026-08-26T02:44:39Z — where the reference implementation's own reader never read +one at all. + +The bound itself is `test_progress_bound.py`'s, because both lanes share it; +this file is what a Codex turn contributes to it. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from gpt_voicecoding.adapters.agent._progress import RECENT_LIMIT +from gpt_voicecoding.adapters.agent.codex.thread_tail import last_activity, moment, recent +from gpt_voicecoding.seams.agent import ProgressRole + +#: The moment the live probe read, and the integer the daemon spelled it as. +MEASURED_SECONDS = 1787712279 +MEASURED = datetime(2026, 8, 26, 2, 44, 39, tzinfo=UTC) + + +def spoke(text: str) -> dict[str, Any]: + """An `agentMessage`, as the daemon writes it.""" + return {"type": "agentMessage", "id": "msg_0e4f", "text": text} + + +def told(text: str) -> dict[str, Any]: + """A `userMessage`, whose words live one level down in `content`.""" + return { + "type": "userMessage", + "id": "01a03bf3", + "clientId": None, + "content": [{"type": "text", "text": text}], + } + + +def turn(*items: dict[str, Any], status: str = "completed") -> dict[str, Any]: + return {"id": "01a03bf0", "status": status, "items": list(items)} + + +def thread(*turns: dict[str, Any], **extra: Any) -> dict[str, Any]: + return {"id": "01a03be8", "status": {"type": "idle"}, "turns": list(turns), **extra} + + +def texts(document: dict[str, Any]) -> list[str]: + entries, _ = recent(document) + return [entry.text for entry in entries] + + +class TestWhatCountsAsProgress: + """Ported selection: what was said, never the machinery of doing the work.""" + + def test_both_sides_are_carried_and_each_says_which_it_is(self) -> None: + entries, truncated = recent(thread(turn(told("do the thing"), spoke("done")))) + + assert [(entry.role, entry.text) for entry in entries] == [ + (ProgressRole.USER, "do the thing"), + (ProgressRole.ASSISTANT, "done"), + ] + assert truncated is False + + def test_the_machinery_of_doing_the_work_is_not_a_report_of_it(self) -> None: + """`reasoning` and `commandExecution` are the two a real turn is full of.""" + document = thread( + turn( + told("look at the diff"), + {"type": "reasoning", "id": "rs_0e4f", "summary": [], "content": []}, + { + "type": "commandExecution", + "id": "exec-33579ceb", + "command": '/bin/zsh -lc "git show"', + }, + spoke("here is what I found"), + ) + ) + + assert texts(document) == ["look at the diff", "here is what I found"] + + def test_an_item_type_this_build_has_never_seen_costs_itself(self) -> None: + """**Adapted** from legacy, which raised: a roster row must not blank.""" + document = thread(turn(spoke("done"), {"type": "somethingNewIn0150", "id": "x"})) + + assert texts(document) == ["done"] + + def test_a_user_message_that_is_only_an_image_says_nothing(self) -> None: + """`image` and `localImage` are the other two content shapes; neither speaks.""" + picture = told("") + picture["content"] = [{"type": "image", "imageUrl": "data:..."}] + + assert texts(thread(turn(picture, spoke("I see it")))) == ["I see it"] + + def test_turns_are_read_in_the_order_they_happened(self) -> None: + document = thread(turn(spoke("first")), turn(spoke("second"))) + + assert texts(document) == ["first", "second"] + + def test_a_turn_still_running_contributes_what_it_has(self) -> None: + """`turn_status` is dropped, so an in-progress turn is not a special case.""" + assert texts(thread(turn(spoke("halfway"), status="inProgress"))) == ["halfway"] + + def test_the_bound_is_the_shared_one(self) -> None: + """One bound, one type, whichever lane the row came from.""" + document = thread(turn(*(spoke(f"step {index}") for index in range(RECENT_LIMIT + 2)))) + + entries, truncated = recent(document) + + assert len(entries) == RECENT_LIMIT + assert truncated is True + + +class TestAThreadWithNoTurns: + """The cheap read the roster takes every tick answers no turn list at all.""" + + def test_a_document_read_without_turns_is_not_an_error(self) -> None: + document = thread() + del document["turns"] + + assert recent(document) == ((), False) + + def test_a_thread_that_has_taken_no_turn_says_nothing_yet(self) -> None: + assert recent(thread()) == ((), False) + + +class TestLastActivity: + """Epoch seconds, and only what the thread itself said.""" + + def test_updated_at_is_when_the_thread_last_moved(self) -> None: + assert last_activity(thread(updatedAt=MEASURED_SECONDS)) == MEASURED + + def test_a_thread_that_named_no_time_has_none(self) -> None: + assert last_activity(thread()) is None + + def test_a_boolean_is_not_a_time(self) -> None: + """`True` is an `int` in Python, and one second past the epoch is nonsense.""" + assert moment(True) is None + + def test_a_shape_this_build_cannot_read_is_no_time_at_all(self) -> None: + assert moment("2026-08-26T02:44:39Z") is None + assert moment(None) is None + + def test_a_number_no_calendar_can_hold_is_no_time_either(self) -> None: + """A field that moved to milliseconds would arrive as one of these.""" + assert moment(10**30) is None diff --git a/tests/test_control_plane_actions.py b/tests/test_control_plane_actions.py index aefb8fd..ad94b14 100644 --- a/tests/test_control_plane_actions.py +++ b/tests/test_control_plane_actions.py @@ -14,12 +14,16 @@ from __future__ import annotations import asyncio +import json +from datetime import UTC, datetime from pathlib import Path import pytest from fakes import FakeAgent, FakeCall, FakeCompanionChannel, instruction_context +from gpt_voicecoding.adapters.agent._progress import encoded_size from gpt_voicecoding.control_plane.actions import ControlPlane +from gpt_voicecoding.control_plane.payloads import progress_document from gpt_voicecoding.core.bridge import BridgeCore from gpt_voicecoding.core.relay_queue import RelayQueue from gpt_voicecoding.core.sessions import Session, SessionRegistry @@ -29,8 +33,17 @@ from gpt_voicecoding.seams.agent import ( ApprovalRequest, AwaitingApproval, + ChildClassification, + ChildKind, + LaneDiscovery, + LaneUnavailable, + Progress, + ProgressEntry, + ProgressRole, ReplyWindow, ReplyWindowChanged, + SessionInspection, + SessionState, ) from gpt_voicecoding.seams.control_plane import Action, ErrorCode, Reply, Request from gpt_voicecoding.seams.identity import AgentKind, SessionLabel, SessionTarget @@ -43,6 +56,9 @@ SECOND_CODEX = SessionTarget(agent=AgentKind.CODEX, session_id="def") CODEX_ADDRESS = {"agent": "codex", "session_id": "abc", "pid": None} +#: One fixed moment, so a rendered reading is compared against a value and not a clock. +READ_AT = datetime(2026, 8, 26, 2, 44, 39, tzinfo=UTC) + class Surface: """One assembled engine-side control plane, and the knobs a test needs.""" @@ -94,6 +110,19 @@ class TestWithEverySwitchOff: def test_every_action_succeeds_with_duty_voice_and_message_off(self) -> None: surface = Surface(duty=False) surface.register() + # `progress` needs something to have been read, or it refuses for a real + # reason — which is not this test's subject. The subject is that no + # *switch* refuses anything, so the lane is given a reading to answer. + surface.agent.discovery = LaneDiscovery( + rows=( + SessionInspection( + target=CODEX, + workspace=WORKSPACE, + state=SessionState.IDLE, + progress=Progress(read_at=READ_AT), + ), + ) + ) asyncio.run( surface.core.approvals.opened( ApprovalRequest(approval_id="a1", target=CODEX, tool_name="Bash") @@ -104,6 +133,7 @@ def test_every_action_succeeds_with_duty_voice_and_message_off(self) -> None: Action.STATUS: surface.ask(Action.STATUS), Action.SWITCH: surface.ask(Action.SWITCH, name="duty", on=False), Action.SESSIONS: surface.ask(Action.SESSIONS), + Action.PROGRESS: surface.ask(Action.PROGRESS, target=CODEX_ADDRESS), Action.LIVE: surface.ask(Action.LIVE), Action.RELAY: surface.ask(Action.RELAY, target=CODEX_ADDRESS, text="carry on"), Action.APPROVE: surface.ask(Action.APPROVE, approval_id="a1", verdict="allow"), @@ -283,3 +313,202 @@ def test_the_command_words_are_exactly_the_action_set(self) -> None: def test_every_action_is_dispatchable(action: Action) -> None: """A closed action set with a handler missing is a wire that lies.""" assert action in ControlPlane(Surface().core).handlers + + +class TestProgress: + """#76's verb: one exact Session, read now, and never a turn.""" + + def stopped(self, *, said: str = "done") -> LaneDiscovery: + return LaneDiscovery( + rows=( + SessionInspection( + target=CODEX, + workspace=WORKSPACE, + state=SessionState.IDLE, + progress=Progress( + recent=(ProgressEntry(role=ProgressRole.ASSISTANT, text=said),), + truncated=True, + read_at=READ_AT, + ), + last_activity=READ_AT, + ), + ) + ) + + def test_one_session_is_read_now_and_rendered_as_a_roster_row(self) -> None: + """The same document `sessions` renders: a surface learns no second shape.""" + surface = Surface() + surface.register() + surface.agent.discovery = self.stopped() + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.ok + session = reply.data["session"] + assert session["target"] == CODEX_ADDRESS + assert session["progress"] == { + "recent": [{"role": "assistant", "text": "done"}], + "truncated": True, + "read_at": READ_AT.isoformat(), + } + assert session["last_activity"] == READ_AT.isoformat() + + def test_it_asks_the_lane_rather_than_answering_from_the_roster(self) -> None: + """A cached answer would make the verb no fresher than the tick beside it.""" + surface = Surface() + surface.register() + surface.agent.discovery = self.stopped() + + surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert surface.agent.inspections == [CODEX] + + def test_the_reading_becomes_the_rosters_truth(self) -> None: + """Asking for progress then asking for status cannot say two things.""" + surface = Surface() + surface.register() + surface.agent.discovery = self.stopped(said="halfway") + + surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + roster = surface.ask(Action.SESSIONS).data["sessions"] + + assert roster[0]["progress"]["recent"] == [{"role": "assistant", "text": "halfway"}] + + def test_it_ends_nothing_it_did_not_look_at(self) -> None: + """A verb asked about one Session concludes nothing about the others.""" + surface = Surface() + surface.register() + surface.register(SECOND_CODEX) + surface.agent.discovery = self.stopped() + + surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert len(surface.ask(Action.SESSIONS).data["sessions"]) == 2 + + def test_a_session_nobody_registered_is_refused_by_identity(self) -> None: + surface = Surface() + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.error is not None + assert reply.error.code is ErrorCode.UNKNOWN_SESSION + + def test_a_lane_that_could_not_look_refuses_rather_than_saying_nothing_happened( + self, + ) -> None: + """ "I could not look" and "it has said nothing" are different facts.""" + surface = Surface() + surface.register() + surface.agent.inspect_raises = LaneUnavailable(AgentKind.CODEX, "`codex` is not on PATH") + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.error is not None + assert reply.error.code is ErrorCode.REFUSED + assert "`codex` is not on PATH" in reply.error.message + + def test_a_lane_that_could_not_look_leaves_the_row_as_it_was(self) -> None: + surface = Surface() + surface.register() + surface.agent.inspect_raises = LaneUnavailable(AgentKind.CODEX, "`codex` is not on PATH") + + surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert surface.ask(Action.SESSIONS).data["sessions"][0]["lifecycle"] == "live" + + def test_a_session_with_no_readable_progress_refuses_rather_than_says_nothing( + self, + ) -> None: + """An unattached Codex row, or one whose first turn wrote no record (#73). + + The refusal is #76's "honest error for an unattached or ended row": a + surface handed a successful reply carrying no progress would render a + Session nobody could read as one that has said nothing. + """ + surface = Surface() + surface.register() + surface.agent.discovery = LaneDiscovery( + rows=(SessionInspection(target=CODEX, workspace=WORKSPACE, state=SessionState.IDLE),) + ) + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.error is not None + assert reply.error.code is ErrorCode.REFUSED + assert "never infers one" in reply.error.message + + def test_a_session_read_and_found_silent_answers_rather_than_refuses(self) -> None: + """The other side of that line, and the whole reason it is drawn.""" + surface = Surface() + surface.register() + surface.agent.discovery = LaneDiscovery( + rows=( + SessionInspection( + target=CODEX, + workspace=WORKSPACE, + state=SessionState.IDLE, + progress=Progress(read_at=READ_AT), + ), + ) + ) + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.ok + assert reply.data["session"]["progress"]["recent"] == [] + + def test_a_session_that_has_ended_is_a_stale_target_not_an_empty_answer(self) -> None: + """#76's other honest error.""" + surface = Surface() + surface.register() + surface.agent.discovery = LaneDiscovery() # a lane that looked and found nothing + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.error is not None + assert reply.error.code is ErrorCode.STALE_SESSION + + def test_it_does_not_end_the_row_itself(self) -> None: + """Ending a row is `observe`'s, and only `observe`'s. + + The value an `inspect` answers for a Session it could not find carries no + workspace and no name, so folding it into the roster would strip the very + fields a surface needs to say what happened to it. The next discovery + ends it properly, within one cadence. + """ + surface = Surface() + held = surface.register() + surface.agent.discovery = LaneDiscovery() + + surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert surface.state.sessions.all() == (held,) + + def test_a_child_process_is_refused_before_any_lane_is_touched(self) -> None: + """Seen, never spoken to — and never asked on its own behalf either (#68).""" + surface = Surface() + surface.state.sessions.register( + Session( + target=CODEX, + workspace=WORKSPACE, + first_seen=0.0, + child=ChildClassification(kind=ChildKind.CHILD, parent=SECOND_CODEX), + ) + ) + + reply = surface.ask(Action.PROGRESS, target=CODEX_ADDRESS) + + assert reply.error is not None + assert surface.agent.inspections == [] + + def test_the_budget_is_measured_on_the_document_that_travels(self) -> None: + """`_progress.encoded_size` and `progress_document` are one shape (#47's lesson).""" + entries = ( + ProgressEntry(role=ProgressRole.USER, text="do the thing"), + ProgressEntry(role=ProgressRole.ASSISTANT, text="done"), + ) + rendered = progress_document(Progress(recent=entries)) + + assert encoded_size(entries) == len( + json.dumps(rendered["recent"], ensure_ascii=False).encode("utf-8") + ) diff --git a/tests/test_control_plane_vocabulary.py b/tests/test_control_plane_vocabulary.py index dab48b8..c2847d8 100644 --- a/tests/test_control_plane_vocabulary.py +++ b/tests/test_control_plane_vocabulary.py @@ -29,12 +29,24 @@ def test_the_action_set_is_exactly_these(self) -> None: "status", "switch", "sessions", + "progress", "live", "relay", "approve", "verify", } + def test_adding_an_action_did_not_move_the_protocol_version(self) -> None: + """`progress` is additive, and the version guards the other direction (#76). + + A version exists so a surface can tell an engine that disagrees from one + too old to have been asked, and what it guards against is a surface + sending something the engine does not have. A surface built against 4 + never sends `progress`; one that does is talking to an engine that has + it. Removing an action can strand a surface, which is what 4 was for. + """ + assert PROTOCOL_VERSION == 4 + def test_launching_and_closing_are_not_actions_this_engine_has(self) -> None: """Parked with the launcher (#72), and their absence is asserted, not assumed. diff --git a/tests/test_progress_bound.py b/tests/test_progress_bound.py new file mode 100644 index 0000000..b55c5ec --- /dev/null +++ b/tests/test_progress_bound.py @@ -0,0 +1,107 @@ +"""How much of what a Session said travels — one bound, both lanes (#76). + +The rule is ported whole from `legacy@1d32845:bridge/transcript.py:2828-2860`; +the numbers are not, and the reason is the shape of the thing that carries them. +Legacy bounded a per-Session verb's answer (12 entries / 32 KB, +`config.plist:449-452`); this bounds **every row** of a roster reply with a 64 KB +ceiling on it. Each lane's own reader is tested beside it +(`test_claude_transcript_tail.py`, `test_codex_thread_tail.py`); this is the +bound they share, tested where it lives so neither lane can quietly grow one of +its own. +""" + +from __future__ import annotations + +import pytest + +from gpt_voicecoding.adapters.agent._progress import ( + RECENT_LIMIT, + RECENT_MAX_BYTES, + bounded, + encoded_size, +) +from gpt_voicecoding.seams.agent import ProgressEntry, ProgressRole + + +def said(text: str, *, role: ProgressRole = ProgressRole.ASSISTANT) -> ProgressEntry: + return ProgressEntry(role=role, text=text) + + +class TestKeepingTheNewest: + def test_nothing_said_is_nothing_dropped(self) -> None: + assert bounded([]) == ((), False) + + def test_a_tail_that_fits_is_carried_whole(self) -> None: + entries = [said("one"), said("two")] + + assert bounded(entries) == (tuple(entries), False) + + def test_the_oldest_go_first_and_the_drop_is_admitted(self) -> None: + entries = [said(f"step {index}") for index in range(RECENT_LIMIT + 2)] + + kept, truncated = bounded(entries) + + assert [entry.text for entry in kept] == [ + f"step {index}" for index in range(2, RECENT_LIMIT + 2) + ] + assert truncated is True + + +class TestTheByteBudget: + def test_a_tail_too_large_is_widened_down_until_it_fits(self) -> None: + """Built from the budget rather than from a guess at it. + + `overhead` is what one entry costs besides its own text, so `big` fills + the budget exactly on its own and cannot share it with anything. + """ + overhead = encoded_size([said("x")]) - len("x") + big = said("x" * (RECENT_MAX_BYTES - overhead)) + newest = said("the last word") + assert encoded_size([big]) == RECENT_MAX_BYTES + + kept, truncated = bounded([big, newest]) + + assert kept == (newest,) + assert truncated is True + + def test_a_single_oversize_entry_leaves_the_row_saying_nothing(self) -> None: + """Legacy's `drop_oversize_tail`, and here it is the only behaviour. + + A reply nobody can read is a worse failure than a row that admits it is + empty: the byte ceiling is on the *whole* roster, so one Session's very + long paragraph must not take every other Session's row down with it. + """ + assert bounded([said("y" * (RECENT_MAX_BYTES * 2))]) == ((), True) + + def test_what_is_kept_is_never_cut(self) -> None: + """`stop_analysis.SUMMARY_MAX_CHARS`'s rule: whole, or not at all.""" + entries = [said("a" * 40), said("b" * 40)] + + kept, _ = bounded(entries) + + assert [len(entry.text) for entry in kept] == [40, 40] + + def test_the_role_counts_because_the_role_travels(self) -> None: + """Measured on the document `progress_document` renders, not on the text.""" + entry = said("hello", role=ProgressRole.USER) + + assert encoded_size([entry]) > len(entry.text) + assert encoded_size([]) == len(b"[]") + + +class TestABoundThatBoundsNothing: + @pytest.mark.parametrize(("limit", "max_bytes"), [(0, 1), (-1, 1), (1, 0), (1, -1)]) + def test_is_refused_rather_than_read_as_no_bound_at_all( + self, limit: int, max_bytes: int + ) -> None: + """Legacy validated its limits instead of silently reading everything.""" + with pytest.raises(ValueError): + bounded([said("anything")], limit, max_bytes=max_bytes) + + +class TestTheNumbersFitTheReplyTheyRideOn: + def test_ten_sessions_at_the_budget_stay_inside_the_protocol_ceiling(self) -> None: + """Why 3 KB and not 12 entries: `status` carries one of these per row.""" + from gpt_voicecoding.seams.control_plane import MAX_REQUEST_BYTES + + assert RECENT_MAX_BYTES * 10 < MAX_REQUEST_BYTES