diff --git a/plans/VIRTUAL-BUS-PLAN.md b/plans/VIRTUAL-BUS-PLAN.md index 54286fdc42..5cde521fca 100644 --- a/plans/VIRTUAL-BUS-PLAN.md +++ b/plans/VIRTUAL-BUS-PLAN.md @@ -88,10 +88,41 @@ Everything lands as one batch through the merge queue, drop-on-fail. Then one in Each phase, its predecessors and the evidence on `origin/main` that proves it done are declared in `src/agent_comms/plan_phases.py`; the commander's plan-idle rule and order refusals read that family, and accepting a task never marks a phase done. -- `git grep -n "offset\|max_bytes\|bus.jsonl\|bus_meta\|_wire_lock_path\|has_private_wire_fields\|/ \"wire\"" src` returns no bus hits (the last pattern catches the hand-spelled `root / "wire"` lock path at about 20 sites). +- `BusDeletionPhase.evidence` finds the five deleted modules absent and no known physical bus symbols, filename literals or lock-path expressions in Python source. Its AST check reads executable uses, including `/ "wire"`, `.with_name("wire")` and `.joinpath("wire")`, without matching prose or its own pattern declaration. This is a known-syntax screen; aliases and dynamically constructed paths still require source review. Remaining byte offsets and budgets are judged by their actual owner, since external native transcripts stay files. - The deleted modules are gone. - A process killed mid-publish leaves the whole message or none of it. -- A reader that meets a held bus lock waits for it and never ends a turn (step C). Today `NativeSourceCursor._advance_proven` (`native_source_cursor.py:83`) takes the boundary with `blocking=False`, and with no contention allowance when no input is committed (`:79-80`). So any concurrent writer ends the turn with `BlockingIOError`: three times on core-5d8da75aa between 12:20 and 12:39 on 2026-10-21, with no install running. +- Concurrent bus publication must not abort native cursor settlement. The current path is `CoordinatedTurn.cursor_status_for` → `CursorDelivery.advance` → `CursorOwner.advance_async` → `CursorOwner._advance_owned`. Bus reads use `Bus.observing` and SQLite snapshots; the cursor owner commits through the coordinator session's existing write transaction. `Coordination.run_async` runs this work through its worker with the default `StoreWait` contention policy. Qualify that path under competing writers within its bounded wait; no physical bus lock is reintroduced. - Every history converts with every record round-tripping. - On every machine: Toad pages history, delivery wakes units, claims hold, and a real message to a live agent is answered. - Lines deleted are reported first; the census confirms. + +Historical regression evidence: on core-5d8da75aa, the former +`NativeSourceCursor._advance_proven` in `native_source_cursor.py` took its +boundary with `blocking=False`, without contention allowance when no input +was committed. Concurrent writers ended turns with `BlockingIOError` three +times between 12:20 and 12:39 on 2026-10-21, with no install running. That +class and file have since been deleted; the incident motivates the current +contention obligation above, rather than describing a current source site. + +## Deletion checkpoint from the current integration branch + +At `acb2ddd0e`, all five old modules are already absent. Three physical bus locks +remain: `NativeReplyPreparation.finished`, `SendRefusal.record`, and +`CollaborationLedger.merge`. The last uses `.with_name("wire")`, which the old +grep misses. The old grep also matches its own `bus_meta` pattern declaration, +so its result does not establish whether physical bus uses remain. + +The deletion branch removes those locks through the existing owners. Unread +publication keeps its local preparation lock, source revision check and SQLite +write transaction; a busy SQLite writer leaves completed readers pending. +Refusal recording keeps maintenance permission under the registry lock until +its SQL transaction finishes. Ledger merge keeps the registered-author snapshot +through the existing document update, in the declared registry-before-ledger +lock order. None creates a replacement bus lock. + +The phase check reads committed source through `MainTree`. Its controls include +the actual remaining path spellings, physical filenames and removed symbols, +alongside comments and docstrings that describe deletion. This source checkpoint +does not mark the phase complete on main or qualify an installed runtime. The +merger still checks the affected copied native/Toad journey, including a fresh +DM, its actual reply and tab return. diff --git a/src/agent_comms/collaboration_ledger.py b/src/agent_comms/collaboration_ledger.py index 66c2dc4913..c5686169e0 100644 --- a/src/agent_comms/collaboration_ledger.py +++ b/src/agent_comms/collaboration_ledger.py @@ -6,7 +6,6 @@ from .errors import UnregisteredThreadError from .registration import Registration from .shared_ledger import SharedLedger -from .store_files import _store_lock @dataclass(frozen=True, slots=True) @@ -14,7 +13,7 @@ class CollaborationLedger(SharedLedger): registry: Registration def merge(self, updates: Mapping[str, object], author: str) -> None: - with _store_lock(self.path.with_name("wire")): - if author not in self.registry: + with self.registry.store.reading_snapshot() as registry: + if registry.canonical_name(author) not in registry.threads: raise UnregisteredThreadError(f"Author {author!r} is not registered.") super().merge(updates, author) diff --git a/src/agent_comms/coordination_tables/refusals.py b/src/agent_comms/coordination_tables/refusals.py index 8298ad08c4..2324a781c8 100644 --- a/src/agent_comms/coordination_tables/refusals.py +++ b/src/agent_comms/coordination_tables/refusals.py @@ -14,7 +14,6 @@ from agent_comms.coordination_session import CoordinationSession from agent_comms.errors import RelationViolationError, ToolRefusedError from agent_comms.maintenance_barrier import MaintenanceBarrier -from agent_comms.store_files import SharedLock, _store_lock from agent_comms.typed_table import Column, DurableTables, Index, SharedFileTables, TypedTable from ..locked_store import RegistryFile @@ -45,8 +44,8 @@ def install(cls, store: Coordination) -> None: @classmethod def record(cls, root: Path, sender: str, target: str, error: ToolRefusedError) -> None: - with guard_original_root_write(root), _store_lock(root / "wire", mode=SharedLock()): - MaintenanceBarrier(RegistryFile.path_in(root)).assert_open_unlocked() + with (guard_original_root_write(root), + MaintenanceBarrier(RegistryFile.path_in(root)).allow_input()): with ( CoordinationSession(str(CoordinationStore.path_in(root))) as store, store.writing() as db, diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 1876365825..466b9b8c80 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -873,14 +873,22 @@ def decode_value(cls, data: object, structural: ValueDecoder[object]) -> object: def from_wire(cls, data: object) -> object: ... -class SingleFieldValue(ExternalJsonShape, WireValue): - """One declared field is the whole wire value, not an object envelope.""" +class SingleValueEncoding(WireValue): + """Encode one declared field as the whole wire value.""" @classmethod def wire_field(cls): entry, = FieldCodec._record_plan(cls).fields return entry + def to_wire(self): + entry = self.wire_field() + return FieldCodec.encode_document(getattr(self, entry.name), entry.annotation) + + +class SingleFieldValue(ExternalJsonShape, SingleValueEncoding): + """One declared field is the whole wire value, not an object envelope.""" + @classmethod def from_wire(cls, data): entry = cls.wire_field() @@ -894,10 +902,6 @@ def schema(cls): def argument_options(cls) -> dict[str, object]: return compile_decoder(cls.wire_field().annotation).argument_options() - def to_wire(self): - entry = self.wire_field() - return FieldCodec.encode_document(getattr(self, entry.name), entry.annotation) - class SubclassUnion(DeclaredFamily, AcceptsNull, ExternalJsonShape, WireValue): """An untagged wire family; each union declares its own root with affix="".""" @@ -983,7 +987,7 @@ def from_json_value(cls, value): return cls(value=value) -class JsonShapeFamily(AcceptsNull, ExternalJsonShape, WireValue): +class JsonShapeFamily(AcceptsNull, ExternalJsonShape, SingleValueEncoding): """Untagged external JSON enters one declaration-owned value family. Put DeclaredFamily before this capability in the root's bases, preserving @@ -1045,9 +1049,6 @@ def schema(cls) -> dict[str, Any]: for _, annotation in cls._json_owners.values()] return FieldCodec.alternatives_schema("anyOf", alternatives) - def to_wire(self): - return self.value - class TextRepresentation[V](FieldRepresentation): """External text scalars share strict decoding at the canonical boundary. diff --git a/src/agent_comms/phase_evidence.py b/src/agent_comms/phase_evidence.py index 18289353ab..461e7f1136 100644 --- a/src/agent_comms/phase_evidence.py +++ b/src/agent_comms/phase_evidence.py @@ -10,6 +10,7 @@ import ast import copy +import re import subprocess from abc import ABC, abstractmethod from collections import Counter @@ -261,6 +262,67 @@ def describe(self) -> str: return f"{self.pattern!r} in {' '.join(self.paths)} on main" +class BusStorageAbsent(MainEvidence, ast.NodeVisitor): + """Read executable bus references, not text describing their deletion.""" + + symbols = re.compile(r"WireLog|wire_log|_wire_lock_path|has_private_wire_fields") + filenames = re.compile(r"bus[.]jsonl|bus_meta[.]json") + found = False + + def read(self, main: MainTree) -> bool: + scan = type(self)() + for tree in main.modules(("src",)): + scan.visit(tree) + return not scan.found + + def describe(self) -> str: + return "no known physical bus symbols, file literals or lock paths in Python source on main" + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self.found |= bool(self.symbols.fullmatch(node.name)) + self.generic_visit(node) + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + self.found |= bool(self.symbols.fullmatch(node.name)) + self.generic_visit(node) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + self.found |= bool(self.symbols.fullmatch((node.module or "").rpartition(".")[2])) + self.generic_visit(node) + + def visit_Name(self, node: ast.Name) -> None: + self.found |= bool(self.symbols.fullmatch(node.id)) + + def visit_Attribute(self, node: ast.Attribute) -> None: + self.found |= bool(self.symbols.fullmatch(node.attr)) + self.generic_visit(node) + + def visit_alias(self, node: ast.alias) -> None: + self.found |= bool(self.symbols.fullmatch(node.name.rpartition(".")[2])) + + def visit_Constant(self, node: ast.Constant) -> None: + if isinstance(node.value, str): + self.found |= bool(self.filenames.fullmatch(node.value)) + + def visit_Expr(self, node: ast.Expr) -> None: + if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): + return # A docstring or inert prose cannot open a physical bus. + self.generic_visit(node) + + def visit_BinOp(self, node: ast.BinOp) -> None: + match node: + case ast.BinOp(op=ast.Div(), right=ast.Constant(value="wire")): + self.found = True + self.generic_visit(node) + + def visit_Call(self, node: ast.Call) -> None: + match node: + case ast.Call(func=ast.Attribute(attr="with_name" | "joinpath"), + args=[*_, ast.Constant(value="wire")]): + self.found = True + self.generic_visit(node) + + @dataclass(frozen=True) class AllEvidence(PhaseEvidence): parts: tuple[PhaseEvidence, ...] diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 1c49d3b01b..5d1ed09e36 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -14,6 +14,7 @@ from agent_comms.declared_family import DeclaredFamily from agent_comms.phase_evidence import ( AllEvidence, + BusStorageAbsent, ClassFlagsAbsent, CommitMerged, NoAbsenceProbes, @@ -322,9 +323,7 @@ class BusDeletionPhase(VirtualBusPhase): "src/agent_comms/wire_record.py", ) ), - PatternAbsent( - r"bus\.jsonl|bus_meta|_wire_lock_path|has_private_wire_fields|/ \"wire\"" - ), + BusStorageAbsent(), ) ) diff --git a/src/agent_comms/view_unread.py b/src/agent_comms/view_unread.py index 45dcf3112e..01838fe908 100644 --- a/src/agent_comms/view_unread.py +++ b/src/agent_comms/view_unread.py @@ -22,7 +22,6 @@ from .private_path import FileObservation, FileRevision, MissingFile from .read_ledger import ReadDocument, ReadLedger from .registry_document import RosterVisibility -from .store_files import SharedLock, _store_lock from .thread_owned_state import ThreadOwnedState from .typed_table import Column, JsonStorage, SQLiteTableName, SQLiteUserVersion, TypedRow, TypedTable from .ui_model.unread import CappedReplyUnread, IndexingUnread, ReplyUnread, UnreadPresentation @@ -341,16 +340,15 @@ def finished( completed = tuple(source for source, reader in self.readers.items() if reader.future.done()) if not completed: return self - # Publication stays under the existing deletion and SQLite owners. + # The state lock owns local retirement; SQLite owns publication exclusion. # A busy writer leaves these exact completed readers for the next observation. try: - with _store_lock(state.path.parent / "wire", mode=SharedLock(), blocking=False): - with (CoordinationReadUnavailable.checking(), - state.publishing_changes(database, state.index_path), - SQLiteFile.write_transaction(database)): - for source in completed: - self.readers[source].commit(database) - except (BlockingIOError, CoordinationReadUnavailable): + with (CoordinationReadUnavailable.checking(), + state.publishing_changes(database, state.index_path), + SQLiteFile.write_transaction(database)): + for source in completed: + self.readers[source].commit(database) + except CoordinationReadUnavailable: return self for source in completed: del self.readers[source] diff --git a/src/agent_comms/watched.py b/src/agent_comms/watched.py index 3a60f9cc02..11ccf42595 100644 --- a/src/agent_comms/watched.py +++ b/src/agent_comms/watched.py @@ -256,7 +256,7 @@ def wake(self) -> None: def _on_ready(self) -> None: try: self._drain() - except Exception as error: + except (OSError, ValueError) as error: # Callback failures must reach the waiting reader, not just stderr. self.loop.remove_reader(self.fd) self._publish(SourceFailed(error)) diff --git a/tests/test_held_sends.py b/tests/test_held_sends.py index e6b31a90bd..7b846f22f0 100644 --- a/tests/test_held_sends.py +++ b/tests/test_held_sends.py @@ -3,16 +3,51 @@ from agent_comms.bus import Bus import asyncio import os +import sqlite3 import sys +import time +from contextlib import closing from pathlib import Path import pytest from agent_comms.comms import Comms from agent_comms.coordination_tables.refusals import SendRefusal +from agent_comms.coordination_database import CoordinationStore +from agent_comms.errors import ToolRefusedError from agent_comms.maintenance_barrier import MaintenanceOperator +from agent_comms.store_files import _store_lock from agent_comms.tool_caller import ThreadCaller from agent_comms.tools import invoke_tool +from test_locked_store import event, process + + +def record_refusal(root, started): + started.set() + SendRefusal.record(root, "sender", "recipient", ToolRefusedError("retained refusal")) + + +def test_refusal_writer_keeps_registry_permission_until_its_sql_write_finishes(private_comms): + root = private_comms.root + started = event() + with closing(sqlite3.connect(CoordinationStore.path_in(root), isolation_level=None)) as writer: + writer.execute("BEGIN IMMEDIATE") + with process(record_refusal, root, started): + try: + assert started.wait(5) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + try: + with _store_lock(private_comms.registry.store.path, blocking=False): + pass + except BlockingIOError: + break + time.sleep(.01) + else: + pytest.fail("The blocked refusal writer released its registry permission") + finally: + writer.execute("ROLLBACK") + assert tuple(row.reason for row in SendRefusal.recent(root)) == ("retained refusal",) @pytest.fixture diff --git a/tests/test_json_shape_field_codec.py b/tests/test_json_shape_field_codec.py index ba4163b6fb..6b491d2fde 100644 --- a/tests/test_json_shape_field_codec.py +++ b/tests/test_json_shape_field_codec.py @@ -2,11 +2,14 @@ from abc import abstractmethod from dataclasses import dataclass +import json import pytest from agent_comms.declared_family import DeclaredFamily -from agent_comms.field_codec import FieldCodec, JsonShapeFamily, JsonShapeMember, MalformedRecord +from agent_comms.field_codec import ( + FieldCodec, JsonShapeFamily, JsonShapeMember, MalformedRecord, SingleFieldValue, +) from agent_comms.launch_runtime import ImportPath, NoImportPath, SpecifiedImportPath from agent_comms.native_claim_epoch import NativeClaimEpoch, RecordedNativeClaim, UnrecordedNativeClaim from agent_comms.selected_triage import ( @@ -79,6 +82,8 @@ def test_external_shape_family_uses_one_codec_for_children_and_new_cases(): assert isinstance(value.value["details"].value[2], BooleanValue) assert isinstance(value.value["details"].value[3], NullValue) assert value.render() == (("failure", ("400",), "False", ""),) + assert value.to_wire() == raw + assert json.loads(json.dumps(value.to_wire())) == raw assert FieldCodec.encode(value) == raw assert FieldCodec.project(value, "status") == raw assert ExternalValue.member_named("text") is TextValue @@ -100,6 +105,19 @@ def render(self): FieldCodec.decode(ExternalValue, float("nan")) +def test_one_field_encoding_preserves_distinct_cli_argument_contracts(): + @dataclass(frozen=True) + class ScalarArgument(SingleFieldValue): + value: int + + assert ScalarArgument(5).to_wire() == 5 + assert ScalarArgument.from_wire(5) == ScalarArgument(5) + assert ScalarArgument.argument_options() == {"type": int} + assert ExternalValue.argument_options() == {} + assert ImportPath.argument_options() == {} + assert NativeClaimEpoch.argument_options() == {} + + def test_semantic_members_stay_out_of_shape_selection(): class Meaning(DeclaredFamily, JsonShapeFamily, affix="Meaning"): pass @@ -124,12 +142,13 @@ def from_json_value(cls, value): (NativeClaimEpoch, None, UnrecordedNativeClaim(), "5"), (NativeClaimEpoch, 5, RecordedNativeClaim(5), "5"), (TriageDecisionRecord, None, AbsentTriageDecisionRecord(), 1), - (TriageDecisionRecord, "IGNORE", DecidedTriageRecord(IgnoreSelectedTriage), 1), + (TriageDecisionRecord, "ignore", DecidedTriageRecord(IgnoreSelectedTriage), "IGNORE"), (ImportPath, None, NoImportPath(), 1), (ImportPath, "/opt/lib", SpecifiedImportPath("/opt/lib"), 1), ]) def test_shape_family_decodes_each_member_and_refuses_an_unowned_type(family, wire, expected, refused): assert FieldCodec.decode(family, wire) == expected + assert expected.to_wire() == wire with pytest.raises(MalformedRecord): FieldCodec.decode(family, refused) diff --git a/tests/test_phase_evidence.py b/tests/test_phase_evidence.py index 480f9c2826..172d921176 100644 --- a/tests/test_phase_evidence.py +++ b/tests/test_phase_evidence.py @@ -27,7 +27,7 @@ UnreadableEvidenceError, ) from agent_comms.pinned_audit import GodFunction -from agent_comms.plan_phases import FLAG, ExternalBoundariesPhase, PlanPhase +from agent_comms.plan_phases import BusDeletionPhase, FLAG, ExternalBoundariesPhase, PlanPhase from agent_comms.repo import Repo from agent_comms.repository_name import RepositoryName @@ -100,6 +100,35 @@ def main_of(tmp_path, files): Repo(checkout=tmp_path, github=RepositoryName.from_wire("OpenHCSDev/agent-comms")))) +def test_bus_deletion_evidence_reads_uses_without_matching_its_own_declaration(tmp_path): + reading = main_of(tmp_path, { + "src/phase.py": "from agent_comms.phase_evidence import BusStorageAbsent\n" + "evidence = BusStorageAbsent()\n" + "# Delete bus.jsonl and bus_meta once the carry is accepted.\n" + "def documentation():\n 'bus.jsonl'\n", + }) + assert reading.holds(BusDeletionPhase.evidence) + # Every production spelling found during the owner trace must make the phase fail. + for source in ( + "class WireLog:\n pass\n", + "def has_private_wire_fields(value):\n return False\n", + "from somewhere.wire_log import WireLog\n", + "from somewhere.wire_log import another_name\n", + "def use(owner):\n return owner._wire_lock_path\n", + "def use(owner):\n return owner / 'wire'\n", + "def use(owner):\n return owner.with_name('wire')\n", + "def use(owner):\n return owner.joinpath('wire')\n", + "def use(owner):\n return owner / 'bus.jsonl'\n", + "def use(owner):\n return owner / 'bus_meta.json'\n", + ): + (tmp_path / "src/uses.py").write_text(source) + git(tmp_path, "add", ".") + git(tmp_path, "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "use") + git(tmp_path, "update-ref", "refs/remotes/origin/main", git(tmp_path, "rev-parse", "HEAD")) + changed = PlanReading((), MainTree.of(reading.main.repo)) + assert not changed.holds(BusDeletionPhase.evidence), source + + def test_source_shape_evidence_reads_declarations_on_main(tmp_path): copy = " def __init__(self, path):\n self.path = path\n" reading = main_of(tmp_path, { diff --git a/tests/test_watched.py b/tests/test_watched.py index f4efc575bc..551fe14c57 100644 --- a/tests/test_watched.py +++ b/tests/test_watched.py @@ -1,6 +1,7 @@ """The declared bus watch uses real native commit and replacement notifications.""" import asyncio +import errno from contextlib import aclosing from concurrent.futures import Future from dataclasses import replace @@ -173,6 +174,20 @@ async def test_native_file_removal_wakes_without_unrelated_events(tmp_path, remo await asyncio.wait_for(changes.wait(), 1) +async def test_failed_native_descriptor_reaches_waiting_reader_and_survives_later_wake(tmp_path): + path = tmp_path / "observed" + async with ChangeSignal.opened(FileSource(path)) as changes: + changes.prepare() + # Retire the real native descriptor before a queued readiness callback runs. + # ExitStack close is idempotent; context retirement will not close it twice. + changes.close() + changes._on_ready() + changes.wake() + with pytest.raises(OSError) as failed: + await asyncio.wait_for(changes.wait(), 1) + assert failed.value.errno == errno.EBADF + + async def test_native_file_burst_wakes_before_the_producer_finishes(tmp_path): path = tmp_path / "stream" path.write_text("0")