Virtual bus batch: W, B1, B2, C, D and E integrated

src/agent_comms/acp.py

@@ -38,10 +38,7 @@
     TextContentBlock,
 )
 
-from agent_comms.coordination_errors import (
-    IdentityConflict,
-    PublicationActivationBlocked,
-)
+from agent_comms.coordination_errors import IdentityConflict
 from agent_comms.coordinator import Coordination
 
 from .acp_extension import (
@@ -53,6 +50,7 @@
 from .comms import Comms, wire
 from .coordination_cohort import next_sealed_assignment
 from .coordination_database import CoordinationStore
+from .cursor_owner import CursorOwner
 from .cursor_publication import CursorPublication
 from .field_codec import FieldCodec, MalformedRecord
 from .image_inputs import prompt_images
@@ -75,7 +73,6 @@
 from .turn_effects import TurnEffects
 from .turn_runner import TurnRunner
 from .turn_terminal import EndedWithoutResult, TurnInterruption, TurnTerminal
-from .wire_log import WireLog
 
 if TYPE_CHECKING:
     from .selected_tool_broker import SelectedToolIntent
@@ -214,15 +211,17 @@ async def _selected_write_request(self, session_id: str, request) -> PromptRespo
         from .selected_write_plan import SelectedWritePlans
 
         owner = self.sessions.require(session_id)
-        root_id = self._private_nk_marker()
         controller = self._runtime.controller.get()
         if controller is None or (
             controller is UNBOUND_CONTROLLER and self.sessions.client is None
         ):
             raise RequestError.invalid_params(
                 {"reason": "Selected write requires attached ACP controller"}
             )
-        acknowledgement = await Coordination.run_worker(partial(SelectedWritePlans(self._comms, root_id).submit,
+        scope = await Coordination.run_worker(partial(self.cursors.scope, owner, session_id))
+        plans = SelectedWritePlans(self._comms, scope.require_live().wire_root_id)
+        acknowledgement = await Coordination.run_worker(partial(
+            plans.submit,
             owner_name=owner,
             source_seq=request.source_seq,
             source_message_id=request.source_message_id,
@@ -319,21 +318,6 @@ async def authenticate(self, method_id: str, **kwargs: Any) -> None:
 
     # ─── Helpers ─────────────────────────────────────────────────────────────
 
-    @property
-    def configured_root_id(self) -> str:
-        """The bus root ID the launch pinned; the bus reader checks it against the marker."""
-        return self._comms.owners.require_private_launch().wire_root_id
-
-    def _private_nk_marker(self) -> str:
-        """Check that the bus belongs to the configured root before a request runs."""
-        configured_root_id = self.configured_root_id
-        with self._comms.bus.log.checked_read() as source:
-            if source.marker.root_id != configured_root_id:
-                raise PublicationActivationBlocked(
-                    "private N/K ACP root does not match configuration"
-                )
-            return source.marker.root_id
-
     async def _drain_private_nk(self, session_id: str, wire_root_id: str) -> int:
         """Run a selected private wake for this ACP session.
 
@@ -348,8 +332,6 @@ async def _drain_private_nk(self, session_id: str, wire_root_id: str) -> int:
 
         launch = self._comms.owners.require_private_launch()
         package = launch.native_package
-        if launch.wire_root_id != wire_root_id:
-            raise PublicationActivationBlocked("private N/K ACP root does not match its launch")
         thread_name = await self.sessions.sync_identity(session_id)
         snapshot = await Coordination.run_worker(self._comms.registry.snapshot)
         owner = snapshot.require(thread_name)
@@ -361,11 +343,14 @@ async def _drain_private_nk(self, session_id: str, wire_root_id: str) -> int:
             return 0
         if not owner.process_identity.runs_here():
             raise IdentityConflict("private N/K ACP recipient is not this process owner")
-        bus = MessageBus(WireLog.path_in(self._comms.root), self._comms.registry, ResponseWrites)
-        def read_claim():
-            with bus.log.checked_read() as source:
-                return source.marker.claimed_after_seq
-        claimed_after_seq = await Coordination.run_worker(read_claim)
+        lookup = owner.incarnation.require_recorded().lookup
+        bus = MessageBus(self._comms.root, self._comms.registry, ResponseWrites)
+        delivery = self.cursors.delivery(session_id)
+        async with delivery.reading() as source_reads:
+            state = await CursorOwner.read_async(
+                bus, wire_root_id=wire_root_id, owner_name=thread_name, source_reads=source_reads,
+            )
+        after_seq = state.covered_seq
         store_path = CoordinationStore.path_in(self._comms.root)
         # One execution owns this wake's package acquisition: acceptance of new
         # originals and the selected launch share it instead of verifying twice.
@@ -374,34 +359,30 @@ def read_claim():
             wire_root_id=wire_root_id,
             owner_name=thread_name,
             native_package=package,
-            cursor_delivery=self.cursors.delivery(session_id),
+            cursor_delivery=delivery,
             write_authority=AcpSelectedWriteAuthority(
                 self, session_id, SelectedWritePlans(self._comms, wire_root_id)
             ),
-            **(
-                {"selected_tool_intent": self._private_selected_tool_intent}
-                if self._private_selected_tool_intent is not None
-                else {}
-            ),
+            selected_tool_intent=self._private_selected_tool_intent,
         )
         with execution.cursor_delivery.measurements.operation("accept_deliveries"):
             await _accept_visible_deliveries(
-                bus, wire_root_id, store_path, owner.incarnation.lookup, 0,
+                bus, wire_root_id, store_path, lookup, after_seq,
                 owner_name=owner.name, acquire_package=execution.validate,
             )
         if self.inputs.owns_execution(session_id, owner):
             # A running turn takes its owner's pending messages as inputs;
             # whatever it cannot take wakes this owner once the turn ends.
             await DeliveredInput.deliver_pending(
-                self.inputs, session_id, owner, bus, wire_root_id, claimed_after_seq
+                self.inputs, session_id, owner, bus, wire_root_id, after_seq
             )
             await self.cursors.publish(session_id, thread_name)
             return 0
         def select(store):
             with store.session.read():
-                participant = store.participants.get(owner.incarnation.lookup)
+                participant = store.participants.get(lookup)
                 candidate = next_sealed_assignment(
-                    store, participant.lookup, owner.name, after_seq=claimed_after_seq,
+                    store, participant.lookup, owner.name, after_seq=after_seq,
                 )
                 if candidate is None:
                     return False

src/agent_comms/acp_extension.py

@@ -97,16 +97,6 @@ class TranscriptChangedUpdate(AgentCommsUpdate):
     cursor: TranscriptCursor | None
 
 
-@dataclass(frozen=True)
-class UnknownDeliveryFailure(BackendDeliveryFailure):
-    wire_root_id: str
-    wire_seq: int
-    message_id: str
-
-    def present(self, receiver, body: str) -> None:
-        receiver.delivery_unknown(self, body)
-
-
 class ClaimBlockedFailure(BackendDeliveryFailure):
     def present(self, receiver, body: str) -> None:
         receiver.delivery_blocked(self.description, body)

src/agent_comms/active_route.py

@@ -18,16 +18,15 @@
 from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Annotated, ClassVar
+from typing import Annotated, ClassVar
 
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
-from .field_codec import FieldCodec, Id128Text, MalformedRecord, PathText
+from .field_codec import FieldCodec, MalformedRecord, PathText
 from .launch_root import LaunchRoot
 from .private_nk_entrypoint import PrivateNkLaunch, RootLaunch
-
-if TYPE_CHECKING:
-    from .wire_log import WireLog
+from .private_path import ObservedPath
+from .recorded_bus import RecordedBusIdentity
 
 
 @dataclass(frozen=True, slots=True)
@@ -99,9 +98,9 @@ def encode(cls, value: object) -> str:
 class ActiveRoute(CommsRoute):
     """The default private root; each installation launches its own pinned native package."""
     root: Annotated[Path, AbsoluteRoutePathText]
-    wire_root_id: Annotated[str, Id128Text]
+    bus_identity: RecordedBusIdentity
     # The file format: the writer stamps it and the reader refuses any other before decoding.
-    version: ClassVar[int] = 2
+    version: ClassVar[int] = 3
     # The CLI command that replaces a file this reader refuses; it registers under this name.
     repair_command: ClassVar[str] = "replace-route"
 
@@ -139,27 +138,20 @@ def from_record(cls, value: object, path: Path) -> ActiveRoute:
             raise MalformedRecord(f"active comms route {path} is invalid: {error}") from error
         return route
 
-    def wire_log(self) -> WireLog:
-        from .wire_log import WireLog  # the wire log's module imports this one
-
-        return WireLog(WireLog.path_in(self.root))
-
     def selects_its_root(self) -> bool:
-        """Whether the root still exists and carries the marker this route was published for."""
-        return self.root.is_dir() and (
-            self.wire_log().read_metadata_unlocked().wire_root_id == self.wire_root_id)
+        """A route selects the database observed when it was published."""
+        from .bus import Bus
+
+        return self.root.is_dir() and self.bus_identity.matches(
+            ObservedPath.at(Bus.path_in(self.root)))
 
     def observe_root(self) -> Path:
-        # The writer atomically replaces this marker. Reuse its owner/mode,
-        # ancestry and protocol decoder, without taking the mutation/durability
-        # barrier or interpreting registry/history merely to observe identity.
-        marker = self.wire_log().private_marker_unlocked()
-        if marker.root_id != self.wire_root_id:
-            raise RelationViolationError("private route root ID changed")
+        if not self.selects_its_root():
+            raise RelationViolationError("route bus was replaced")
         return super(ActiveRoute, self).observe_root()
 
     def launch(self) -> PrivateNkLaunch:
-        return PrivateNkLaunch.installed(self.root, self.wire_root_id)
+        return RootLaunch.derive(self).require_private()
 
 
 def resolve_comms_route(root: CommsRoute | Path | str | None = None) -> CommsRoute: