Integrate SQL bus, independent requests and installer runtime ownership

plans/HARNESS-PLAN.md

@@ -2,6 +2,30 @@
 
 Decision (Tristan, 2026-10-21 22:45): the commander fixes the harness it commands with. It works with full authority on this plan only: it makes each change, inspects it with full attention, runs only the tests the change touches, and pushes straight to `main`. Audited against `origin/main` at `39fdfe810`.
 
+## Expanded installer integration, 2026-10-22
+
+Tristan requested the next paired installation include the useful remaining open
+work. Core's virtual-bus checkpoint passed the full configured-root rehearsal:
+19,182 records, 194 threads, preserved read ledger, 208.38 seconds. That is a
+rehearsal result, not an installed checkpoint.
+
+Root owns porting the remaining reporting and bridge-lifetime work from #2556
+and #2527 into the current `ChangedInstall` and `PreparedMigration` family.
+The old branches' detached migration paths stay deleted. Reports before carry
+must execute in the outgoing runtime; reports after accepted carry use the
+incoming runtime. Rehearsals publish nothing. The report lifetime includes
+target-tree cleanup, so cleanup failure cannot leave a successful terminal post.
+Bridge outcomes belong to the actual captured bridge and survive failed exits.
+
+The reused finished checkout is `/home/ts/wt/harness-plan`, branch
+`integration/installer-report-lifetime`. Before-source evidence is
+`/home/ts/.cache/agent-scratch/root-installer-report-before.json`: NRA parsed
+953 production/test/tool modules and retained 5,341 original class declarations;
+5,151 project into the conservative family index. The remaining 190 original
+declarations stay visible rather than being counted as resolved. No parse
+omissions; source evidence does not prove behavior. Implementation and affected
+CLI/application qualification remain open.
+
 ## Why
 
 On 2026-10-21 the plans Tristan asked for at 18:30 the day before had not landed 28 hours later. Every error pulled the fleet off them, and the errors came from the harness itself:
@@ -194,15 +218,26 @@ Neither attempt establishes the recorded SelectCurrent diagnosis or a deadlock.
 The persistence cause is traced: Toad's `Session` declared SQL directives in
 dataclass metadata after Core's column compiler moved to `Annotated`. Fresh
 session rows therefore had a NULL `id` while insertion returned SQLite's rowid.
-The Toad binding fork owns the complete DB family repair: annotation-derived
-session and model-history tables, shared durable schema reconciliation, one
-transaction path and surfaced SQLite failures. Eight focused checks passed on
-the source checkpoint; final connection-owner and dependency cleanup follows.
-The installed user's sessions table already has a valid primary key and no rows;
-its model-history table has two rows requiring a one-time schema carry.
-That carry uses an explicit SQLite backup and transaction outside the bus-only
-installer carry contract. Root owns its execution and paired installation after
-the Toad fork's copy proof; the virtual-bus integration is not a prerequisite.
+The DB family repair is installed: Core `d9ce9cd95` with Toad `a5c627b95`.
+Session and model-history rows share annotation-derived schema and the existing
+SQLite connection/transaction owner; SQLite errors propagate. The unused second
+connection path, success-only bools and aiosqlite dependency are deleted. The
+one-time SQLite carry preserved both model-history rows and the existing sessions;
+`/home/ts/.cache/agent-scratch/toad-db-before-install.db` retains the backup.
+The accepted transient converter is retired. All 194 registered threads stayed
+unchanged, with no owner restarts. Actual installed saved-fork/fresh-DM/reply/tab
+return and complete native client-request journeys passed: 2 checks in 54.95 seconds,
+recorded in `installer-db-batch-native.txt`. This includes reading the newly
+created session's persisted title through the installed DB.
+
+The Toad binding fork separately owns the confirmed new-session ordering defect:
+the server reuses load's replay-before-response behavior before the client knows
+the new ID. Retained wire records 43–51 in `toad-test-ich92jvs` prove seven new-ID updates
+preceded their creation response. The repair belongs to the existing subscription
+request family; a fresh subscription announces identity before replay, while load
+keeps its required replay-before-response contract. Source implementation and an
+actual stdio ordering check are in progress; the passing DB journeys do not cover
+this packet-order assertion.
 
 ### 2. Commander verbs take a fixed, small time
 
@@ -262,3 +297,27 @@ The growth alarm is commander-native phase 7's (`plans/commander-native-plan.md`
 - Each change deletes what it replaces in the same commit.
 - Never touch a file another plan deletes (2026-10-21: the reverted 047bccc65 added 16 conflicts to #2549).
 - Say what is slow or broken the moment it shows, out loud, instead of working around it.
+
+
+### Installer reporting implementation checkpoint; user pause
+
+Draft #2620 now moves publication into the existing RuntimeProbe/ServiceSender family,
+selects the runtime that owns each migration stage, and keeps terminal reporting outside
+TargetTree cleanup. RunMode owns rehearsal/live behavior while FieldCodec retains the
+single Boolean launcher representation. The outgoing report retains its declared probe
+source in memory; the incoming report runs its installed module after source retirement.
+Probe commands use runtime home instead of the retired checkout. Existing bridge resume
+ownership now covers partial pause failure and decodes a broken socket into the existing
+per-bridge failure type.
+
+Latest focused report/cleanup tests pass14/14 (22.44s); the preceding command/bridge batch
+passed43/43 before retained-source execution was added. These are source checks. Actual
+new installer build/rollback/fresh-root qualification and the paired installed journey
+remain open. The earlier package-probe test failure from the main editable environment
+is retained in the root evidence. Bridge state discovery still misses custom/default
+Claude directories; this is an unresolved installer defect owned by root. The full old
+nominal bridge-outcome port is not claimed complete.
+
+Tristan explicitly paused the fleet to reopen Codex after its thread tools retained a
+dead MCP address. Preserve current dirty owned checkouts and uncertain inputs. The
+existing root handoff contains precise heads, results and remaining integration work.

pyproject.toml

@@ -92,6 +92,9 @@ include = [
 [tool.hatch.version]
 path = "src/agent_comms/__init__.py"
 
+[tool.uv]
+cache-keys = [{ file = "pyproject.toml" }, { git = { commit = true } }]
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 pythonpath = [".", "src", "tools"]

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 (
@@ -54,6 +51,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
@@ -76,7 +74,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
@@ -215,15 +212,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,
@@ -321,21 +320,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.
 
@@ -350,8 +334,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)
@@ -363,11 +345,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.
@@ -376,34 +361,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)
@@ -56,7 +55,7 @@ def hold_client_route(self) -> Iterator[None]:
         Independent private roots do not participate in default publication.
         Refuse a publication in progress rather than park a half-open client.
         """
-        current_root = CommsRoute.current().root
+        current_root = RouteFile.read(active_route_path()).launch_root()
         if self.root.resolve() == current_root.resolve():
             with guard_default_route_write(self.root, blocking=False):
                 yield
@@ -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().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: