Remove residual bus-file locks and repair shared boundary behavior

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.

src/agent_comms/collaboration_ledger.py

@@ -6,15 +6,14 @@
 from .errors import UnregisteredThreadError
 from .registration import Registration
 from .shared_ledger import SharedLedger
-from .store_files import _store_lock
 
 
 @dataclass(frozen=True, slots=True)
 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)

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,

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.

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, ...]

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(),
         )
     )
 

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]

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))

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

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)