From 153222aa000048f4883c5b0c994c1eda70931205 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 07:48:25 -0400 Subject: [PATCH 01/10] Give DM peers one typed encoding and repin the native build Three tests and every fresh native build were red on main. - DM history kept its peer as `str | ThreadTarget` (historical_views.py DMHistory.second, read_basis direct_history and fetch_page, history_views dm_display_page) and re-decoded it with CommandTarget.named, which #2733 (f58e2e39c) correctly narrowed to raw strings. Any typed peer, including Toad's retained IncarnationThreadTarget, crashed on removeprefix. The peer is now a ThreadTarget from the boundary in; the re-decode and the redundant require_dm_peer calls are gone, and ThreadViewRead decodes its thread name once. - test_historical_views read bus_files() keys as names; since 1a6fc31c1 (#2709) the keys are typed file identities and the paths are the member maps. The test reads members through one helper. - test_outbound_transcript_source hand-built RecordedThreadIncarnation from Thread.created_at, which d5cd4bbcb (#2724) made a CreationTime. The test now asks the owner, Thread.incarnation / recipient.require_recorded(). - f4ca048a3 (#2748) changed stack/native-turn-context.{mjs,d.ts} without --update-pins, so prepare-pi-native refused every fresh build (and the next install). Repinned: only turn-context.js and turn-context.d.ts move. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/historical_views.py | 10 ++++------ src/agent_comms/history_views.py | 4 ++-- src/agent_comms/read_basis.py | 14 ++++++-------- stack/pi-native.sha256 | 8 ++++---- tests/test_channel_display.py | 4 ++-- tests/test_dm_scoped_ack.py | 16 ++++++++-------- tests/test_dm_scoped_ack_successor.py | 18 +++++++++--------- tests/test_historical_views.py | 18 ++++++++++++------ tests/test_ordinary_nk_delivery.py | 3 ++- tests/test_outbound_transcript_source.py | 5 ++--- tests/test_read_ledger.py | 15 ++++++++------- tests/test_thread_identity.py | 5 +++-- tests/test_user_thread_lookup.py | 4 ++-- 13 files changed, 64 insertions(+), 60 deletions(-) diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py index 2690c5b1d7..5fe859c7bd 100644 --- a/src/agent_comms/historical_views.py +++ b/src/agent_comms/historical_views.py @@ -13,7 +13,7 @@ from typing import TYPE_CHECKING from .bus import Bus -from .channel_targets import CommandTarget, NamedTarget, ThreadTarget +from .channel_targets import NamedTarget, ThreadTarget from .channels import Channel from .coordination_errors import HistoricalSourceChanged from .errors import UnregisteredThreadError @@ -194,11 +194,11 @@ def capture(self, snapshot: RegistryProvenance) -> ChannelDisplayScope: @dataclass(frozen=True, slots=True) class DMHistory(HistoryView): first: str - second: str | ThreadTarget + second: ThreadTarget def capture(self, snapshot: RegistryProvenance) -> DMDisplayScope: return DMDisplayScope.capture( - self.first, CommandTarget.named(self.second).selected_name(snapshot), snapshot) + self.first, self.second.selected_name(snapshot), snapshot) @dataclass(frozen=True, slots=True) @@ -231,10 +231,8 @@ def live_page(self, presentation: BusPresentation, bounds: PageBounds) -> Messag from .read_basis import DMDisplayBasis bus = presentation.bus - target = CommandTarget.named(self.second) - target.require_dm_peer() with presentation.registry.store.reading_snapshot() as snapshot: - peer = target.selected_name(snapshot) + peer = self.second.selected_name(snapshot) missing = peer not in snapshot.threads if missing and bus.history.threads(peer): return MessagePage((), False, False, NoDisplaySelection()) diff --git a/src/agent_comms/history_views.py b/src/agent_comms/history_views.py index 3cb4562f63..7193e3352b 100644 --- a/src/agent_comms/history_views.py +++ b/src/agent_comms/history_views.py @@ -119,7 +119,7 @@ def external(self): self.through.wire.require_source(AssignedSourceIdentity(str(self.views.root), self.thread.incarnation)) page = DMDisplayBasis.fetch_page( self.views.bus, self.views.registry, self.views.root, self.viewer.name, - self.thread.name, worktree=self.viewer.worktree, + ThreadTarget(self.thread.name), worktree=self.viewer.worktree, bounds=PageBounds(LiveCursor(self.through.wire.sequence + 1).older()), ) if page.messages: @@ -216,7 +216,7 @@ def historical_threads(self, name: str | None = None) -> tuple[HistoricalThread, return self.bus.history.threads(name) def dm_display_page( - self, peer: str | ThreadTarget, *, worktree: str, bounds: PageBounds = NEWEST_PAGE + self, peer: ThreadTarget, *, worktree: str, bounds: PageBounds = NEWEST_PAGE ) -> MessagePage: viewer = self.messaging.user_identity(worktree).name return DMDisplayHistory(viewer, peer, worktree).page(self.presentation, bounds) diff --git a/src/agent_comms/read_basis.py b/src/agent_comms/read_basis.py index cded4de71e..f631db74b9 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -37,7 +37,7 @@ class DisplayViewer(SubclassUnion, affix="Viewer"): """The reader owns read evidence, aliases and cursor selection.""" @abstractmethod - def direct_history(self, direct: type[DMHistory], peer: str | ThreadTarget) -> DMHistory: ... + def direct_history(self, direct: type[DMHistory], peer: ThreadTarget) -> DMHistory: ... @abstractmethod def resolve(self, snapshot: RegistryProvenance) -> DisplayViewer: ... @@ -69,7 +69,7 @@ class AnonymousViewer(NullMember, DisplayViewer): def resolve(self, snapshot: RegistryProvenance) -> AnonymousViewer: return self - def direct_history(self, direct: type[DMHistory], peer: str | ThreadTarget) -> DMHistory: + def direct_history(self, direct: type[DMHistory], peer: ThreadTarget) -> DMHistory: raise ValueError("DM history requires --as .") def historical_selection(self, page: MessagePage, presentation: BusPresentation, historical: HistorySource, source: Bus, document: ReadDocument, snapshot: RegistrySnapshot) -> NoDisplaySelection: @@ -101,7 +101,7 @@ class NamedViewer(SingleFieldValue, DisplayViewer): def resolve(self, snapshot: RegistryProvenance) -> NamedViewer: return type(self)(snapshot.canonical_name(self.name)) - def direct_history(self, direct: type[DMHistory], peer: str | ThreadTarget) -> DMHistory: + def direct_history(self, direct: type[DMHistory], peer: ThreadTarget) -> DMHistory: return direct(self.name, peer) def historical_selection(self, page: MessagePage, presentation: BusPresentation, historical: HistorySource, source: Bus, document: ReadDocument, snapshot: RegistrySnapshot) -> HistoricalSelection: @@ -679,7 +679,7 @@ def fetch_page( registry: Registration, root: Path, viewer: str, - peer: str | ThreadTarget, + peer: ThreadTarget, *, worktree: str, bounds: PageBounds, @@ -689,11 +689,9 @@ def fetch_page( Fetching is not paint proof. A UI may use the basis only after proving that the corresponding inbound tail was contiguous and visibly painted. """ - target = CommandTarget.named(peer) - target.require_dm_peer() with registry.store.reading_snapshot() as snapshot: viewer_name = snapshot.canonical_name(viewer) - peer_name = target.selected_name(snapshot) + peer_name = peer.selected_name(snapshot) scope = DMDisplayScope.capture_human(viewer_name, peer_name, snapshot) viewer_thread = snapshot.threads[viewer_name] peer_thread = snapshot.participant(peer_name) @@ -708,7 +706,7 @@ def fetch_page( root=str(root.resolve()), root_identity=(root_info.st_dev, root_info.st_ino), worktree=str(Path(worktree).resolve()), - requested_peer=target.name, + requested_peer=peer.name, viewer_identity=viewer_thread.incarnation, viewer_names=viewer_names, peer_identity=peer_thread.incarnation, diff --git a/stack/pi-native.sha256 b/stack/pi-native.sha256 index f18799e6db..b13826a331 100644 --- a/stack/pi-native.sha256 +++ b/stack/pi-native.sha256 @@ -40,8 +40,8 @@ e747c952fa4d19e9dce98f54a6f3ae0ceafcaf07facb4adba733320e74f81dc3 node_modules/@ 5d41b9551078157d74b4104c37fb6040533575b4ce6a743528b681390fc36520 node_modules/@earendil-works/pi-ai/dist/api/pi-messages.js 4e0184d73736b60c2e32acc12a4b4eebad1324d7f61a4c955e896b3a1239db8d node_modules/@earendil-works/pi-ai/dist/api/simple-options.js 36d8bd607a5219f0b10d4e7d0e434f77333407da94c5ce94fde3cb3c8f608d75 node_modules/@earendil-works/pi-ai/dist/utils/estimate.js -ca056ed491fefa4739e193ba0d3551687c1a427b16379433194f4e6e017487ba dist/core/turn-context.js -acf1393262881035c793bb9f8defad011c15a2a2c249715b0cadd64cd86f9522 dist/core/turn-context.d.ts +b56a60c854b411dd0680842a540bb3ec0a0e906b630964e4bbaaf7e33efde9b3 dist/core/turn-context.js +e86392ace03df89e8c861206bc2ad83f3353e70c339dc066860d7548fd46efe1 dist/core/turn-context.d.ts 176621c0f4bcfb7f3d53bca5eaf9c5b7c57d829c0e1529844e8e3ee2100d5ab9 dist/core/tools/agent-comms-file-artifact.js b3ad505e285cbd5aa349011e03111212bd88d1a89a0e31b117272e4108548f18 dist/core/tools/edit.js 233ab990ddbb3f54f19baa02ab11620da01588a664fd8da639ffe5fa268f3f0f dist/core/tools/native-file-artifact-schema.mjs @@ -56,5 +56,5 @@ d64e8c05c7932d377034820b1db0df45612afe64885c0765f8f348a12e1bc955 dist/core/skil 8e8276bbabcd898136d431345611928626fad6b6f03ac572f1d8506e6ece326f dist/core/system-prompt.d.ts 0391543cf703a93a0163b37b3ccebe53e0f12bff6f6162b86d56fb017fe49011 dist/core/system-prompt.js # agent-comms-native-stock-v1 @earendil-works/pi-coding-agent@0.85.1 -# agent-comms-native-tree-v1 e0c899f88db25438981bb373aaa4861ea9aa4091a16a2bd8c8d61803a1c2569f -# agent-comms-native-inputs-v1 7d410d9b0f468e42d554171afe589c96301bf3f2dc13e050277437c97e8bdfd0 +# agent-comms-native-tree-v1 17edf8badd73bafccf592b651b5a9c0f71b2b05b89db470d1042d3ccb544a0b9 +# agent-comms-native-inputs-v1 a420ec4c8414f1e9b5355eeaf303c1c1f798d47c876d5612bf2512daf445294b diff --git a/tests/test_channel_display.py b/tests/test_channel_display.py index 8ed1d25f24..dfbfe103e5 100644 --- a/tests/test_channel_display.py +++ b/tests/test_channel_display.py @@ -7,7 +7,7 @@ import pytest -from agent_comms.channel_targets import CommandTarget +from agent_comms.channel_targets import CommandTarget, ThreadTarget from agent_comms.channels import AnyOfMatch, ViewPredicate, ActivityView from agent_comms.comms import wire from agent_comms.coordination_errors import DMPaintedBoundChanged, DMScopeChanged @@ -213,7 +213,7 @@ def burst(): page = comms.views.channel_display_page("#api", viewer=NamedViewer(comms.messaging.user_identity(str(tmp_path)).name)) if page.newest_seq is not None: comms.views.mark_page_read(page, worktree=str(tmp_path)) - dm = comms.views.dm_display_page("alice", worktree=str(tmp_path)) + dm = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path)) assert bodies(dm) == ["DM original"] comms.views.mark_page_read(dm, worktree=str(tmp_path)) finally: diff --git a/tests/test_dm_scoped_ack.py b/tests/test_dm_scoped_ack.py index 86d05e0cff..9cc5bf2a4c 100644 --- a/tests/test_dm_scoped_ack.py +++ b/tests/test_dm_scoped_ack.py @@ -1,7 +1,7 @@ """No-provider bounded human DM read evidence; never use global inbox ACK.""" from __future__ import annotations -from agent_comms.channel_targets import CommandTarget +from agent_comms.channel_targets import CommandTarget, ThreadTarget from agent_comms.registration import Registration import os @@ -29,7 +29,7 @@ def test_dm_page_marks_covered_range_preserving_omitted_older_unread(tmp_path: P viewer = comms.messaging.user_identity(str(tmp_path)).name for index in range(50): comms.messaging.send("peer", viewer, f"old {index}").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path), bounds=PageBounds(limit=5)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path), bounds=PageBounds(limit=5)) assert len(page.messages) == 5 and page.has_older assert isinstance(page.selection, DMSelection) comms.views.mark_page_read(page, worktree=str(tmp_path)) @@ -50,7 +50,7 @@ def test_peer_delete_same_name_rebind_rejects_stale_page_without_read_ack(tmp_pa comms.messaging.send("other", "#all", "OTHER_CHANNEL_4").message.message_id comms.messaging.send("peer", viewer, "PEER_PAINTED_5").message.message_id comms.messaging.send("other", viewer, "OTHER_UNPAINTED_6").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert [message.body for message in page.messages] == ["PEER_PAINTED_2", "PEER_PAINTED_5"] assert page.newest_seq == 5 and isinstance(page.selection, DMSelection) comms.registry.unregister("peer") @@ -74,7 +74,7 @@ def test_foreign_root_wrong_peer_and_unbounded_or_bool_through_rejected(tmp_path comms.registry.declare(_thread(comms.root, name)) viewer = comms.messaging.user_identity(str(comms.root)).name comms.messaging.send("peer", viewer, "painted").message.message_id - page = first.views.dm_display_page("peer", worktree=str(first.root)) + page = first.views.dm_display_page(ThreadTarget("peer"), worktree=str(first.root)) assert isinstance(page.selection, DMSelection) and page.newest_seq is not None for comms, target, through, error in ( (second, "peer", page.newest_seq, DMScopeChanged), @@ -96,7 +96,7 @@ def test_painted_bound_uses_existing_transcript_refresh_error(tmp_path: Path): comms.registry.declare(_thread(tmp_path, "peer")) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "painted").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) with pytest.raises(TranscriptChanged): page.selection.basis.acknowledge( comms.bus, comms.registry, comms.root, "peer", @@ -110,7 +110,7 @@ def test_root_replaced_after_fetch_rejects_mark_without_reading_new_root(tmp_pat comms.registry.declare(_thread(tmp_path, "peer")) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "painted in original root").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) previous = tmp_path.with_name(tmp_path.name + "-previous") tmp_path.rename(previous) shutil.copytree(previous, tmp_path, symlinks=True) @@ -124,7 +124,7 @@ def test_independent_read_is_idempotent_and_never_retargets_page(tmp_path: Path) comms.registry.declare(_thread(tmp_path, "peer")) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "painted").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) comms.views.mark_user_view_read("peer", worktree=str(tmp_path)) comms.messaging.send("peer", viewer, "unpainted after explicit mark").message.message_id comms.views.mark_page_read(page, worktree=str(tmp_path)) @@ -134,7 +134,7 @@ def test_independent_read_is_idempotent_and_never_retargets_page(tmp_path: Path) def test_empty_dm_basis_is_no_op_without_creating_read_ledger(tmp_path: Path): comms = installed_comms(tmp_path) comms.registry.declare(_thread(tmp_path, "peer")) - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert page.selection.basis.displayed.tail == EmptyTail() page.selection.basis.acknowledge( comms.bus, comms.registry, comms.root, "peer", worktree=str(tmp_path), through=0, diff --git a/tests/test_dm_scoped_ack_successor.py b/tests/test_dm_scoped_ack_successor.py index dd298368d1..f0a42c494c 100644 --- a/tests/test_dm_scoped_ack_successor.py +++ b/tests/test_dm_scoped_ack_successor.py @@ -8,7 +8,7 @@ import pytest -from agent_comms.channel_targets import CommandTarget +from agent_comms.channel_targets import CommandTarget, ThreadTarget from agent_comms.comms import wire from agent_comms.child_process import ProcessIdentity from conftest import installed_comms @@ -41,7 +41,7 @@ def test_alias_rename_invalidates_old_basis_but_fresh_alias_page_can_mark(tmp_pa comms.registry.declare(_peer(tmp_path, "peer")) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "before rename").message.message_id - old = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + old = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) comms.threads.rename_managed_thread("peer", "renamed", owner=ProcessIdentity.capture(os.getpid())) markers = tmp_path / "read_ledger.json" marker_before = markers.read_bytes() if markers.exists() else None @@ -52,7 +52,7 @@ def test_alias_rename_invalidates_old_basis_but_fresh_alias_page_can_mark(tmp_pa # The old spelling is now an alias for the SAME incarnation. A fresh # page, not stale name text, is authoritative after the rename. - fresh = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + fresh = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert isinstance(fresh.selection, DMSelection) and fresh.selection.basis.peer == "renamed" _mark(comms, "peer", tmp_path, fresh) assert comms.bus.pending_count(viewer, CommandTarget.named("renamed")) == 0 @@ -65,7 +65,7 @@ def test_delete_and_same_name_reregister_invalidates_old_peer_basis(tmp_path: Pa comms.registry.declare(_peer(tmp_path, "peer")) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "old peer painted").message.message_id - old = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + old = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert isinstance(old.selection, DMSelection) comms.registry.unregister("peer") comms.registry.remove("peer") @@ -79,7 +79,7 @@ def test_delete_and_same_name_reregister_invalidates_old_peer_basis(tmp_path: Pa _mark(comms, "peer", tmp_path, old) assert (markers.read_bytes() if markers.exists() else None) == marker_before assert comms.bus.pending_count(viewer, CommandTarget.named("peer")) == 1 - fresh = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + fresh = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert isinstance(fresh.selection, DMSelection) assert fresh.selection.basis.peer_identity != old.selection.basis.peer_identity @@ -89,7 +89,7 @@ def test_viewer_rebind_and_foreign_worktree_reject_old_basis(tmp_path: Path): comms.registry.declare(_peer(tmp_path, "peer")) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "old viewer painted").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert isinstance(page.selection, DMSelection) with pytest.raises(DMScopeChanged): comms.views.mark_page_read(page, worktree=str(tmp_path / "foreign")) @@ -109,7 +109,7 @@ def test_old_viewer_alive_but_new_human_selected_rejects_stale_basis(tmp_path: P comms.registry.declare(_peer(tmp_path, "peer")) old_viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", old_viewer, "old viewer painted").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) assert isinstance(page.selection, DMSelection) and page.selection.basis.viewer == old_viewer comms.registry.declare(comms.registry.new_thread('new_user', frozenset(), str(tmp_path), role=UserThreadRole)) # Normal registry rename retains the old human declaration but moves its @@ -134,7 +134,7 @@ def test_newest_page_mark_leaves_older_message_unread(tmp_path: Path): viewer = comms.messaging.user_identity(str(tmp_path)).name older = comms.messaging.send("peer", viewer, "older").message newest = comms.messaging.send("peer", viewer, "newest").message - page = comms.views.dm_display_page("peer", worktree=str(tmp_path), bounds=PageBounds(limit=1)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path), bounds=PageBounds(limit=1)) assert page.messages == (newest,) _mark(comms, "peer", tmp_path, page) assert comms.bus.inbox(viewer, CommandTarget.named("peer")) == [older] @@ -146,7 +146,7 @@ def test_lifecycle_and_marker_lock_order_does_not_deadlock(tmp_path: Path): comms.registry.declare(_peer(tmp_path, name)) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("peer", viewer, "painted").message.message_id - page = comms.views.dm_display_page("peer", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path)) with ThreadPoolExecutor(max_workers=2) as pool: marked = pool.submit(_mark, comms, "peer", tmp_path, page) lifecycle = pool.submit( diff --git a/tests/test_historical_views.py b/tests/test_historical_views.py index e14ab93945..da3a0c155d 100644 --- a/tests/test_historical_views.py +++ b/tests/test_historical_views.py @@ -8,7 +8,7 @@ import pytest from agent_comms.bus import Bus, PublicRecord -from agent_comms.channel_targets import CommandTarget +from agent_comms.channel_targets import CommandTarget, ThreadTarget from agent_comms.comms import wire from agent_comms.coordination_errors import HistoricalSourceChanged from agent_comms.historical_views import HistoricalMessage, HistoryArchive @@ -87,7 +87,7 @@ def migrated(tmp_path): "method,args,reader", [ ("channel_display_page", ("#team",), lambda live: dict(viewer=live.views.display_viewer(str(live.root)))), - ("dm_display_page", ("alice",), lambda live: dict(worktree=str(live.root))), + ("dm_display_page", (ThreadTarget("alice"),), lambda live: dict(worktree=str(live.root))), ("channel_display_page", ("#any",), lambda live: dict(viewer=live.views.display_viewer(str(live.root)))), ], ) @@ -159,7 +159,7 @@ def test_select_painted_page_keeps_its_source_with_attached_history(migrated, co "#team", viewer=live.views.display_viewer(str(live.root)), bounds=bounds, ) else: - page = live.views.dm_display_page("alice", worktree=str(live.root), bounds=bounds) + page = live.views.dm_display_page(ThreadTarget("alice"), worktree=str(live.root), bounds=bounds) message = page.messages[-1] selected = page.select((message.seq,)) assert selected.messages == (message,) @@ -309,13 +309,19 @@ def test_normal_comms_api_contains_history_and_execution_stays_live(migrated): live.bus.bus).messages) == 4 +def root_bytes(root: Path) -> dict[Path, bytes]: + """The bytes of every member the backup owner records for this root.""" + return {path: (root / path).read_bytes() + for members in bus_files(root).values() for path in members} + + def test_attach_only_reads_original_roots_and_rejects_changed_snapshot(tmp_path): old = setup(tmp_path / "old") (old.root / "unknown-input.json").write_text('{"verdict":"UNKNOWN"}') - original = {name: (old.root / name).read_bytes() for name in bus_files(old.root)} + original = root_bytes(old.root) live = setup(tmp_path / "live", 1) source = live.views.attach_history(old.root) - assert {name: (old.root / name).read_bytes() for name in bus_files(old.root)} == original + assert root_bytes(old.root) == original assert not (Path(source.root) / "unknown-input.json").exists() page = live.views.channel_display_page( "#team", viewer=NamedViewer(live.messaging.user_identity(str(live.root)).name), @@ -348,7 +354,7 @@ def test_missing_historical_peer_is_browsable_without_live_registration(migrated _, _, live, source = migrated live.registry.unregister("alice") live.registry.remove("alice") - page = live.views.dm_display_page("alice", worktree=str(live.root)) + page = live.views.dm_display_page(ThreadTarget("alice"), worktree=str(live.root)) assert page.messages assert all(isinstance(message, HistoricalMessage) for message in page.messages) assert "alice" not in live.registry diff --git a/tests/test_ordinary_nk_delivery.py b/tests/test_ordinary_nk_delivery.py index 344b26c007..8e46981449 100644 --- a/tests/test_ordinary_nk_delivery.py +++ b/tests/test_ordinary_nk_delivery.py @@ -7,6 +7,7 @@ import os +from agent_comms.channel_targets import ThreadTarget from agent_comms.child_process import ProcessIdentity from conftest import installed_comms from test_coordinated_runtime import tmp_path as private_root_fixture @@ -30,4 +31,4 @@ def test_fresh_send_uses_canonical_publication_and_human_delivery_has_no_wake(tm assert isinstance(human.message.role, NoticeRole) assert human.audience.recipients == () assert human.decisions == () - assert comms.views.dm_display_page("sender", worktree=str(tmp_path)).messages[-1] == reply + assert comms.views.dm_display_page(ThreadTarget("sender"), worktree=str(tmp_path)).messages[-1] == reply diff --git a/tests/test_outbound_transcript_source.py b/tests/test_outbound_transcript_source.py index eeefd98c3d..9509a7982b 100644 --- a/tests/test_outbound_transcript_source.py +++ b/tests/test_outbound_transcript_source.py @@ -19,7 +19,6 @@ UnrecordedProgress, ) from agent_comms.pi_payloads import PublishedMessageToolDetails -from agent_comms.thread_identity import RecordedThreadIncarnation from agent_comms.tool_caller import NoThreadCaller from agent_comms.tools import invoke_tool from agent_comms.transcripts import TranscriptRead @@ -98,7 +97,7 @@ def test_frozen_unhandled_delivery_projects_recovery_without_claiming(tmp_path): beta = next(notice for notice in notices if notice.recipient=='beta') assert beta.state == ContextualProgress(UnrecordedProgress()) assert beta.state.detail == '' - assert beta.recipient_identity.recipient_lookup==RecordedThreadIncarnation("fixture", comms.registry.require('beta').created_at).lookup + assert beta.recipient_identity.recipient_lookup==comms.registry.require('beta').incarnation.require_recorded().lookup with Coordination(str(root/'coordination.sqlite3')) as store: assert WakeAssignment.select(store.session._connection)==before assert not any(row.wire_seq==message.seq for row in before) @@ -148,7 +147,7 @@ def test_publication_intent_joins_original_sender_and_target_only(tmp_path): # comms.root, comms.registry.require("beta"), comms.bus.bus ) original = source.rows(sequences=(reply.seq,)) - lookup = RecordedThreadIncarnation("fixture", source.recipient.created_at).lookup + lookup = source.recipient.require_recorded().lookup intent = TurnRouting((initial.message.reference,), MessageRoute("beta", ("sender", "alpha"))) # A later send can fail; this first committed subset stays a valid relation. intent.require_publications(original, lookup) diff --git a/tests/test_read_ledger.py b/tests/test_read_ledger.py index 15d8ec974f..048c70443c 100644 --- a/tests/test_read_ledger.py +++ b/tests/test_read_ledger.py @@ -7,7 +7,7 @@ import pytest from agent_comms.bus import Bus -from agent_comms.channel_targets import CommandTarget +from agent_comms.channel_targets import CommandTarget, ThreadTarget from agent_comms.child_process import ProcessIdentity from agent_comms.comms import wire from agent_comms.coordination_errors import DMPaintedBoundChanged, StaleRevision @@ -189,10 +189,10 @@ def test_turn_lease_and_unrelated_registry_changes_do_not_invalidate_dm(tmp_path comms.registry.register(replace(peer, process_identity=ProcessIdentity.capture(os.getpid()))) viewer = comms.messaging.user_identity(str(tmp_path)).name comms.messaging.send("alice", viewer, "painted").message.message_id - page = comms.views.dm_display_page("alice", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path)) assert page.selection.basis.peer_identity == peer.incarnation comms.registry.lease_local_turn("alice", "ordinary-turn") - fresh = comms.views.dm_display_page("alice", worktree=str(tmp_path)) + fresh = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path)) assert fresh.selection.basis.peer_identity == page.selection.basis.peer_identity comms.registry.register(comms.registry.new_thread('unrelated', frozenset(), str(tmp_path))) comms.views.mark_page_read(page, worktree=str(tmp_path)) @@ -211,9 +211,10 @@ def test_dm_ack_survives_process_exit_and_rebind_does_not_inherit_reads(tmp_path sys.executable, "-c", "import os, sys\n" + "from agent_comms.channel_targets import ThreadTarget\n" "from agent_comms.comms import wire\n" "comms = wire(sys.argv[1])\n" - "page = comms.views.dm_display_page('alice', worktree=sys.argv[1])\n" + "page = comms.views.dm_display_page(ThreadTarget('alice'), worktree=sys.argv[1])\n" "comms.views.mark_page_read(page, worktree=sys.argv[1])\n" "os._exit(9)\n", str(tmp_path), @@ -353,7 +354,7 @@ def test_recorded_bus_identity_is_one_member(tmp_path): assert not ledger.read().bus_identity.matches(ObservedPath.at(comms.bus.bus.path)) assert comms.views.viewer_snapshot(str(tmp_path)).unread["alice"].count == 1 - page = comms.views.dm_display_page("alice", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path)) comms.views.mark_page_read(page, worktree=str(tmp_path)) recorded = RecordedBus(FileIdentity.from_stat(comms.bus.bus.path.stat())) assert ledger.read().bus_identity == recorded == page.selection.basis.bus_identity @@ -374,7 +375,7 @@ def test_recorded_bus_identity_is_one_member(tmp_path): assert reopened.views.viewer_snapshot(str(tmp_path)).unread["alice"].count == 1 with pytest.raises(DMScopeChanged): reopened.views.mark_page_read(page, worktree=str(tmp_path)) - fresh = reopened.views.dm_display_page("alice", worktree=str(tmp_path)) + fresh = reopened.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path)) reopened.views.mark_page_read(fresh, worktree=str(tmp_path)) assert reopened.views.viewer_snapshot(str(tmp_path)).unread["alice"].count == 0 @@ -447,7 +448,7 @@ def current_cursor(facts): comms.channels.set_channel_any_mode("#team", rng.choice([True, False])) elif operation == "display": if rng.choice([True, False]): - page = comms.views.dm_display_page("alice", worktree=str(tmp_path), bounds=PageBounds(limit=3)) + page = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path), bounds=PageBounds(limit=3)) basis = page.selection.basis.displayed kind, target = "dm", "alice" proof = page.selection.basis diff --git a/tests/test_thread_identity.py b/tests/test_thread_identity.py index 1aa92e9cc5..ec25b4610a 100644 --- a/tests/test_thread_identity.py +++ b/tests/test_thread_identity.py @@ -7,6 +7,7 @@ import pytest +from agent_comms.channel_targets import ThreadTarget from agent_comms.session_file import SavedSessionFile from agent_comms.child_process import NoProcess, ProcessIdentity from agent_comms.comms import wire @@ -248,7 +249,7 @@ def test_saved_read_ledger_survives_reopen_and_new_ack(tmp_path): comms.registry.declare(comms.registry.new_thread('owner', frozenset(), str(tmp_path))) viewer = comms.messaging.user_identity(str(tmp_path)).name message = comms.messaging.send("owner", viewer, "already painted").message - page = comms.views.dm_display_page("owner", worktree=str(tmp_path)) + page = comms.views.dm_display_page(ThreadTarget("owner"), worktree=str(tmp_path)) comms.views.mark_page_read(page, worktree=str(tmp_path)) path = comms.bus.reads.path document = FieldCodec.decode(ReadDocument, json.loads(path.read_text())) @@ -262,7 +263,7 @@ def test_saved_read_ledger_survives_reopen_and_new_ack(tmp_path): conversation = reopened.bus.reads.conversation(message, snapshot) assert message.seq in reopened.bus.reads.cursor(viewer, snapshot).for_conversation(conversation) next_message = reopened.messaging.send("owner", viewer, "new paint").message - page = reopened.views.dm_display_page("owner", worktree=str(tmp_path)) + page = reopened.views.dm_display_page(ThreadTarget("owner"), worktree=str(tmp_path)) reopened.views.mark_page_read(page, worktree=str(tmp_path)) seen = reopened.bus.reads.cursor(viewer, reopened.registry.snapshot()).for_conversation(conversation) assert all(sequence in seen for sequence in (message.seq, next_message.seq)) diff --git a/tests/test_user_thread_lookup.py b/tests/test_user_thread_lookup.py index dadc74e996..91b56dd0d7 100644 --- a/tests/test_user_thread_lookup.py +++ b/tests/test_user_thread_lookup.py @@ -1,6 +1,6 @@ """The stored registry revision owns human identity and UI history uses it.""" -from agent_comms.channel_targets import CommandTarget +from agent_comms.channel_targets import CommandTarget, ThreadTarget from dataclasses import replace @@ -41,7 +41,7 @@ def test_channel_and_dm_history_keep_the_human_as_sender(tmp_path): assert channel.sender_role is direct.sender_role is UserThreadRole channel_page = comms.views.channel_display_page( '#comms', viewer=comms.views.display_viewer(str(tmp_path))) - direct_page = comms.views.dm_display_page('peer', worktree=str(tmp_path)) + direct_page = comms.views.dm_display_page(ThreadTarget('peer'), worktree=str(tmp_path)) assert channel.message_id in {message.message_id for message in channel_page.messages} assert direct.message_id in {message.message_id for message in direct_page.messages} assert all(message.sender == human.name for message in (*channel_page.messages, *direct_page.messages)) From 4c535dd675ff8d9b0bff1f532c59d5366330ad4e Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:06:44 -0400 Subject: [PATCH 02/10] Declare the retired commander file durable and stop tests restating owners - RetiredCommanderFile is retained bus state its carry consumes: a DurableFile, so every declared root file is durable or runtime again (a012a03a1, #2743). - The ratchet guard composes owners through Black, so the formatting measure counts nothing and each comparison measures only the classes; 97256db52 flipped the expectation to status 0 while the fixture stayed unformatted. - bus_files keys are DeclaredFile since #2709: assert the declared owner. - Routes are written owner-only by their owner; the backup reads the saved route through read_active_route since #2731. - The bus-history round trip runs every step the chain orders before the goal-request carry, under the maintenance hold, instead of a hand-picked subset that #2701's claim-field carry fell between. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/locked_store.py | 2 +- tests/guards/test_debt_ratchet.py | 18 +++++++++++++----- tests/test_bus_files.py | 4 ++-- tests/test_carry_bus_history.py | 22 ++++++++++++++-------- tests/test_store_backup.py | 2 ++ 5 files changed, 32 insertions(+), 16 deletions(-) diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 2bbc41c9cd..167f6e98fe 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -390,7 +390,7 @@ def copy_group(cls, source: Path, target: Path) -> frozenset[Path]: return copied -class RetiredCommanderFile(RootFile): +class RetiredCommanderFile(DurableFile): """The retired commander file that tools/cutover/carry_commander_project.py consumes. The practice copy holds declared files only, so an undeclared input made the practice carry find nothing while the live carry deleted it. Deleted with that carry.""" diff --git a/tests/guards/test_debt_ratchet.py b/tests/guards/test_debt_ratchet.py index 4be1de7b30..13a9c89234 100644 --- a/tests/guards/test_debt_ratchet.py +++ b/tests/guards/test_debt_ratchet.py @@ -429,6 +429,14 @@ def owner(name: str, lines: int, indent: str = "") -> str: ) +def module(*owners: str) -> str: + """The owners as one module in Black's form, so the formatting ratchet counts nothing and + each comparison measures only the classes.""" + import black + + return black.format_str("".join(owners), mode=black.Mode()) + + @pytest.mark.parametrize("root", ["src/agent_comms", "src/toad", "src/textual"]) def test_family_flattening_cannot_offset_another_file_or_admit_a_named_facade( tmp_path: Path, root: str, ratchet @@ -512,7 +520,7 @@ def test_family_name_owner_reads_and_codec_detours(source: str, expected: int): def test_small_owners_stay_within_the_class_threshold(repo: Repository) -> None: base = repo.commit({"owners.py": owner("Small", 2)}) - head = repo.commit({"owners.py": owner("Small", 500) + owner("New", 500)}) + head = repo.commit({"owners.py": module(owner("Small", 500), owner("New", 500))}) status, report = repo.compare(base, head) assert status == 0 assert report["delta"][f"GodClassExcess:{repo.root}/owners.py::Small"] == 0 @@ -520,8 +528,8 @@ def test_small_owners_stay_within_the_class_threshold(repo: Repository) -> None: def test_class_growth_cannot_be_cancelled_by_another_class(repo: Repository) -> None: - base = repo.commit({"classes.py": owner("Growing", 501) + owner("Shrinking", 600)}) - head = repo.commit({"classes.py": owner("Growing", 502) + owner("Shrinking", 500)}) + base = repo.commit({"classes.py": module(owner("Growing", 501), owner("Shrinking", 600))}) + head = repo.commit({"classes.py": module(owner("Growing", 502), owner("Shrinking", 500))}) status, report = repo.compare(base, head) assert status == 1 assert report["delta"][f"GodClassExcess:{repo.root}/classes.py::Growing"] == 1 @@ -540,13 +548,13 @@ def test_new_or_crossing_god_class_is_refused(repo: Repository, before: int) -> def test_class_move_keeps_baseline_and_new_small_owner_has_no_class_debt(repo: Repository) -> None: base = repo.commit({"old.py": owner("Existing", 501)}) - moved = repo.commit({"old.py": None, "new.py": owner("Existing", 501) + owner("NewOwner", 25)}) + moved = repo.commit({"old.py": None, "new.py": module(owner("Existing", 501), owner("NewOwner", 25))}) status, report = repo.compare(base, moved) assert status == 0 assert report["base"][f"GodClassExcess:{repo.root}/new.py::Existing"] == 1 assert report["delta"][f"GodClassExcess:{repo.root}/new.py::Existing"] == 0 assert report["delta"][f"GodClassExcess:{repo.root}/new.py::NewOwner"] == 0 - grown = repo.commit({"new.py": owner("Existing", 502) + owner("NewOwner", 26)}) + grown = repo.commit({"new.py": module(owner("Existing", 502), owner("NewOwner", 26))}) assert repo.compare(moved, grown)[0] == 1 diff --git a/tests/test_bus_files.py b/tests/test_bus_files.py index 663347c0c8..65dbd40916 100644 --- a/tests/test_bus_files.py +++ b/tests/test_bus_files.py @@ -14,7 +14,7 @@ from agent_comms.comms import Comms, wire from agent_comms.coordination_session import NativeBoundary from agent_comms.envelope_claim_transitions import ExistingFileClaim, NativeClaim -from agent_comms.locked_store import DurableFile, RootFile, RuntimeFile, SQLiteFile +from agent_comms.locked_store import DeclaredFile, DurableFile, RootFile, RuntimeFile, SQLiteFile from agent_comms.recorded_time import RecordedEventTime from agent_comms.repository_name import RepositoryName from agent_comms.todos import TodoStore @@ -81,7 +81,7 @@ def test_parameterized_sqlite_resources_follow_their_file_owner(tmp_path: Path, actual = {path for path in tmp_path.rglob("*") if path.is_file()} assert len(actual) > 1 # The real SQLite writer has an open journal/WAL. assert actual <= NativeInputProof.files_in(tmp_path) - assert str(proof.relative_to(tmp_path)) in bus_files(tmp_path) + assert DeclaredFile(NativeInputProof, proof.relative_to(tmp_path)) in bus_files(tmp_path) def test_installer_reconciles_a_native_proof_journal_onto_its_declaration(tmp_path: Path) -> None: diff --git a/tests/test_carry_bus_history.py b/tests/test_carry_bus_history.py index 60cce66493..a472f89b5f 100644 --- a/tests/test_carry_bus_history.py +++ b/tests/test_carry_bus_history.py @@ -181,14 +181,20 @@ def test_predecessor_publisher_history_round_trips_into_bus(tmp_path): spill = tmp_path / "hdd-spill" spill.mkdir() StoreBackup.copy_tree(original, root) - # This publisher authored a pre-carry registry. Its finished-turn carry - # precedes the goal-request carry and every current registry reader. - finished = FinishedTurns(root, original, spill) - finished_dry_run = finished.carry(apply=False) - finished_result = finished.carry(apply=True) - assert finished_result.changed > 0 - assert finished_result.detail == finished_dry_run.detail - assert finished.carry(apply=False).changed == 0 + # This publisher authored a pre-carry registry: every step the chain orders before the + # goal-request carry runs first, as the installer runs them. + preceding = {} + maintenance = MaintenanceBarrier(RegistryStore.at(root).path) + hold = maintenance.begin(MaintenanceOperator.current()) + with hold.holding(): + for step in chain.steps[:chain.steps.index(OwnerRequests)]: + carry = step(root, original, spill) + dry_run = carry.carry(apply=False) + preceding[step] = carry.carry(apply=True) + assert preceding[step].detail == dry_run.detail + assert carry.carry(apply=False).changed == 0 + maintenance.clear(hold) + assert preceding[FinishedTurns].changed > 0 relatives = (Path(), archived.relative_to(original)) originals = {relative: (original / relative / BusHistory.filename).read_bytes() for relative in relatives} diff --git a/tests/test_store_backup.py b/tests/test_store_backup.py index 8e7a4c74d7..fa65b2620f 100644 --- a/tests/test_store_backup.py +++ b/tests/test_store_backup.py @@ -129,6 +129,7 @@ def test_store_backup_restores_wal_rows_and_removes_new_files(tmp_path): root.mkdir() store = root / 'transcripts.sqlite3' route = tmp_path / 'route.json' + route.touch(mode=0o600) # The route's owner writes it private. route.write_text('original route') created = root / 'created.json' with sqlite3.connect(store) as writer: @@ -222,6 +223,7 @@ def test_route_only_backup_preserves_a_concurrent_change(tmp_path, had_route): root.mkdir() route = tmp_path / 'route.json' if had_route: + route.touch(mode=0o600) # The route's owner writes it private. route.write_text('original route') spill_dir = tmp_path / 'spill' backup = StoreBackup.take(root, route, (), spill_dir) From 747e00671543ce02500f8a97bb7152fc1adcb57f Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:10:04 -0400 Subject: [PATCH 03/10] Delete the orphaned fake helpers that kill refusal-test collection e1e82d650 (#2684) deleted test_backend._replay_stub and _NATIVE_PROMPT_START; test_native_refusal_recovery.py still imported them for helpers no test calls since e16d8df09 deleted the fake-backed tests. Collection died on main. The one real test stays. Co-Authored-By: Claude Opus 5.5 --- tests/test_native_refusal_recovery.py | 40 --------------------------- 1 file changed, 40 deletions(-) diff --git a/tests/test_native_refusal_recovery.py b/tests/test_native_refusal_recovery.py index 84fac30c40..04faeb05d6 100644 --- a/tests/test_native_refusal_recovery.py +++ b/tests/test_native_refusal_recovery.py @@ -1,50 +1,10 @@ """Native refusal recovery: Core reports one outcome, never blocks on a retry.""" from agent_comms.native_arguments import SelectedModel -import json - import pytest import agent_comms.native_pi as native_pi import agent_comms.pi_events as pi_events -from test_backend import _NATIVE_PROMPT_START, _replay_stub - -OVERLOADED = "Our servers are currently overloaded. Please try again later." - - - -def _stub_with(tmp_path, records): - lines = "\n".join(json.dumps(record) for record in records) - return _replay_stub( - tmp_path, - lines, - startup="def emit(value):\n print(json.dumps(value), flush=True)\n" - + _NATIVE_PROMPT_START - + "prompt_request = prompt\n", - ) - - -def _refused_attempt(attempt, delay_ms): - """The recovery announces its retry decision before the refused message_end.""" - return [ - {"type": "message_start", "message": {"role": "assistant", "content": []}}, - {"type": "native_retry_start", "attempt": attempt, "maxAttempts": 3, - "delayMs": delay_ms, "errorMessage": OVERLOADED}, - {"type": "message_end", "message": {"role": "assistant", "stopReason": "error", - "errorMessage": OVERLOADED, "content": []}}, - ] - - -def _settled_records(): - return [ - {"type": "agent_settled"}, - {"type": "response", "command": "get_session_stats", "success": True, - "data": {"contextUsage": {}}}, - ] - - - - @pytest.mark.asyncio From f228d09cc46dc32f3e8260c51de33576098b5299 Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:31:44 -0400 Subject: [PATCH 04/10] Finish killed restores, refuse undeclared bus files, and stop tests restating owners - StoreBackup.complete_interrupted gives StoreBackup.interrupted its caller: the install entry finishes and retires any restore a killed install left, under a maintenance hold, before anything reads the bus. - RootFile refuses a declaration that is neither DurableFile nor RuntimeFile; UnfinishedOwnerRename (from #2749) becomes DurableFile. - UnionDecoder.decode_candidates: when no alternative's shape admits the value, the first alternative's own decode states the refusal, so total failure keeps its declared cause again (lost in #2712). - Tests use the owners: DeclaredWireType and RecordMember members in the union test, typed carried paths, a committed GivenThreadText title, and rehearsal buses owned by the installed runtime and read through its CLI. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/field_codec.py | 7 +++-- src/agent_comms/installer/backup.py | 16 ++++++++++ src/agent_comms/installer/upgrade.py | 1 + src/agent_comms/locked_store.py | 20 ++++++++---- tests/test_carry_dead_turns.py | 2 +- tests/test_field_codec.py | 26 +++++----------- tests/test_install_rehearsal.py | 46 ++++++++++++++++++---------- tests/test_interrupted_rollback.py | 14 +++++---- tools/cutover/thread_name_owners.py | 4 +-- 9 files changed, 83 insertions(+), 53 deletions(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index f5fce8564e..9ea8ed0994 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -2415,10 +2415,11 @@ def decode_candidates( admits: Callable[[ValueDecoder[T]], bool], ) -> T: first_error = None + # Shapes only rule alternatives out; when they rule out every one, the first + # alternative's own decode states the refusal. + candidates = tuple(filter(admits, self.alternatives)) or self.alternatives[:1] try: - for alternative in self.alternatives: - if not admits(alternative): - continue + for alternative in candidates: try: return decode(alternative) except MalformedRecord as error: diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index 4bdacd754e..ec0d68086e 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -26,6 +26,7 @@ from agent_comms.field_codec import FieldCodec, MalformedRecord, PathText from agent_comms.installer.interpreter_operation import RouteSelection from agent_comms.locked_store import DeclaredFile, LockedStore, RootFile, RuntimeFile, UndeclaredFile +from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator from agent_comms.private_path import FileObservation, FileRevision, MissingFile from agent_comms.registry_store import RegistryStore from agent_comms.store_files import AtomicFile @@ -199,6 +200,21 @@ def interrupted(cls, root: Path, spill_dir: Path) -> tuple[StoreBackup, ...]: ) return tuple(backup for backup in backups if not backup.original()) + @classmethod + def complete_interrupted(cls, root: Path, spill_dir: Path) -> tuple[Path, ...]: + """Finish every restore a killed install left unfinished, before anything reads + the bus: take its abandoned maintenance hold, restore, release, then retire the + backup so the next install never meets it again.""" + retired = [] + for backup in cls.interrupted(root, spill_dir): + maintenance = MaintenanceBarrier(RegistryStore.at(root).path) + hold = maintenance.begin(MaintenanceOperator.current()) + with hold.holding(): + backup.restore() + maintenance.clear(hold) + retired.append(backup.retire()) + return tuple(retired) + def retire(self) -> Path: archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}') self.path.rename(archived) diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index 4c3c6879da..cdf87bf6dd 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -760,6 +760,7 @@ def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice, tree = TargetTree(checkout, scratch, target) bridges = report.bridges with report.running(tree): + StoreBackup.complete_interrupted(root, spill_dir) installed = Stack(checkout, commands) current = commands.run("git", "-C", checkout, "rev-parse", "HEAD").strip() ancestry = ["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target] diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 3cf81942d6..c5a69b4c5b 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -59,6 +59,14 @@ class RootFile: filename: ClassVar[str] root_directory: ClassVar[str] = "." + def __init_subclass__(cls, **kwargs) -> None: + """A declared file is either retained across installations or rewritten by its + running owner; a declaration that is neither has no install behaviour at all.""" + super().__init_subclass__(**kwargs) + if "filename" in vars(cls) and not issubclass(cls, (DurableFile, RuntimeFile)): + raise TypeError(f"{cls.__qualname__} declares {cls.filename!r} but is neither " + "a DurableFile nor a RuntimeFile") + @classmethod def path_in(cls, root: Path | str, **parts: str) -> Path: return Path(root) / cls.filename.format(**parts) @@ -340,6 +348,12 @@ class DurableFile(RootFile): __slots__ = () +class RuntimeFile(RootFile): + """A bus file no carry writes: running owners rewrite it at any time, maintenance included.""" + + __slots__ = () + + class BusFile(DurableFile): """Declared data and lifecycle files that rule out a fresh bus installation.""" @@ -660,12 +674,6 @@ def publish_change(cls, path: Path) -> None: ) from error -class RuntimeFile(RootFile): - """A bus file no carry writes: running owners rewrite it at any time, maintenance included.""" - - __slots__ = () - - @dataclass(frozen=True, slots=True) class LockedStore(FileStore, StoreLockOwner, JsonDocument[T], declared_name=None): """Ordered document transactions; callbacks never mutate or reenter.""" diff --git a/tests/test_carry_dead_turns.py b/tests/test_carry_dead_turns.py index 376fea5d95..ef53f8538e 100644 --- a/tests/test_carry_dead_turns.py +++ b/tests/test_carry_dead_turns.py @@ -20,6 +20,6 @@ def test_deleted_turn_files_are_named_and_the_current_one_kept(tmp_path): step = DeadTurns(comms.root, comms.root, tmp_path / "spill") assert step.carry(apply=False).rewritten == () carried = step.carry(apply=True) - assert carried.rewritten == (str(dead.relative_to(comms.root)),) + assert carried.rewritten == (dead.relative_to(comms.root),) assert current.exists() and not dead.exists() assert step.carry(apply=False).changed == 0 diff --git a/tests/test_field_codec.py b/tests/test_field_codec.py index 410e262d66..5451efb836 100644 --- a/tests/test_field_codec.py +++ b/tests/test_field_codec.py @@ -9,6 +9,7 @@ from agent_comms.declared_family import DeclaredFamily, UnknownFamilyMember from agent_comms.field_codec import ( + DeclaredWireType, FieldCodec, InlineRecord, JsonShapeFamily, @@ -16,6 +17,7 @@ MalformedRecord, NullMember, PathText, + RecordMember, RecordedTimestamp, SubclassUnion, TextRepresentation, @@ -278,7 +280,8 @@ class Second(SubclassUnion, affix=""): def declare_number(union): @dataclass(frozen=True, slots=True) - class Number(union): + class Number(DeclaredWireType, union): + wire_type = int value: int def to_wire(self): @@ -289,18 +292,14 @@ def from_wire(cls, data): calls.append(cls) return cls(FieldCodec.decode(int, data)) - @classmethod - def schema(cls): - return FieldCodec.value_schema(int) - return Number first = declare_number(First) second = declare_number(Second) @dataclass(frozen=True, slots=True) - class Text(OwnedWireText, First): - pass + class Text(DeclaredWireType, OwnedWireText, First): + wire_type = str @dataclass(frozen=True, slots=True) class Empty(NullMember, First): @@ -323,20 +322,9 @@ class Empty(NullMember, First): FieldCodec.decode(Second, "text") @dataclass(frozen=True, slots=True) - class Later(First): + class Later(RecordMember, First): entries: tuple[int, ...] - def to_wire(self): - return FieldCodec.encode_record(self) - - @classmethod - def from_wire(cls, data): - return FieldCodec.decode_record(cls, data) - - @classmethod - def schema(cls): - return FieldCodec.record_schema(cls) - encoded = {"entries": [1, 2]} assert FieldCodec.encode(Later((1, 2))) == encoded assert type(FieldCodec.decode(First, encoded)) is Later diff --git a/tests/test_install_rehearsal.py b/tests/test_install_rehearsal.py index fc90ee57cc..bc1617d1fb 100644 --- a/tests/test_install_rehearsal.py +++ b/tests/test_install_rehearsal.py @@ -22,9 +22,8 @@ import pytest -from agent_comms.cli_commands import HistoryCliCommand -from agent_comms.comms import wire from agent_comms.field_codec import FieldCodec +from agent_comms.history_result import HistoryPageResult from agent_comms.installer.backup import bus_files from agent_comms.installer.commands import InstallCommands from agent_comms.installer.practice import TestHome @@ -37,9 +36,11 @@ Stack, Upgrade, ) +from agent_comms.launch_runtime import LaunchRuntime from agent_comms.maintenance_barrier import MaintenanceBarrier from agent_comms.registry_store import RegistryStore from agent_comms.repo import Checkout, ScratchCheckout +from agent_comms.root_runtime import RootRuntime from agent_comms.storage_rules import StorageRule from agent_comms.typed_table import ReconciledFile from conftest import declare_test_bus, publish_initial @@ -77,6 +78,16 @@ def installed_home(tmp_path_factory) -> dict[str, str]: return environment +@pytest.fixture +def installed_comms(private_comms, installed_home): + """An installed machine's bus belongs to its installed runtime, not to the test + process that declared it.""" + with pytest.MonkeyPatch.context() as installed: + installed.setenv("HOME", installed_home["HOME"]) + RootRuntime.publish(private_comms.root, LaunchRuntime.installed()) + return private_comms + + @pytest.fixture(scope="module") def install_checkout(tmp_path_factory) -> Path: """Unchanged targets reuse one built revision; fault cases restore this checkout.""" @@ -106,13 +117,13 @@ def drop_one_index(root: Path) -> str: @pytest.mark.parametrize("repos, drifted", ( ((), False), (("--repos",), True), ), ids=("kept-repos-current-stores", "given-repos-drifted-store")) -def test_install_rehearses_without_changing_the_bus(private_comms, installed_home, install_checkout, tmp_path, +def test_install_rehearses_without_changing_the_bus(installed_comms, installed_home, install_checkout, tmp_path, repos, drifted): """Retained carry steps inspect a practice copy, even when no step is pending. Only the deliberately drifted store is rewritten; neither case changes the bus. """ - comms = private_comms + comms = installed_comms declare_test_bus(comms, channels=("comms",), threads=( comms.registry.new_thread(name="worker", tags=frozenset(), worktree=str(tmp_path)),)) publish_initial(comms, "worker", "#all", "One message before the install") @@ -136,8 +147,8 @@ def test_install_rehearses_without_changing_the_bus(private_comms, installed_hom @pytest.mark.timeout(280) def test_failed_live_migration_restores_data_checkout_and_runtime( - private_comms, installed_home, install_checkout, tmp_path): - comms, checkout = private_comms, install_checkout + installed_comms, installed_home, install_checkout, tmp_path): + comms, checkout = installed_comms, install_checkout declare_test_bus(comms, channels=("comms",), threads=( comms.registry.new_thread(name="worker", tags=frozenset(), worktree=str(tmp_path)),)) publish_initial(comms, "worker", "#all", "Retained before migration") @@ -150,8 +161,8 @@ def test_failed_live_migration_restores_data_checkout_and_runtime( step = checkout / "tools/cutover/carry_fault.py" step.write_text('''from pathlib import Path from agent_comms.installer.carry import Carry, Carried -from agent_comms.locked_store import RootFile -class Saved(RootFile): +from agent_comms.locked_store import DurableFile +class Saved(DurableFile): filename = "saved" class Fault(Carry): opened_files = (Saved,) @@ -190,9 +201,9 @@ def carry(self, *, apply): @pytest.mark.timeout(280) def test_failed_publication_restores_absent_route_without_store_changes( - private_comms, installed_home, install_checkout, tmp_path): + installed_comms, installed_home, install_checkout, tmp_path): checkout = install_checkout - declare_test_bus(private_comms, channels=("comms",)) + declare_test_bus(installed_comms, channels=("comms",)) current = subprocess.check_output( ["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip() manifest = checkout / "pyproject.toml" @@ -222,7 +233,7 @@ def test_failed_publication_restores_absent_route_without_store_changes( assert not route.exists() completed = subprocess.run( - [pointer / "bin/agent-comms", "--root", private_comms.root, "install", + [pointer / "bin/agent-comms", "--root", installed_comms.root, "install", "--checkout", checkout, "--to", target, "--project", checkout, "--spill-dir", tmp_path / "backups"], cwd=checkout, env={**installed_home, "HOME": str(home)}, capture_output=True, text=True, timeout=270, @@ -235,17 +246,17 @@ def test_failed_publication_restores_absent_route_without_store_changes( assert blocker.read_text() == "An owner file that publication must preserve" assert subprocess.check_output( ["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip() == current - MaintenanceBarrier(RegistryStore.at(private_comms.root).path).read().phase.new_sends.require_open() + MaintenanceBarrier(RegistryStore.at(installed_comms.root).path).read().phase.new_sends.require_open() @pytest.mark.timeout(280) @pytest.mark.parametrize("fresh, live_kind", ((False, CarriedRun), (True, CreatedRun)), ids=("retained-bus", "fresh-bus")) def test_installed_entry_reports_completion_after_target_source_retirement( - private_comms, installed_home, install_checkout, tmp_path, fresh, live_kind): - root = tmp_path / "new-bus" if fresh else private_comms.root + installed_comms, installed_home, install_checkout, tmp_path, fresh, live_kind): + root = tmp_path / "new-bus" if fresh else installed_comms.root if not fresh: - declare_test_bus(private_comms, channels=("comms",)) + declare_test_bus(installed_comms, channels=("comms",)) scratch = tmp_path / "install-scratch" root.mkdir(mode=0o700, parents=True, exist_ok=True) StorageRule.path_in(root).write_text(json.dumps(FieldCodec.encode( @@ -263,7 +274,10 @@ def test_installed_entry_reports_completion_after_target_source_retirement( assert completed.returncode == 0, completed.stderr result = FieldCodec.decode(Upgrade, FieldCodec.parse(completed.stdout)) assert isinstance(result, InstalledUpgrade) and type(result.live) is live_kind - history = tuple(reversed(HistoryCliCommand(target="#comms").apply(wire(root)).messages)) + # The bus now belongs to the installed runtime, so its own CLI reads it. + read = subprocess.run([outgoing, "--root", root, "history", "--target", "#comms"], + env=installed_home, capture_output=True, text=True, check=True, timeout=60) + history = tuple(reversed(FieldCodec.decode(HistoryPageResult, FieldCodec.parse(read.stdout)).messages)) assert len(history) == (1 if fresh else 2) assert history[-1].sender == "installer" assert history[-1].body.startswith("Install finished") diff --git a/tests/test_interrupted_rollback.py b/tests/test_interrupted_rollback.py index e35d82268d..b6f8fa3758 100644 --- a/tests/test_interrupted_rollback.py +++ b/tests/test_interrupted_rollback.py @@ -19,6 +19,7 @@ from agent_comms.installer.upgrade import upgrade from agent_comms.locked_store import RegistryFile from agent_comms.registry_store import RegistryStore +from agent_comms.thread_text import GivenThreadText from bus_record_fixture import prepare_native_comms @@ -29,8 +30,9 @@ def carry(self, *, apply): if apply: with registry.editing() as edit: thread = edit.document.require("worker") - edit.document.threads[thread.name] = replace(thread, title="Carried title") - return Carried(1, rewritten=tuple(str(path.relative_to(self.root)) for path in registry.written)) + edit.document.threads[thread.name] = replace(thread, title=GivenThreadText("Carried title")) + edit.commit() + return Carried(1, rewritten=tuple(path.relative_to(self.root) for path in registry.written)) KILLED_RESTORE = """ @@ -58,9 +60,9 @@ def killed_publication(source, target, **directories): @pytest.mark.skipif(os.name != "posix", reason="SIGKILL and private registry require POSIX") @pytest.mark.parametrize("publication", (0, 1), ids=("after-registry", "after-guard")) -def test_next_install_completes_a_killed_registry_restore(tmp_path, monkeypatch, publication): - # HOME selects only this test's route. Neither process can replace the developer's route. - monkeypatch.setenv("HOME", str(tmp_path / "home")) +def test_next_install_completes_a_killed_registry_restore(tmp_path, installed_runtime, publication): + # The temporary installation's HOME selects only this test's route and runtime. + # Neither process can replace the developer's route. comms = prepare_native_comms(wire(tmp_path / "wire")) comms.registry.declare(comms.registry.new_thread("worker", frozenset(), str(tmp_path))) registry = RegistryStore.at(comms.root) @@ -85,4 +87,4 @@ def test_next_install_completes_a_killed_registry_restore(tmp_path, monkeypatch, False, spill, tmp_path / "install-scratch", lambda: None, InstallReport(), ) assert {path: StoreBackup.digest(path) for path in registry.written} == expected - assert comms.registry.require("worker").title != "Carried title" + assert comms.registry.require("worker").title != GivenThreadText("Carried title") diff --git a/tools/cutover/thread_name_owners.py b/tools/cutover/thread_name_owners.py index 0b3612b26e..49aecbae1b 100644 --- a/tools/cutover/thread_name_owners.py +++ b/tools/cutover/thread_name_owners.py @@ -17,7 +17,7 @@ from agent_comms.coordination_tables.participants import OwnerGenerations, Participants from agent_comms.errors import RelationViolationError from agent_comms.installer.carry import Carried, Carry -from agent_comms.locked_store import RootFile, SQLiteFile +from agent_comms.locked_store import DurableFile, SQLiteFile from agent_comms.native_prompt_binding import PromptBinding from agent_comms.native_runtime_input import CurrentNativeCursor, NativeRuntimeInput from agent_comms.private_sidecar import ( @@ -27,7 +27,7 @@ from agent_comms.typed_table import HeldBusFile, SQLiteCatalogObject, SQLiteSchemaObject -class UnfinishedOwnerRename(RootFile): +class UnfinishedOwnerRename(DurableFile): filename = ".private-owner-rename.pending" @classmethod From 9e9b471756367bcbd7fc47d408ca874a1c205f80 Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:34:43 -0400 Subject: [PATCH 05/10] Require assignment activity where the notification reads it SelectedResponseNotification, CompletedAssignment and the delivered state read activity.progress or call activity's notification methods, yet declared activity as optional; #2450 (69304caf5) gave the selected-response override a None default, so a missing activity became an AttributeError instead of the TypeError the call contract owes. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/assignment_states.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/assignment_states.py b/src/agent_comms/assignment_states.py index 8af56a6cbd..d0212f04b8 100644 --- a/src/agent_comms/assignment_states.py +++ b/src/agent_comms/assignment_states.py @@ -235,7 +235,7 @@ def notification( sender_lookup: str, *, observation: RecipientActivity, - activity: AssignmentActivity | None = None, + activity: AssignmentActivity, updated_at_ms: int = 0, triage_inflight: bool = False, ) -> MessageNotification: @@ -396,7 +396,7 @@ def notification( sender_lookup: str, *, observation: RecipientActivity, - activity: AssignmentActivity | None, + activity: AssignmentActivity, updated_at_ms: int = 0, triage_inflight: bool = False, ) -> MessageNotification: @@ -474,7 +474,7 @@ def notification( sender_lookup: str, *, observation: RecipientActivity, - activity: AssignmentActivity | None, + activity: AssignmentActivity, updated_at_ms: int = 0, triage_inflight: bool = False, ) -> MessageNotification: From b75a280d68ed1f894d7f233e2bd9ba8a1561ac7d Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:43:41 -0400 Subject: [PATCH 06/10] Read rehearsal results in their declared types The drifted-store case compares typed carried paths, and the installed-entry case decodes the history page messages instead of the unresolvable result shell. Co-Authored-By: Claude Opus 5.5 --- tests/test_install_rehearsal.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/test_install_rehearsal.py b/tests/test_install_rehearsal.py index bc1617d1fb..a92d803840 100644 --- a/tests/test_install_rehearsal.py +++ b/tests/test_install_rehearsal.py @@ -23,7 +23,7 @@ import pytest from agent_comms.field_codec import FieldCodec -from agent_comms.history_result import HistoryPageResult +from agent_comms.history_result import ChannelHistoryPageResult from agent_comms.installer.backup import bus_files from agent_comms.installer.commands import InstallCommands from agent_comms.installer.practice import TestHome @@ -38,6 +38,7 @@ ) from agent_comms.launch_runtime import LaunchRuntime from agent_comms.maintenance_barrier import MaintenanceBarrier +from agent_comms.messages import Message from agent_comms.registry_store import RegistryStore from agent_comms.repo import Checkout, ScratchCheckout from agent_comms.root_runtime import RootRuntime @@ -100,7 +101,7 @@ def install_checkout(tmp_path_factory) -> Path: return checkout -def drop_one_index(root: Path) -> str: +def drop_one_index(root: Path) -> Path: """One declared store's schema drifts from both runtimes' declarations.""" for path in sorted(ReconciledFile.existing_in(root)): with closing(sqlite3.connect(path)) as db: @@ -109,7 +110,7 @@ def drop_one_index(root: Path) -> str: if index: db.execute(f'DROP INDEX "{index[0]}"') db.commit() - return str(path.relative_to(root)) + return path.relative_to(root) raise AssertionError("No declared store has an index to drop") @@ -277,7 +278,8 @@ def test_installed_entry_reports_completion_after_target_source_retirement( # The bus now belongs to the installed runtime, so its own CLI reads it. read = subprocess.run([outgoing, "--root", root, "history", "--target", "#comms"], env=installed_home, capture_output=True, text=True, check=True, timeout=60) - history = tuple(reversed(FieldCodec.decode(HistoryPageResult, FieldCodec.parse(read.stdout)).messages)) + page = FieldCodec.parse(read.stdout)[ChannelHistoryPageResult.messages_key] + history = tuple(reversed(FieldCodec.decode(tuple[Message, ...], page))) assert len(history) == (1 if fresh else 2) assert history[-1].sender == "installer" assert history[-1].body.startswith("Install finished") From 48731273433621f1fc89fce15ebe9d3bf20bf9ff Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:54:42 -0400 Subject: [PATCH 07/10] Give session bridge buses to the runtime that serves them #2713 (c3bd96773) made SessionBridge.pause return one ResumableBridge; the tests still unpacked a tuple. Since #2709 (4cad001cc) a root belongs to one runtime: session_bridge now publishes its bus to the runtime serving the bridge, and the swap test hands it to the new runtime at publication, as the installer does. SourcePrograms reports its runtime. RootFile.__init_subclass__ annotates its keyword arguments. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/locked_store.py | 2 +- tests/source_programs.py | 7 +++++++ tests/test_install_session_bridge.py | 14 ++++++++++---- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index c5a69b4c5b..5968423578 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -59,7 +59,7 @@ class RootFile: filename: ClassVar[str] root_directory: ClassVar[str] = "." - def __init_subclass__(cls, **kwargs) -> None: + def __init_subclass__(cls, **kwargs: object) -> None: """A declared file is either retained across installations or rewritten by its running owner; a declaration that is neither has no install behaviour at all.""" super().__init_subclass__(**kwargs) diff --git a/tests/source_programs.py b/tests/source_programs.py index aeac66f78f..8950544292 100644 --- a/tests/source_programs.py +++ b/tests/source_programs.py @@ -10,6 +10,8 @@ from pathlib import Path from typing import Self +from agent_comms.launch_runtime import LaunchRuntime + @dataclass(frozen=True) class SourcePrograms: @@ -22,6 +24,11 @@ def bind(self, directory: Path) -> None: for package in self.packages: (directory / package.name).symlink_to(package, target_is_directory=True) + @property + def runtime(self) -> LaunchRuntime: + """The runtime these programs report: the interpreter this prefix declares.""" + return LaunchRuntime(self.prefix / 'bin' / 'python') + def script(self, function: Callable[..., object]) -> Path: target = f'{function.__module__}:{function.__name__}' (entry,) = (entry for entry in self.entries if entry.value == target) diff --git a/tests/test_install_session_bridge.py b/tests/test_install_session_bridge.py index 2bc854f706..d9263cefeb 100644 --- a/tests/test_install_session_bridge.py +++ b/tests/test_install_session_bridge.py @@ -23,6 +23,7 @@ from agent_comms.comms import wire from agent_comms.field_codec import FieldCodec from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.root_runtime import RootRuntime from agent_comms.thread_execution import ExternalThreadExecution from agent_comms_hooks.session import started from agent_comms_hooks.watch import LapsedWatch, LiveWatch, WatchClaim, WatchRecord @@ -32,9 +33,12 @@ @contextmanager -def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE, *, state_directory=None): +def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE, *, state_directory=None, + runtime=LaunchRuntime.running()): + """``runtime`` serves ``program`` and so owns the bus it attaches to.""" comms = wire(tmp_path / 'bus') Bus.install(Bus.path_in(comms.root)) + RootRuntime.publish(comms.root, runtime) comms.registry.declare(comms.registry.new_thread( 'bridge-copy', frozenset(), str(tmp_path), execution=ExternalThreadExecution)) transcript = tmp_path / 'copied-session.jsonl' @@ -140,7 +144,7 @@ def test_installer_pause_closes_the_real_bridge_and_eof_never_replays_or_relaunc (bridge,) = Serve.running(home) assert bridge.observed.process == observed.process with ExitStack() as resources: - (paused,) = bridge.pause(resources) + paused = bridge.pause(resources) assert paused.bridge == bridge assert child.poll() is None # The paused process keeps the claim; no session hook sees it lapse. @@ -165,7 +169,7 @@ def test_installer_discovers_the_live_claim_independent_of_state_directory( assert bridge.claim == claim assert bridge.observed.process == observed.process with ExitStack() as resources: - (paused,) = bridge.pause(resources) + paused = bridge.pause(resources) assert paused.bridge == bridge assert isinstance(claim.state(), LiveWatch) assert child.wait(timeout=15) == 0, log.read_text() @@ -202,11 +206,13 @@ def bridges_serving(state): def test_install_swap_keeps_the_claim_and_resumes_the_same_process_on_the_new_runtime( tmp_path, bridge_home, monkeypatch): old, new = (SourcePrograms.create(tmp_path / name) for name in ('old', 'new')) - with session_bridge(tmp_path, bridge_home, (str(old.script(claude.main)),)) as ( + with session_bridge(tmp_path, bridge_home, (str(old.script(claude.main)),), runtime=old.runtime) as ( home, claim, observed, child, log): current = home / LaunchRuntime.current_pointer().relative_to(Path.home()) current.parent.mkdir(parents=True) current.symlink_to(new.prefix.absolute(), target_is_directory=True) + # Publication hands the bus to the installed runtime, as the installer's does. + RootRuntime.publish(tmp_path / 'bus', new.runtime) monkeypatch.setenv('HOME', str(home)) assert observed.runtime.prefix == old.prefix.resolve() env = {**os.environ, 'HOME': str(home)} From 3aab54857cb5ca551eb60d3e49f1667536d5349a Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:58:03 -0400 Subject: [PATCH 08/10] Admit external reads of a representation through its wire structure PiReported decodes its value with the external reader, which tolerates keys Pi adds (contextUsage.percent), but _RepresentedDecoder admitted only the exact wire record, so the #2712 union prefilter ruled ReportedContextUsage out and PiOmitted refused the payload. The representation's external admit now derives from its structure's external admit. Also relocks the stack onto toad main, which carries toad #1149. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/field_codec.py | 9 +++++++++ stack/uv.lock | 2 +- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 9ea8ed0994..07b6e996e4 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -2184,6 +2184,15 @@ def admits_wire(self, data: object) -> bool: # Null handlers can change the structural null contract; leave that decision to decode. return data is None or self.wire_structure.admits_wire(data) + def admits_external(self, data: object) -> bool: + # The representation reads through its wire structure, so an external read admits + # whatever that structure's external reader admits. + return ( + any(self.external_decoders(data)) + or data is None + or self.wire_structure.admits_external(data) + ) + def _rewrite_records( self, data: object, change: WireRecordChange, select: WireRecordFilter, ) -> object: diff --git a/stack/uv.lock b/stack/uv.lock index 1d127d4ff7..c0abdbcb4b 100644 --- a/stack/uv.lock +++ b/stack/uv.lock @@ -243,7 +243,7 @@ wheels = [ [[package]] name = "batrachian-toad" version = "0.6.20" -source = { git = "https://github.com/OpenHCSDev/toad.git?branch=main#aeef59ee093561b7a1e5a87d1f33ff952e8343ef" } +source = { git = "https://github.com/OpenHCSDev/toad.git?branch=main#336cf3ec3074496aac2c0b783d31719ac0da3801" } dependencies = [ { name = "agent-client-protocol" }, { name = "agent-comms" }, From 6507a3e9ebf30a29858f774a1130e331153cbb38 Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 09:02:09 -0400 Subject: [PATCH 09/10] Name the codec's shape checks fits_* instead of the banned admits_* admit is banned vocabulary (debt_ratchet BannedVocabulary), and the new _RepresentedDecoder override added four more sites of it. The whole family in field_codec.py is renamed in this change, and the override now reuses the base check instead of restating its null test. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/field_codec.py | 70 ++++++++++++++++------------------ 1 file changed, 33 insertions(+), 37 deletions(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 07b6e996e4..b752237b7f 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -1010,7 +1010,7 @@ def argument_options(cls) -> dict[str, object]: class DeclaredWireType(ExternalJsonShape, WireValue): """A member whose whole wire value is one declared JSON type, converted by its own from_wire and to_wire. Its shape and schema derive from that type; inherited from a - union, its shape would be the union holding it, and admitting would never return.""" + union, its shape would be the union holding it, and the shape check would never return.""" wire_type: ClassVar[object] @@ -1949,16 +1949,16 @@ def encode(self, value: T) -> object: ... def wire_children(self) -> tuple[ValueDecoder[object], ...]: return () - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: """Check outer wire shape without decoding nested values. - This may admit a value decoding rejects, but must never reject one decoding accepts. + This may pass a value decoding rejects, but must never reject one decoding accepts. """ return True - def admits_external(self, data: object) -> bool: + def fits_external(self, data: object) -> bool: """A sibling-aware external reader may accept more than the stored wire shape.""" - return any(self.external_decoders(data)) or self.admits_wire(data) + return any(self.external_decoders(data)) or self.fits_wire(data) def selects_record(self, select: WireRecordFilter) -> bool: return False @@ -2110,7 +2110,7 @@ class _ExactScalarDecoder(_ScalarDecoder): def accepts(self, data: object) -> bool: return type(data) is self.target - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return self.accepts(data) def accepts_value(self, value: object) -> bool: @@ -2158,11 +2158,11 @@ class _RealDecoder(_NumberDecoder): __slots__ = () annotations = (float,) - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return type(data) in (int, float) def accepts(self, data: object) -> bool: - return self.admits_wire(data) and math.isfinite(cast(float, data)) + return self.fits_wire(data) and math.isfinite(cast(float, data)) @dataclass(frozen=True, slots=True) @@ -2180,18 +2180,14 @@ def wire_structure(self) -> ValueDecoder[object]: def wire_children(self) -> tuple[ValueDecoder[object], ...]: return (self.wire_structure,) - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: # Null handlers can change the structural null contract; leave that decision to decode. - return data is None or self.wire_structure.admits_wire(data) + return data is None or self.wire_structure.fits_wire(data) - def admits_external(self, data: object) -> bool: - # The representation reads through its wire structure, so an external read admits - # whatever that structure's external reader admits. - return ( - any(self.external_decoders(data)) - or data is None - or self.wire_structure.admits_external(data) - ) + def fits_external(self, data: object) -> bool: + # The representation reads through its wire structure, so an external read takes + # whatever that structure's external reader takes. + return ValueDecoder.fits_external(self, data) or self.wire_structure.fits_external(data) def _rewrite_records( self, data: object, change: WireRecordChange, select: WireRecordFilter, @@ -2302,7 +2298,7 @@ def accepts_value(self, value: object) -> bool: class _FamilyNameDecoder(_TypeDecoder, _ChoiceDecoder[Any]): - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return type(data) is str def encode(self, value: type[DeclaredFamily]) -> str: @@ -2330,8 +2326,8 @@ def from_annotation(cls, target, args): def accepts_value(self, value: object) -> bool: return any(type(value) is type(item) and value == item for item in self.values) - def admits_wire(self, data: object) -> bool: - return any(plan.admits_wire(data) for plan in self.value_plans) + def fits_wire(self, data: object) -> bool: + return any(plan.fits_wire(data) for plan in self.value_plans) def encode(self, value: object) -> object: for item, plan in zip(self.values, self.value_plans, strict=True): @@ -2357,8 +2353,8 @@ class UnionDecoder[T](_ContextDecoder[T]): def wire_children(self) -> tuple[ValueDecoder[object], ...]: return self.alternatives - def admits_wire(self, data: object) -> bool: - return any(alternative.admits_wire(data) for alternative in self.alternatives) + def fits_wire(self, data: object) -> bool: + return any(alternative.fits_wire(data) for alternative in self.alternatives) def _rewrite_records( self, data: object, change: WireRecordChange, select: WireRecordFilter, @@ -2394,7 +2390,7 @@ def decode_external(self, data: object, record: Mapping[str, object]) -> T: return self.decode_bound(data, record) return self.decode_resolved(self.decode_candidates( lambda alternative: alternative.decode_external(data, record), - admits=lambda alternative: alternative.admits_external(data), + fits=lambda alternative: alternative.fits_external(data), )) @property @@ -2416,17 +2412,17 @@ def encode(self, value: T) -> object: def decode_bound(self, data: object, context: Mapping[str, object]) -> T: return self.decode_candidates( lambda alternative: alternative.decode_bound(data, context), - admits=lambda alternative: alternative.admits_wire(data), + fits=lambda alternative: alternative.fits_wire(data), ) def decode_candidates( self, decode: Callable[[ValueDecoder[T]], T], *, - admits: Callable[[ValueDecoder[T]], bool], + fits: Callable[[ValueDecoder[T]], bool], ) -> T: first_error = None # Shapes only rule alternatives out; when they rule out every one, the first # alternative's own decode states the refusal. - candidates = tuple(filter(admits, self.alternatives)) or self.alternatives[:1] + candidates = tuple(filter(fits, self.alternatives)) or self.alternatives[:1] try: for alternative in candidates: try: @@ -2454,7 +2450,7 @@ class _ArrayDecoder(ValueDecoder[Any]): def wire_children(self) -> tuple[ValueDecoder[object], ...]: return self.items - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return isinstance(data, list) def _rewrite_records( @@ -2533,8 +2529,8 @@ class _VariadicTupleDecoder(_TupleDecoder): class _FixedTupleDecoder(_TupleDecoder): - def admits_wire(self, data: object) -> bool: - return super().admits_wire(data) and len(cast(list[object], data)) == len(self.items) + def fits_wire(self, data: object) -> bool: + return super().fits_wire(data) and len(cast(list[object], data)) == len(self.items) def plan_items(self, data: list[object]) -> Iterable[tuple[ValueDecoder[object], object]]: with MalformedRecord.refusing("Tuple length does not match declaration"): @@ -2635,7 +2631,7 @@ def decode_keys(self, data, keys, selections): class _StringKeyDictDecoder(_DictDecoder): entries = staticmethod(_object_entries) - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return isinstance(data, dict) @staticmethod @@ -2656,7 +2652,7 @@ def decode_document(self, data: object) -> dict[str, object]: class _PairKeyDictDecoder(_DictDecoder): entries = staticmethod(_object_or_pair_entries) - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return isinstance(data, (dict, list)) @staticmethod @@ -2674,7 +2670,7 @@ class _FamilyRecordDecoder[Member: DeclaredFamily](_ContextDecoder[Member]): target: type[Member] - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: return isinstance(data, dict) and isinstance(data.get(self.target.family_discriminator), str) @property @@ -2723,9 +2719,9 @@ def decode_bound(self, data: object, context: Mapping[str, object]) -> Member: class _LeafDocumentDecoder(_FamilyRecordDecoder): """A directly declared leaf document omits its family selector.""" - def admits_wire(self, data: object) -> bool: - return (super().admits_wire(data) - or _member_decoder(self.target).admits_wire(data)) + def fits_wire(self, data: object) -> bool: + return (super().fits_wire(data) + or _member_decoder(self.target).fits_wire(data)) def encode_document(self, value: DeclaredFamily) -> object: return _member_decoder(type(value)).encode(value) @@ -2746,7 +2742,7 @@ class _RecordDecoder(_ContextDecoder[Any]): target: type - def admits_wire(self, data: object) -> bool: + def fits_wire(self, data: object) -> bool: if not isinstance(data, dict): return False plan = FieldCodec._record_plan(self.target) From 708008e209bb0ea63f9e231721fc441d58a78167 Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 09:06:04 -0400 Subject: [PATCH 10/10] Declare the representation's wire shape check a boundary decoder It meets raw wire null from outside, which boundary_decoder declares; the rename touched its line, so NoneAbsence counted the old null test as new. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/field_codec.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index b752237b7f..ccf85d6216 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -2180,6 +2180,7 @@ def wire_structure(self) -> ValueDecoder[object]: def wire_children(self) -> tuple[ValueDecoder[object], ...]: return (self.wire_structure,) + @boundary_decoder def fits_wire(self, data: object) -> bool: # Null handlers can change the structural null contract; leave that decision to decode. return data is None or self.wire_structure.fits_wire(data)