From 02e084227765694657a45cefaf7efc795d6db41f Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Fri, 23 Oct 2026 07:51:01 -0400
Subject: [PATCH 01/13] Declare selected tool slots beside their native input
 owner

---
 src/agent_comms/native_runtime_input.py | 67 +++++++++++++++++++++++++
 1 file changed, 67 insertions(+)

diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py
index 4d6d23623c..12db467367 100644
--- a/src/agent_comms/native_runtime_input.py
+++ b/src/agent_comms/native_runtime_input.py
@@ -16,8 +16,10 @@
 from agent_comms.private_runtime_schema import PrivateCoordinatorTable
 
 from .coordination_errors import IdentityConflict, PublicationActivationBlocked, StaleFence
+from .declared_family import DeclaredFamily
 from .field_codec import FieldCodec, MalformedRecord, NativeContext
 from .native_claim_epoch import NativeClaimEpoch, UnrecordedNativeClaim
+from .native_tool_call import SelectedToolDenied
 from .native_input_record import (
     NativeColumn,
     NativeInputContext,
@@ -43,6 +45,7 @@
     Column,
     FinalizedColumn,
     FrozenColumn,
+    ForeignKey,
     IntegerStorage,
     MonotoneColumns,
     NoDelete,
@@ -331,6 +334,70 @@ def commit_context(
         if updated.rowcount != 1:
             raise StaleFence("native proof was previously committed")
 
+class SelectedSlotState(DeclaredFamily, affix="Slot"):
+    """Consumption is irreversible; only a matching live terminal can acknowledge it."""
+
+    @classmethod
+    def acknowledge(cls) -> type[SelectedSlotState]:
+        raise SelectedToolDenied("Selected tool terminal was already acknowledged; no retry")
+
+    @classmethod
+    def require_acknowledged(cls) -> None:
+        raise SelectedToolDenied("Selected tool outcome UNKNOWN; no terminal acknowledgement")
+
+
+class ConsumedSlot(SelectedSlotState):
+    @classmethod
+    def acknowledge(cls) -> type[SelectedSlotState]:
+        return AcknowledgedSlot
+
+
+class AcknowledgedSlot(SelectedSlotState):
+    @classmethod
+    def require_acknowledged(cls) -> None:
+        pass
+
+
+@dataclass(frozen=True)
+class SelectedToolSlot(WriteOnceColumns, NativeRuntimeTable, WithoutRowid):
+    """One durable tool slot, retained and retired with its original native input."""
+
+    input_id: Annotated[str, FrozenColumn(primary_key=True)]
+    call_id: Annotated[str, FrozenColumn()]
+    native_input_id: Annotated[str, FrozenColumn(index=True)]
+    state: Annotated[
+        type[SelectedSlotState], Column(check=SelectedSlotState.sql_constraint("state"))
+    ] = ConsumedSlot
+
+    foreign_keys = (
+        ForeignKey(target=NativeRuntimeInput, target_columns=("input_id",),
+                   columns=("native_input_id",), on_delete="CASCADE"),
+    )
+
+    @classmethod
+    def consume(
+        cls, db: sqlite3.Connection, input_id: str, call_id: str, native_input_id: str,
+    ) -> None:
+        if cls(input_id, call_id, native_input_id).insert_if_absent(db).rowcount != 1:
+            raise SelectedToolDenied("Selected tool input already consumed; outcome may be UNKNOWN")
+
+    def require_call(self, call_id: str) -> None:
+        if self.call_id != call_id:
+            raise SelectedToolDenied("Selected tool has no matching terminal acknowledgement")
+
+    def acknowledge(self, db: sqlite3.Connection, call_id: str) -> None:
+        self.require_call(call_id)
+        state = self.state.acknowledge()
+        if self.update(db, where="input_id=? AND state=?", parameters=(
+            self.input_id, self.state.declared_name,
+        ), state=state).rowcount != 1:
+            raise SelectedToolDenied("Selected tool terminal changed; outcome UNKNOWN; no retry")
+
+    def verify(self, call_id: str) -> None:
+        self.require_call(call_id)
+        self.state.require_acknowledged()
+
+
 @dataclass(frozen=True)
 class CurrentNativeCursor(
     MonotoneColumns,

From 813961aec13e787f4f5b3d8e9eabffe16b6658c3 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Fri, 23 Oct 2026 07:53:25 -0400
Subject: [PATCH 02/13] Commit selected tool consumption and terminal state
 through coordination

---
 src/agent_comms/channel_coding_tools.py  |  2 +-
 src/agent_comms/native_runtime_input.py  | 14 ++---
 src/agent_comms/native_tool_call.py      | 28 ++++-----
 src/agent_comms/native_tool_transport.py |  3 +-
 src/agent_comms/selected_tool_broker.py  | 79 +++++++-----------------
 5 files changed, 45 insertions(+), 81 deletions(-)

diff --git a/src/agent_comms/channel_coding_tools.py b/src/agent_comms/channel_coding_tools.py
index 81f44818bf..6ec53027ed 100644
--- a/src/agent_comms/channel_coding_tools.py
+++ b/src/agent_comms/channel_coding_tools.py
@@ -71,7 +71,7 @@ def check(store: Coordination) -> None:
             self.goal_permission.require_current(snapshot.require(self.owner_name).goal)
             self.verify(store, turn, snapshot)
             slot = call.slot(turn.command.input_id, round_number)
-            consume_selected_slot(turn.launch.session.directory, slot, slot)
+            consume_selected_slot(store, slot, slot, native_input_id=turn.command.input_id)
             owner = self.comms.registry.require(self.owner_name)
             resource = call.allowed_claim()
             if resource is not None:
diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py
index 12db467367..59fa1dc6c1 100644
--- a/src/agent_comms/native_runtime_input.py
+++ b/src/agent_comms/native_runtime_input.py
@@ -359,20 +359,18 @@ def require_acknowledged(cls) -> None:
 
 
 @dataclass(frozen=True)
-class SelectedToolSlot(WriteOnceColumns, NativeRuntimeTable, WithoutRowid):
+class SelectedToolSlot(NativeRuntimeTable, WithoutRowid):
     """One durable tool slot, retained and retired with its original native input."""
 
     input_id: Annotated[str, FrozenColumn(primary_key=True)]
     call_id: Annotated[str, FrozenColumn()]
     native_input_id: Annotated[str, FrozenColumn(index=True)]
-    state: Annotated[
-        type[SelectedSlotState], Column(check=SelectedSlotState.sql_constraint("state"))
-    ] = ConsumedSlot
+    state: type[SelectedSlotState] = ConsumedSlot
 
-    foreign_keys = (
-        ForeignKey(target=NativeRuntimeInput, target_columns=("input_id",),
-                   columns=("native_input_id",), on_delete="CASCADE"),
-    )
+    @classmethod
+    def references(cls) -> tuple[ForeignKey, ...]:
+        return (ForeignKey(target=NativeRuntimeInput, target_columns=("input_id",),
+                           columns=("native_input_id",), on_delete="CASCADE"),)
 
     @classmethod
     def consume(
diff --git a/src/agent_comms/native_tool_call.py b/src/agent_comms/native_tool_call.py
index 672113e1e8..7c4c454455 100644
--- a/src/agent_comms/native_tool_call.py
+++ b/src/agent_comms/native_tool_call.py
@@ -77,7 +77,7 @@ def start(self, call: NativeToolCall) -> None:
         raise SelectedToolDenied("Native tool start is missing an announcement or was repeated")
 
     def finish(
-        self, call: NativeToolCall, is_error: bool, directory: Path,
+        self, call: NativeToolCall, is_error: bool, coordination_path: Path,
         input_id: str, round_number: int,
     ) -> None:
         raise SelectedToolDenied("Native tool terminal is missing or repeated")
@@ -100,13 +100,13 @@ def start(self, call: NativeToolCall) -> None:
 class ExecutingCall(Executing, CallObservation):
 
     def finish(
-        self, call: NativeToolCall, is_error: bool, directory: Path,
+        self, call: NativeToolCall, is_error: bool, coordination_path: Path,
         input_id: str, round_number: int,
     ) -> None:
-        # A failed terminal fsync cannot be retried or converted to a completed
+        # A failed terminal commit cannot be retried or converted to a completed
         # error by a subsequent event. The live acknowledgement must succeed exactly once.
         call.progress.observation = UncertainCall()
-        call.progress.claim_scope.finish(call, is_error, directory, input_id, round_number)
+        call.progress.claim_scope.finish(call, is_error, coordination_path, input_id, round_number)
         call.progress.observation = CompletedCall()
 
 
@@ -126,7 +126,7 @@ async def consume(self, call: NativeToolCall, action: Callable[[], Awaitable[Non
         raise SelectedToolDenied("Native tool request was already consumed; outcome may be UNKNOWN")
 
     def finish(
-        self, call: NativeToolCall, is_error: bool, directory: Path,
+        self, call: NativeToolCall, is_error: bool, coordination_path: Path,
         input_id: str, round_number: int,
     ) -> None:
         raise SelectedToolDenied("Native tool ran without owner claim_scope")
@@ -137,7 +137,7 @@ def refuse(self, call: NativeToolCall, refusal: CallClaim) -> None:
         call.progress.observation.refuse(call, refusal)
 
     async def consume(self, call: NativeToolCall, action: Callable[[], Awaitable[None]]) -> None:
-        # No retry even when the callback is cancelled or its durable fsync fails.
+        # No retry even when the callback is cancelled or its durable commit fails.
         call.progress.claim_scope = UncertainClaim()
         await action()
         call.progress.observation.require(Executing, ToolNotExecuting())
@@ -146,10 +146,10 @@ async def consume(self, call: NativeToolCall, action: Callable[[], Awaitable[Non
 
 class AuthorizedCall(CallClaim):
     def finish(
-        self, call: NativeToolCall, is_error: bool, directory: Path,
+        self, call: NativeToolCall, is_error: bool, coordination_path: Path,
         input_id: str, round_number: int,
     ) -> None:
-        call.commit_terminal(is_error, directory, input_id, round_number)
+        call.commit_terminal(is_error, coordination_path, input_id, round_number)
 
 
 class UncertainClaim(CallClaim):
@@ -158,7 +158,7 @@ def refuse(self, call: NativeToolCall, refusal: CallClaim) -> None:
         pass
 
     def finish(
-        self, call: NativeToolCall, is_error: bool, directory: Path,
+        self, call: NativeToolCall, is_error: bool, coordination_path: Path,
         input_id: str, round_number: int,
     ) -> None:
         if not is_error:
@@ -224,19 +224,19 @@ async def authorize(self, action: Callable[[], Awaitable[None]]) -> None:
             raise
 
     def finish(
-        self, name: str, is_error: bool | None, directory: Path,
+        self, name: str, is_error: bool | None, coordination_path: Path,
         input_id: str, round_number: int,
     ) -> None:
         if name != self.declared_name or type(is_error) is not bool:
             raise SelectedToolDenied("Native tool terminal identity/result changed")
-        self.progress.observation.finish(self, is_error, directory, input_id, round_number)
+        self.progress.observation.finish(self, is_error, coordination_path, input_id, round_number)
 
     def accepts_current_request(self, candidate: NativeToolCall) -> bool:
         return self.progress.observation.accepts_current_request(self, candidate)
 
     @abstractmethod
     def commit_terminal(
-        self, is_error: bool, directory: Path, input_id: str, round_number: int
+        self, is_error: bool, coordination_path: Path, input_id: str, round_number: int
     ) -> None: ...
 
     @abstractmethod
@@ -267,12 +267,12 @@ def decide_storage(self, registry: Registration, thread: str) -> tuple[StorageEf
         return effects
 
     def commit_terminal(
-        self, is_error: bool, directory: Path, input_id: str, round_number: int
+        self, is_error: bool, coordination_path: Path, input_id: str, round_number: int
     ) -> None:
         from .selected_tool_broker import record_selected_terminal
 
         slot = self.slot(input_id, round_number)
-        record_selected_terminal(directory, slot, slot)
+        record_selected_terminal(coordination_path, slot, slot)
 
     def claim_failed(self) -> None:
         # A denied ordinary tool may report an error and let Pi explain it. The
diff --git a/src/agent_comms/native_tool_transport.py b/src/agent_comms/native_tool_transport.py
index dea99c6acd..af33eee5c5 100644
--- a/src/agent_comms/native_tool_transport.py
+++ b/src/agent_comms/native_tool_transport.py
@@ -17,6 +17,7 @@
 from typing import TYPE_CHECKING, Generic, TypeVar
 
 from .child_process import join_retirement
+from .coordination_database import CoordinationStore
 from .field_codec import FieldCodec, MalformedRecord, Token256Text
 from .lifecycle_capability import Complete
 from .native_tool_call import (
@@ -320,7 +321,7 @@ def tool_finished(self, event: ToolExecutionEnd, input_id: str) -> None:
         call.finish(
             FieldCodec.encode(event.tool_name),
             FieldCodec.encode(event.is_error),
-            self.listener.session_directory,
+            CoordinationStore.path_in(self.turn.launch.session.root),
             input_id,
             identity.round_number,
         )
diff --git a/src/agent_comms/selected_tool_broker.py b/src/agent_comms/selected_tool_broker.py
index dbf814fecd..3de56f790f 100644
--- a/src/agent_comms/selected_tool_broker.py
+++ b/src/agent_comms/selected_tool_broker.py
@@ -9,8 +9,8 @@
 from __future__ import annotations
 
 import logging
-import os
 import re
+import sqlite3
 import stat
 from collections.abc import Callable
 from dataclasses import dataclass, field
@@ -18,8 +18,8 @@
 from pathlib import Path
 from typing import ClassVar
 
-from agent_comms.child_process import Platform
 from agent_comms.coordinator import Coordination
+from .coordination_session import ReadNativeCoordinationSession
 
 from .comms import Comms
 from agent_comms.native_runtime_input import NativeRuntimeTable
@@ -28,7 +28,7 @@
 from .envelope_claim_transitions import ExistingFileClaim, WakeClaim
 from .field_codec import FieldCodec, Id128Text, MalformedRecord
 from .native_input_record import FullNativeExecution, NativeInputIdentity
-from .native_runtime_input import NativeRuntimeInput
+from .native_runtime_input import NativeRuntimeInput, SelectedToolSlot
 from .native_tool_call import (
     CoordinationBusy,
     NativeToolCall,
@@ -41,7 +41,6 @@
     ToolTransport,
     UnboundToolTransport,
 )
-from .private_path import PrivateDirectoryRole, PrivateFileRole
 from .selected_actions import SelectedAction
 from .selected_claim import publish_selected_resource_claim, write_selected_claimed_file
 from .turn_context import InstructionFile
@@ -136,11 +135,11 @@ def __post_init__(self) -> None:
     declared_name: ClassVar[str] = "selected_claimed_write"
 
     def commit_terminal(
-        self, is_error: bool, directory: Path, input_id: str, round_number: int
+        self, is_error: bool, coordination_path: Path, input_id: str, round_number: int
     ) -> None:
         if is_error:
             raise SelectedToolDenied("Native Pi selected tool did not finish successfully")
-        verify_selected_terminal(directory, input_id, self.call_id)
+        verify_selected_terminal(coordination_path, input_id, self.call_id)
 
     def claim_failed(self) -> None:
         raise SelectedToolDenied("Native Pi selected tool claim_scope failed; outcome UNKNOWN")
@@ -173,59 +172,29 @@ def selected_extension(package: Path) -> Path:
     return source
 
 
-def consume_selected_slot(directory: Path, input_id: str, call_id: str) -> None:
-    """Persist one consumed slot BEFORE a bus append; even failed fsync is UNKNOWN.
-
-    The caller must provide a private, physical owner-owned session directory.
-    An existing slot is never reopened or reclaimed, even after a process crash.
-    """
+def consume_selected_slot(
+    store: Coordination, input_id: str, call_id: str, *, native_input_id: str,
+) -> None:
+    """Commit consumption BEFORE any action; a failed commit is UNKNOWN, never retried."""
     try:
         Id128Text.decode(input_id)
     except MalformedRecord as error:
         raise SelectedToolDenied("Selected input identity is invalid") from error
     if type(call_id) is not str or not _CALL_ID.fullmatch(call_id):
         raise SelectedToolDenied("Selected tool call identity is invalid")
-    directory = Path(directory).absolute()
-    info = directory.lstat()
-    if PrivateDirectoryRole.violation(info) is not None or directory.resolve() != directory:
-        raise SelectedToolDenied("Selected tool session directory is not private")
-    ledger = directory / "selected-tool-ledger"
-    try:
-        ledger.mkdir(mode=0o700, exist_ok=True)
-        info = ledger.lstat()
-        if PrivateDirectoryRole.violation(info) is not None:
-            raise SelectedToolDenied("Selected tool ledger is not private")
-        Platform.current().sync_directory(directory)
-        fd = os.open(ledger / input_id, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
-    except FileExistsError as error:
-        raise SelectedToolDenied(
-            "Selected tool input already consumed; outcome may be UNKNOWN"
-        ) from error
-    except OSError as error:
-        raise SelectedToolDenied("Selected tool ledger unavailable; outcome UNKNOWN") from error
     try:
-        with os.fdopen(fd, "wb") as stream:
-            stream.write((call_id + "\n").encode("ascii"))
-            stream.flush()
-            os.fsync(stream.fileno())
-        Platform.current().sync_directory(ledger)
-    except OSError as error:
-        # Do NOT remove the visible slot; a failed fsync or a crash may have
-        # committed it. No automatic re-append, retry, or write follows.
+        with store.session.writing() as db:
+            SelectedToolSlot.consume(db, input_id, call_id, native_input_id)
+    except sqlite3.Error as error:
         raise SelectedToolDenied("Selected tool consumption UNKNOWN; do not retry") from error
 
 
-def record_selected_terminal(directory: Path, input_id: str, call_id: str) -> None:
-    """Only a synced terminal acknowledgement lets the tool return success."""
-    result = Path(directory).absolute() / "selected-tool-ledger" / (input_id + ".done")
+def record_selected_terminal(coordination_path: Path, input_id: str, call_id: str) -> None:
+    """Only a committed terminal acknowledgement lets the tool return success."""
     try:
-        descriptor = os.open(result, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
-        with os.fdopen(descriptor, "wb") as output:
-            output.write((call_id + "\n").encode("ascii"))
-            output.flush()
-            os.fsync(output.fileno())
-        Platform.current().sync_directory(result.parent)
-    except OSError as error:
+        with Coordination(coordination_path) as store, store.session.writing() as db:
+            SelectedToolSlot.require(db, input_id=input_id).acknowledge(db, call_id)
+    except sqlite3.Error as error:
         raise SelectedToolDenied("Selected tool terminal acknowledgement UNKNOWN; no retry") from error
 
 
@@ -306,18 +275,14 @@ def _require_sent_full_input(
         raise SelectedToolDenied("Selected input was already settled")
 
 
-def verify_selected_terminal(directory: Path, input_id: str, call_id: str) -> None:
+def verify_selected_terminal(coordination_path: Path, input_id: str, call_id: str) -> None:
     """Corroborate a live owner result; a persisted acknowledgement alone grants nothing."""
     try:
         Id128Text.decode(input_id)
     except MalformedRecord as error:
         raise SelectedToolDenied("Selected input identity is invalid") from error
-    acknowledgement = Path(directory).absolute() / "selected-tool-ledger" / (input_id + ".done")
-    info = acknowledgement.lstat()
-    if PrivateFileRole.violation(info) is not None:
-        raise SelectedToolDenied("Selected tool acknowledgement is not private")
-    if acknowledgement.read_bytes() != (call_id + "\n").encode("ascii"):
-        raise SelectedToolDenied("Selected tool has no matching terminal acknowledgement")
+    with Coordination(str(coordination_path), session_type=ReadNativeCoordinationSession) as store, store.session.read():
+        SelectedToolSlot.require(store.session._connection, input_id=input_id).verify(call_id)
 
 
 def perform_selected_write(
@@ -330,7 +295,7 @@ def perform_selected_write(
     request: SelectedToolRequest,
 ) -> None:
     """Only the owner calls the existing selected claim and writer boundaries."""
-    consume_selected_slot(session_dir, input_id, request.call_id)
+    consume_selected_slot(store, input_id, request.call_id, native_input_id=input_id)
     # Any exception from here leaves the slot consumed. The caller reports
     # UNKNOWN and never automatically reissues the claim or mutation.
     claimed = publish_selected_resource_claim(
@@ -339,7 +304,7 @@ def perform_selected_write(
     write_selected_claimed_file(
         comms, store, claim_scope, owner_name, claimed, request.arguments.contents.encode("utf-8")
     )
-    record_selected_terminal(session_dir, input_id, request.call_id)
+    record_selected_terminal(store.session.path, input_id, request.call_id)
 
 
 class SelectedToolSocket(ToolTransport[SelectedToolRequest]):

From 6bfed42abed83ed801f4cbf2cea8b7e28bb887ef Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:00:07 -0400
Subject: [PATCH 03/13] Preserve ordinary coding slots without inventing a
 native input reference

---
 src/agent_comms/channel_coding_tools.py |  2 +-
 src/agent_comms/native_runtime_input.py | 13 +++----------
 src/agent_comms/selected_tool_broker.py | 13 ++++---------
 3 files changed, 8 insertions(+), 20 deletions(-)

diff --git a/src/agent_comms/channel_coding_tools.py b/src/agent_comms/channel_coding_tools.py
index 6ec53027ed..f665b2b142 100644
--- a/src/agent_comms/channel_coding_tools.py
+++ b/src/agent_comms/channel_coding_tools.py
@@ -71,7 +71,7 @@ def check(store: Coordination) -> None:
             self.goal_permission.require_current(snapshot.require(self.owner_name).goal)
             self.verify(store, turn, snapshot)
             slot = call.slot(turn.command.input_id, round_number)
-            consume_selected_slot(store, slot, slot, native_input_id=turn.command.input_id)
+            consume_selected_slot(store, slot, slot)
             owner = self.comms.registry.require(self.owner_name)
             resource = call.allowed_claim()
             if resource is not None:
diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py
index 59fa1dc6c1..f9530563f6 100644
--- a/src/agent_comms/native_runtime_input.py
+++ b/src/agent_comms/native_runtime_input.py
@@ -45,7 +45,6 @@
     Column,
     FinalizedColumn,
     FrozenColumn,
-    ForeignKey,
     IntegerStorage,
     MonotoneColumns,
     NoDelete,
@@ -360,23 +359,17 @@ def require_acknowledged(cls) -> None:
 
 @dataclass(frozen=True)
 class SelectedToolSlot(NativeRuntimeTable, WithoutRowid):
-    """One durable tool slot, retained and retired with its original native input."""
+    """The consumed slot and its live terminal acknowledgement, in one transaction store."""
 
     input_id: Annotated[str, FrozenColumn(primary_key=True)]
     call_id: Annotated[str, FrozenColumn()]
-    native_input_id: Annotated[str, FrozenColumn(index=True)]
     state: type[SelectedSlotState] = ConsumedSlot
 
-    @classmethod
-    def references(cls) -> tuple[ForeignKey, ...]:
-        return (ForeignKey(target=NativeRuntimeInput, target_columns=("input_id",),
-                           columns=("native_input_id",), on_delete="CASCADE"),)
-
     @classmethod
     def consume(
-        cls, db: sqlite3.Connection, input_id: str, call_id: str, native_input_id: str,
+        cls, db: sqlite3.Connection, input_id: str, call_id: str,
     ) -> None:
-        if cls(input_id, call_id, native_input_id).insert_if_absent(db).rowcount != 1:
+        if cls(input_id, call_id).insert_if_absent(db).rowcount != 1:
             raise SelectedToolDenied("Selected tool input already consumed; outcome may be UNKNOWN")
 
     def require_call(self, call_id: str) -> None:
diff --git a/src/agent_comms/selected_tool_broker.py b/src/agent_comms/selected_tool_broker.py
index 3de56f790f..caf9811d42 100644
--- a/src/agent_comms/selected_tool_broker.py
+++ b/src/agent_comms/selected_tool_broker.py
@@ -72,7 +72,6 @@ def mode(self, owner):
             owner.store,
             owner.resource_scope,
             owner.owner_name,
-            owner.session_dir,
             owner.input_id,
         )
 
@@ -172,9 +171,7 @@ def selected_extension(package: Path) -> Path:
     return source
 
 
-def consume_selected_slot(
-    store: Coordination, input_id: str, call_id: str, *, native_input_id: str,
-) -> None:
+def consume_selected_slot(store: Coordination, input_id: str, call_id: str) -> None:
     """Commit consumption BEFORE any action; a failed commit is UNKNOWN, never retried."""
     try:
         Id128Text.decode(input_id)
@@ -184,7 +181,7 @@ def consume_selected_slot(
         raise SelectedToolDenied("Selected tool call identity is invalid")
     try:
         with store.session.writing() as db:
-            SelectedToolSlot.consume(db, input_id, call_id, native_input_id)
+            SelectedToolSlot.consume(db, input_id, call_id)
     except sqlite3.Error as error:
         raise SelectedToolDenied("Selected tool consumption UNKNOWN; do not retry") from error
 
@@ -203,7 +200,6 @@ def selected_tool_mode_for_owner(
     store: Coordination,
     claim_scope: WakeClaim,
     owner_name: str,
-    session_dir: Path,
     input_id: str,
 ) -> SelectedToolMode:
     """Capture only owner-derived typed fields for this one FULL selected input.
@@ -226,7 +222,7 @@ def owner_action(request: SelectedToolRequest) -> None:
         with Coordination(path) as worker_store:
             verify_sent_full_input(worker_store, claim_scope, owner_name, input_id)
             perform_selected_write(
-                comms, worker_store, claim_scope, owner_name, session_dir, input_id, request
+                comms, worker_store, claim_scope, owner_name, input_id, request
             )
 
     return SelectedToolMode(owner_action)
@@ -290,12 +286,11 @@ def perform_selected_write(
     store: Coordination,
     claim_scope: WakeClaim,
     owner_name: str,
-    session_dir: Path,
     input_id: str,
     request: SelectedToolRequest,
 ) -> None:
     """Only the owner calls the existing selected claim and writer boundaries."""
-    consume_selected_slot(store, input_id, request.call_id, native_input_id=input_id)
+    consume_selected_slot(store, input_id, request.call_id)
     # Any exception from here leaves the slot consumed. The caller reports
     # UNKNOWN and never automatically reissues the claim or mutation.
     claimed = publish_selected_resource_claim(

From 28ce471cf22aa6376eed6ea7f207212e34fc061b Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:01:05 -0400
Subject: [PATCH 04/13] Migrate selected tool fixtures to their coordination
 slot state

---
 tests/test_channel_coding_tools.py    |  3 +-
 tests/test_native_argument_refusal.py |  4 ++-
 tests/test_selected_tool_broker.py    | 45 +++++++++++++++++----------
 3 files changed, 34 insertions(+), 18 deletions(-)

diff --git a/tests/test_channel_coding_tools.py b/tests/test_channel_coding_tools.py
index 423539392e..b97bb57ed6 100644
--- a/tests/test_channel_coding_tools.py
+++ b/tests/test_channel_coding_tools.py
@@ -8,6 +8,7 @@
 import os
 import secrets
 from pathlib import Path
+from types import SimpleNamespace
 from dataclasses import replace
 from tempfile import TemporaryDirectory
 
@@ -223,7 +224,7 @@ async def allow(self, turn, round_number, call):
 
     owner = Owner()
     socket = CodingToolSocket(tool_listener(tmp_path, secrets.token_hex(32)), owner, pid=os.getpid())
-    socket.turn = object()
+    socket.turn = SimpleNamespace(launch=SimpleNamespace(session=SimpleNamespace(root=tmp_path)))
     socket.announce((ToolCallContent(id="late", name="read", arguments={"path": "a"}),))
     socket.tool_started(ToolExecutionStart(
         tool_call_id=ReportedText("late"), tool_name=ReportedText("read"), args={"path": "a"}
diff --git a/tests/test_native_argument_refusal.py b/tests/test_native_argument_refusal.py
index c14fb8ab15..599daa6427 100644
--- a/tests/test_native_argument_refusal.py
+++ b/tests/test_native_argument_refusal.py
@@ -4,6 +4,7 @@
 import asyncio
 import json
 import os
+from types import SimpleNamespace
 
 import pytest
 
@@ -31,6 +32,7 @@ async def allow(self, round_number, call):
             raise AssertionError("invalid arguments must never acquire authority")
 
     socket = CodingToolSocket(tool_listener(tmp_path, "a" * 64), Owner(), pid=os.getpid())
+    socket.turn = SimpleNamespace(launch=SimpleNamespace(session=SimpleNamespace(root=tmp_path)))
     args = {"command": False}
     raw = (json.dumps({"token": socket.listener.token, "call_id": "bad", "name": "bash",
                        "arguments": {"command": "false"}}) + "\n").encode()
@@ -47,6 +49,6 @@ async def allow(self, round_number, call):
         tool_call_id=ReportedText("bad"), tool_name=ReportedText("bash"), is_error=ReportedFlag(True)
     ), "b" * 32)
     socket.assert_complete()
-    assert not (tmp_path / "selected-tool-ledger").exists()
+    assert not (tmp_path / "coordination.sqlite3").exists()
 
 
diff --git a/tests/test_selected_tool_broker.py b/tests/test_selected_tool_broker.py
index c5f70514bc..84548670dc 100644
--- a/tests/test_selected_tool_broker.py
+++ b/tests/test_selected_tool_broker.py
@@ -10,6 +10,7 @@
 import secrets
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -25,6 +26,14 @@
 )
 from agent_comms.pi_events import ToolExecutionEnd, ToolExecutionStart
 from agent_comms.pi_payloads import ToolCallContent
+from agent_comms.native_runtime_input import AcknowledgedSlot, ConsumedSlot, SelectedToolSlot
+
+
+@pytest.fixture
+def slot_store(tmp_path):
+    with Coordination(str(tmp_path / "coordination.sqlite3")) as store:
+        PrivateCoordinatorTable.install(store)
+        yield store
 
 
 def _wire(token: str, **changes: object) -> bytes:
@@ -71,19 +80,21 @@ def test_pre_turn_intent_is_distinct_from_owner_bound_mode() -> None:
         broker.SelectedToolMode(None)  # type: ignore[arg-type]
 
 
-@pytest.mark.skipif(sys.platform != "linux", reason="selected tool storage requires POSIX dirfd")
-def test_consumption_is_once_only_and_terminal_is_separate(tmp_path: Path) -> None:
-    tmp_path.chmod(0o700)
+def test_consumption_is_once_only_and_terminal_is_separate(slot_store) -> None:
     input_id = secrets.token_hex(16)
-    broker.consume_selected_slot(tmp_path, input_id, "call_1")
-    ledger = tmp_path / "selected-tool-ledger"
-    assert (ledger / input_id).read_text() == "call_1\n"
-    assert not (ledger / (input_id + ".done")).exists()
+    broker.consume_selected_slot(slot_store, input_id, "call_1")
+    row = SelectedToolSlot.require(slot_store.session._connection, input_id=input_id)
+    assert row.call_id == "call_1"
+    assert row.state is ConsumedSlot
+    with pytest.raises(SelectedToolDenied):
+        broker.verify_selected_terminal(slot_store.session.path, input_id, "call_1")
     with pytest.raises(SelectedToolDenied):
-        broker.consume_selected_slot(tmp_path, input_id, "call_2")
-    broker.record_selected_terminal(tmp_path, input_id, "call_1")
+        broker.consume_selected_slot(slot_store, input_id, "call_2")
+    broker.record_selected_terminal(slot_store.session.path, input_id, "call_1")
+    broker.verify_selected_terminal(slot_store.session.path, input_id, "call_1")
+    assert SelectedToolSlot.require(slot_store.session._connection, input_id=input_id).state is AcknowledgedSlot
     with pytest.raises(SelectedToolDenied):
-        broker.record_selected_terminal(tmp_path, input_id, "call_1")
+        broker.record_selected_terminal(slot_store.session.path, input_id, "call_1")
 
 
 @pytest.mark.skipif(sys.platform != "linux", reason="selected tool storage requires POSIX dirfd")
@@ -112,7 +123,7 @@ def test_owner_mode_denies_unreserved_input_before_consuming_ledger(tmp_path: Pa
         PrivateCoordinatorTable.install(store)
         input_id = secrets.token_hex(16)
         mode = broker.selected_tool_mode_for_owner(
-            comms, store, claim_scope, "owner", session_dir, input_id
+            comms, store, claim_scope, "owner", input_id
         )
         with pytest.raises(SelectedToolDenied):
             mode.action(
@@ -120,7 +131,7 @@ def test_owner_mode_denies_unreserved_input_before_consuming_ledger(tmp_path: Pa
                     "call_1", broker.SelectedWriteArguments("notes.txt", "write")
                 )
             )
-        assert not (session_dir / "selected-tool-ledger").exists()
+        assert SelectedToolSlot.count(store.session._connection) == 0
 
 
 @pytest.mark.skipif(sys.platform != "linux", reason="selected tool socket requires SO_PEERCRED")
@@ -197,10 +208,11 @@ async def run() -> None:
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("abandoned", (False, True))
-async def test_listener_isolated_from_previous_turn_and_keeps_session_ledger(
+async def test_listener_isolated_from_previous_turn_and_keeps_consumed_slot(
     tmp_path: Path,
     tool_listener,
     abandoned: bool,
+    slot_store,
 ) -> None:
     tmp_path.chmod(0o700)
     previous = tool_listener(tmp_path, secrets.token_hex(32))
@@ -213,11 +225,12 @@ async def test_listener_isolated_from_previous_turn_and_keeps_session_ledger(
     observed: list[broker.SelectedToolRequest] = []
 
     def accept(request: broker.SelectedToolRequest) -> None:
-        broker.consume_selected_slot(tmp_path, input_id, request.call_id)
-        broker.record_selected_terminal(tmp_path, input_id, request.call_id)
+        broker.consume_selected_slot(slot_store, input_id, request.call_id)
+        broker.record_selected_terminal(slot_store.session.path, input_id, request.call_id)
         observed.append(request)
 
     server = broker.SelectedToolSocket(tool_listener(tmp_path, token), accept, pid=os.getpid())
+    server.turn = SimpleNamespace(launch=SimpleNamespace(session=SimpleNamespace(root=tmp_path)))
     assert server.listener.path != previous.path
     assert server.listener.path.parent.stat().st_mode & 0o777 == 0o700
     await server.start()
@@ -255,7 +268,7 @@ def accept(request: broker.SelectedToolRequest) -> None:
                 broker.SelectedWriteArguments("notes.txt", "é"),
             )
         ]
-        assert (tmp_path / "selected-tool-ledger" / (input_id + ".done")).exists()
+        assert SelectedToolSlot.require(slot_store.session._connection, input_id=input_id).state is AcknowledgedSlot
     finally:
         await server.close()
     assert not server.listener.path.exists()

From 8bd0aee16d2e7d3654abd5107ed4d04d3513dc0c Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:08:23 -0400
Subject: [PATCH 05/13] Carry selected tool identities directly from their
 ledger files

---
 src/agent_comms/native_runtime_input.py    |  20 ++++
 tools/cutover/carry_selected_tool_slots.py | 114 +++++++++++++++++++++
 2 files changed, 134 insertions(+)
 create mode 100644 tools/cutover/carry_selected_tool_slots.py

diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py
index f9530563f6..50c3445c4e 100644
--- a/src/agent_comms/native_runtime_input.py
+++ b/src/agent_comms/native_runtime_input.py
@@ -344,6 +344,10 @@ def acknowledge(cls) -> type[SelectedSlotState]:
     def require_acknowledged(cls) -> None:
         raise SelectedToolDenied("Selected tool outcome UNKNOWN; no terminal acknowledgement")
 
+    @classmethod
+    def carried(cls, incoming: type[SelectedSlotState]) -> type[SelectedSlotState]:
+        return incoming
+
 
 class ConsumedSlot(SelectedSlotState):
     @classmethod
@@ -356,6 +360,12 @@ class AcknowledgedSlot(SelectedSlotState):
     def require_acknowledged(cls) -> None:
         pass
 
+    @classmethod
+    def carried(cls, incoming: type[SelectedSlotState]) -> type[SelectedSlotState]:
+        # A carry interrupted after deleting .done files must not erase the
+        # acknowledgement already committed in this store.
+        return cls
+
 
 @dataclass(frozen=True)
 class SelectedToolSlot(NativeRuntimeTable, WithoutRowid):
@@ -388,6 +398,16 @@ def verify(self, call_id: str) -> None:
         self.require_call(call_id)
         self.state.require_acknowledged()
 
+    def carry(self, db: sqlite3.Connection) -> None:
+        """Transfer a legacy slot, preserving an earlier committed acknowledgement."""
+        if self.insert_if_absent(db).rowcount == 1:
+            return
+        current = self.require(db, input_id=self.input_id)
+        current.require_call(self.call_id)
+        state = current.state.carried(self.state)
+        if state is not current.state:
+            current.acknowledge(db, self.call_id)
+
 
 @dataclass(frozen=True)
 class CurrentNativeCursor(
diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
new file mode 100644
index 0000000000..537bdfa1e7
--- /dev/null
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -0,0 +1,114 @@
+"""Move the selected-tool files into their one coordination row, then retire them.
+
+The filename and its contents are the original slot identity. No input table or
+transcript can reconstruct them. Delete this carry after every root has crossed.
+"""
+
+from __future__ import annotations
+
+from contextlib import closing
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_comms.child_process import Platform
+from agent_comms.coordination_database import CoordinationStore
+from agent_comms.errors import RelationViolationError
+from agent_comms.installer.carry import Carried, Carry
+from agent_comms.locked_store import SQLiteFile
+from agent_comms.native_runtime_input import (
+    AcknowledgedSlot,
+    ConsumedSlot,
+    SelectedToolSlot,
+)
+from agent_comms.threads import ThreadCatalog
+from agent_comms.typed_table import HeldBusFile
+
+
+@dataclass(frozen=True)
+class LedgerSlot:
+    """One legacy consumed file and its optional matching acknowledgement."""
+
+    consumed: Path
+    acknowledgements: tuple[Path, ...]
+
+    @property
+    def files(self) -> tuple[Path, ...]:
+        return (self.consumed, *self.acknowledgements)
+
+    def read(self) -> SelectedToolSlot:
+        original = self.consumed.read_bytes()
+        state = ConsumedSlot
+        for acknowledgement in self.acknowledgements:
+            if acknowledgement.read_bytes() != original:
+                raise RelationViolationError(
+                    f"Selected tool acknowledgement differs from {self.consumed}: {acknowledgement}"
+                )
+            state = AcknowledgedSlot
+        return SelectedToolSlot(
+            self.consumed.name, original.decode("ascii").removesuffix("\n"), state,
+        )
+
+
+class SelectedToolSlots(Carry):
+    opened_files = (CoordinationStore,)
+
+    def slots(self) -> tuple[LedgerSlot, ...]:
+        slots = []
+        for directory in sorted(self.root.glob(str(
+            ThreadCatalog.pattern().parent / "selected-tool-ledger"
+        ))):
+            self.require_confined(self.root, directory)
+            files = frozenset(directory.iterdir())
+            for path in files:
+                self.require_confined(self.root, path)
+                if not path.is_file():
+                    raise RelationViolationError(f"Selected tool ledger entry is not a file: {path}")
+                if path.suffix == ".done" and path.with_suffix("") not in files:
+                    raise RelationViolationError(f"Selected tool acknowledgement has no consumed file: {path}")
+            slots.extend(
+                LedgerSlot(path, tuple(ack for ack in (path.with_name(path.name + ".done"),)
+                                       if ack in files))
+                for path in sorted(files) if path.suffix != ".done"
+            )
+        return tuple(slots)
+
+    def input_files(self) -> tuple[Path, ...]:
+        return (*super().input_files(), *(path for slot in self.slots() for path in slot.files))
+
+    def carry(self, *, apply: bool) -> Carried:
+        slots = self.slots()
+        if not slots:
+            return Carried(0)
+        rows = tuple(slot.read() for slot in slots)
+        files = tuple(path for slot in slots for path in slot.files)
+        acknowledged = sum(len(slot.acknowledgements) for slot in slots)
+        detail = {
+            "consumed_files": len(slots), "acknowledgement_files": acknowledged,
+            "consumed_rows": len(slots) - acknowledged, "acknowledged_rows": acknowledged,
+            "slot_rows": len(slots), "deleted_files": len(files) if apply else 0,
+        }
+        if not apply:
+            return Carried(len(files), detail)
+        path = CoordinationStore.path_in(self.root)
+        self.require_confined(self.root, path)
+        HeldBusFile(CoordinationStore, path).require()
+        with closing(SQLiteFile.writer(path)) as db, SQLiteFile.write_transaction(db):
+            SelectedToolSlot.create(db, if_not_exists=True)
+            for row in rows:
+                row.carry(db)
+        # Every row is committed before the first unlink. Persist removal of
+        # acknowledgements before removing consumed files: a killed carry can
+        # leave a consumed file, but can never manufacture an orphan .done.
+        directories = tuple(dict.fromkeys(slot.consumed.parent for slot in slots))
+        for slot in slots:
+            for acknowledgement in slot.acknowledgements:
+                acknowledgement.unlink()
+        for directory in directories:
+            Platform.current().sync_directory(directory)
+        for slot in slots:
+            slot.consumed.unlink()
+        for directory in directories:
+            directory.rmdir()
+            Platform.current().sync_directory(directory.parent)
+        return Carried(len(files), detail,
+                       (path.relative_to(self.root), *(file.relative_to(self.root) for file in files)))

From 96c69ab397097a468020a867aee87828eab94845 Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:10:20 -0400
Subject: [PATCH 06/13] Discover legacy ledgers through every saved-session
 declaration

---
 tools/cutover/carry_selected_tool_slots.py | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index 537bdfa1e7..db10d8c875 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -18,9 +18,10 @@
 from agent_comms.native_runtime_input import (
     AcknowledgedSlot,
     ConsumedSlot,
+    SelectedSlotState,
     SelectedToolSlot,
 )
-from agent_comms.threads import ThreadCatalog
+from agent_comms.native_session_files import SavedSessionFile
 from agent_comms.typed_table import HeldBusFile
 
 
@@ -37,7 +38,7 @@ def files(self) -> tuple[Path, ...]:
 
     def read(self) -> SelectedToolSlot:
         original = self.consumed.read_bytes()
-        state = ConsumedSlot
+        state: type[SelectedSlotState] = ConsumedSlot
         for acknowledgement in self.acknowledgements:
             if acknowledgement.read_bytes() != original:
                 raise RelationViolationError(
@@ -54,9 +55,11 @@ class SelectedToolSlots(Carry):
 
     def slots(self) -> tuple[LedgerSlot, ...]:
         slots = []
-        for directory in sorted(self.root.glob(str(
-            ThreadCatalog.pattern().parent / "selected-tool-ledger"
-        ))):
+        directories = {
+            path for owner, root in SavedSessionFile.declarations_in(self.root)
+            for path in root.glob(str(owner.pattern().parent / "selected-tool-ledger"))
+        }
+        for directory in sorted(directories):
             self.require_confined(self.root, directory)
             files = frozenset(directory.iterdir())
             for path in files:

From dfbcc47ee3e82c414c3a74acd5cd33c63450befb Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:10:59 -0400
Subject: [PATCH 07/13] Transfer both ledger observations and make interrupted
 retirement restartable

---
 tools/cutover/carry_selected_tool_slots.py | 46 ++++++++++++----------
 1 file changed, 26 insertions(+), 20 deletions(-)

diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index db10d8c875..a28c00e3f6 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from contextlib import closing
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 
 from agent_comms.child_process import Platform
@@ -18,7 +18,6 @@
 from agent_comms.native_runtime_input import (
     AcknowledgedSlot,
     ConsumedSlot,
-    SelectedSlotState,
     SelectedToolSlot,
 )
 from agent_comms.native_session_files import SavedSessionFile
@@ -36,31 +35,36 @@ class LedgerSlot:
     def files(self) -> tuple[Path, ...]:
         return (self.consumed, *self.acknowledgements)
 
-    def read(self) -> SelectedToolSlot:
+    def read(self) -> tuple[SelectedToolSlot, ...]:
         original = self.consumed.read_bytes()
-        state: type[SelectedSlotState] = ConsumedSlot
+        consumed = SelectedToolSlot(
+            self.consumed.name, original.decode("ascii").removesuffix("\n"), ConsumedSlot,
+        )
+        records = [consumed]
         for acknowledgement in self.acknowledgements:
             if acknowledgement.read_bytes() != original:
                 raise RelationViolationError(
                     f"Selected tool acknowledgement differs from {self.consumed}: {acknowledgement}"
                 )
-            state = AcknowledgedSlot
-        return SelectedToolSlot(
-            self.consumed.name, original.decode("ascii").removesuffix("\n"), state,
-        )
+            records.append(replace(consumed, state=AcknowledgedSlot))
+        return tuple(records)
 
 
 class SelectedToolSlots(Carry):
     opened_files = (CoordinationStore,)
 
-    def slots(self) -> tuple[LedgerSlot, ...]:
-        slots = []
+    def directories(self) -> tuple[Path, ...]:
         directories = {
             path for owner, root in SavedSessionFile.declarations_in(self.root)
             for path in root.glob(str(owner.pattern().parent / "selected-tool-ledger"))
         }
-        for directory in sorted(directories):
+        for directory in directories:
             self.require_confined(self.root, directory)
+        return tuple(sorted(directories))
+
+    def slots(self) -> tuple[LedgerSlot, ...]:
+        slots = []
+        for directory in self.directories():
             files = frozenset(directory.iterdir())
             for path in files:
                 self.require_confined(self.root, path)
@@ -80,9 +84,10 @@ def input_files(self) -> tuple[Path, ...]:
 
     def carry(self, *, apply: bool) -> Carried:
         slots = self.slots()
-        if not slots:
+        directories = self.directories()
+        if not directories:
             return Carried(0)
-        rows = tuple(slot.read() for slot in slots)
+        rows = tuple(row for slot in slots for row in slot.read())
         files = tuple(path for slot in slots for path in slot.files)
         acknowledged = sum(len(slot.acknowledgements) for slot in slots)
         detail = {
@@ -90,19 +95,20 @@ def carry(self, *, apply: bool) -> Carried:
             "consumed_rows": len(slots) - acknowledged, "acknowledged_rows": acknowledged,
             "slot_rows": len(slots), "deleted_files": len(files) if apply else 0,
         }
+        changed = len(files) + len(directories) - len({slot.consumed.parent for slot in slots})
         if not apply:
-            return Carried(len(files), detail)
+            return Carried(changed, detail)
         path = CoordinationStore.path_in(self.root)
         self.require_confined(self.root, path)
         HeldBusFile(CoordinationStore, path).require()
-        with closing(SQLiteFile.writer(path)) as db, SQLiteFile.write_transaction(db):
-            SelectedToolSlot.create(db, if_not_exists=True)
-            for row in rows:
-                row.carry(db)
+        if rows:
+            with closing(SQLiteFile.writer(path)) as db, SQLiteFile.write_transaction(db):
+                SelectedToolSlot.create(db, if_not_exists=True)
+                for row in rows:
+                    row.carry(db)
         # Every row is committed before the first unlink. Persist removal of
         # acknowledgements before removing consumed files: a killed carry can
         # leave a consumed file, but can never manufacture an orphan .done.
-        directories = tuple(dict.fromkeys(slot.consumed.parent for slot in slots))
         for slot in slots:
             for acknowledgement in slot.acknowledgements:
                 acknowledgement.unlink()
@@ -113,5 +119,5 @@ def carry(self, *, apply: bool) -> Carried:
         for directory in directories:
             directory.rmdir()
             Platform.current().sync_directory(directory.parent)
-        return Carried(len(files), detail,
+        return Carried(changed, detail,
                        (path.relative_to(self.root), *(file.relative_to(self.root) for file in files)))

From 3af1369990499272d2a5772810b36428f96c77fe Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:12:11 -0400
Subject: [PATCH 08/13] Format the selected tool carry

---
 tools/cutover/carry_selected_tool_slots.py | 40 +++++++++++++++-------
 1 file changed, 28 insertions(+), 12 deletions(-)

diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index a28c00e3f6..5c23a70281 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -38,7 +38,9 @@ def files(self) -> tuple[Path, ...]:
     def read(self) -> tuple[SelectedToolSlot, ...]:
         original = self.consumed.read_bytes()
         consumed = SelectedToolSlot(
-            self.consumed.name, original.decode("ascii").removesuffix("\n"), ConsumedSlot,
+            self.consumed.name,
+            original.decode("ascii").removesuffix("\n"),
+            ConsumedSlot,
         )
         records = [consumed]
         for acknowledgement in self.acknowledgements:
@@ -55,7 +57,8 @@ class SelectedToolSlots(Carry):
 
     def directories(self) -> tuple[Path, ...]:
         directories = {
-            path for owner, root in SavedSessionFile.declarations_in(self.root)
+            path
+            for owner, root in SavedSessionFile.declarations_in(self.root)
             for path in root.glob(str(owner.pattern().parent / "selected-tool-ledger"))
         }
         for directory in directories:
@@ -69,13 +72,20 @@ def slots(self) -> tuple[LedgerSlot, ...]:
             for path in files:
                 self.require_confined(self.root, path)
                 if not path.is_file():
-                    raise RelationViolationError(f"Selected tool ledger entry is not a file: {path}")
+                    raise RelationViolationError(
+                        f"Selected tool ledger entry is not a file: {path}"
+                    )
                 if path.suffix == ".done" and path.with_suffix("") not in files:
-                    raise RelationViolationError(f"Selected tool acknowledgement has no consumed file: {path}")
+                    raise RelationViolationError(
+                        f"Selected tool acknowledgement has no consumed file: {path}"
+                    )
             slots.extend(
-                LedgerSlot(path, tuple(ack for ack in (path.with_name(path.name + ".done"),)
-                                       if ack in files))
-                for path in sorted(files) if path.suffix != ".done"
+                LedgerSlot(
+                    path,
+                    tuple(ack for ack in (path.with_name(path.name + ".done"),) if ack in files),
+                )
+                for path in sorted(files)
+                if path.suffix != ".done"
             )
         return tuple(slots)
 
@@ -91,9 +101,12 @@ def carry(self, *, apply: bool) -> Carried:
         files = tuple(path for slot in slots for path in slot.files)
         acknowledged = sum(len(slot.acknowledgements) for slot in slots)
         detail = {
-            "consumed_files": len(slots), "acknowledgement_files": acknowledged,
-            "consumed_rows": len(slots) - acknowledged, "acknowledged_rows": acknowledged,
-            "slot_rows": len(slots), "deleted_files": len(files) if apply else 0,
+            "consumed_files": len(slots),
+            "acknowledgement_files": acknowledged,
+            "consumed_rows": len(slots) - acknowledged,
+            "acknowledged_rows": acknowledged,
+            "slot_rows": len(slots),
+            "deleted_files": len(files) if apply else 0,
         }
         changed = len(files) + len(directories) - len({slot.consumed.parent for slot in slots})
         if not apply:
@@ -119,5 +132,8 @@ def carry(self, *, apply: bool) -> Carried:
         for directory in directories:
             directory.rmdir()
             Platform.current().sync_directory(directory.parent)
-        return Carried(changed, detail,
-                       (path.relative_to(self.root), *(file.relative_to(self.root) for file in files)))
+        return Carried(
+            changed,
+            detail,
+            (path.relative_to(self.root), *(file.relative_to(self.root) for file in files)),
+        )

From 239181152fdd906f91bcf83e964f5a9d325bbd40 Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:17:38 -0400
Subject: [PATCH 09/13] Derive consumed files from one acknowledgement
 selection

---
 tools/cutover/carry_selected_tool_slots.py | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index 5c23a70281..7c71ebccde 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -75,7 +75,10 @@ def slots(self) -> tuple[LedgerSlot, ...]:
                     raise RelationViolationError(
                         f"Selected tool ledger entry is not a file: {path}"
                     )
-                if path.suffix == ".done" and path.with_suffix("") not in files:
+            acknowledgements = frozenset(path for path in files if path.match("*.done"))
+            consumed = files - acknowledgements
+            for path in acknowledgements:
+                if path.with_suffix("") not in consumed:
                     raise RelationViolationError(
                         f"Selected tool acknowledgement has no consumed file: {path}"
                     )
@@ -84,8 +87,7 @@ def slots(self) -> tuple[LedgerSlot, ...]:
                     path,
                     tuple(ack for ack in (path.with_name(path.name + ".done"),) if ack in files),
                 )
-                for path in sorted(files)
-                if path.suffix != ".done"
+                for path in sorted(consumed)
             )
         return tuple(slots)
 
@@ -100,7 +102,7 @@ def carry(self, *, apply: bool) -> Carried:
         rows = tuple(row for slot in slots for row in slot.read())
         files = tuple(path for slot in slots for path in slot.files)
         acknowledged = sum(len(slot.acknowledgements) for slot in slots)
-        detail = {
+        detail: dict[str, int | bool] = {
             "consumed_files": len(slots),
             "acknowledgement_files": acknowledged,
             "consumed_rows": len(slots) - acknowledged,

From 3a08309d084b72fc722cfc749afafa2ea3042690 Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:19:03 -0400
Subject: [PATCH 10/13] Keep slot predicates typed and take the coordination
 root from the launch

---
 src/agent_comms/native_runtime_input.py  |  7 ++++---
 src/agent_comms/native_tool_transport.py |  6 ++++--
 src/agent_comms/selected_tool_broker.py  | 10 ++++------
 tests/test_channel_coding_tools.py       |  2 +-
 tests/test_native_argument_refusal.py    |  2 +-
 tests/test_selected_tool_broker.py       |  2 +-
 6 files changed, 15 insertions(+), 14 deletions(-)

diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py
index 50c3445c4e..7569888204 100644
--- a/src/agent_comms/native_runtime_input.py
+++ b/src/agent_comms/native_runtime_input.py
@@ -389,9 +389,10 @@ def require_call(self, call_id: str) -> None:
     def acknowledge(self, db: sqlite3.Connection, call_id: str) -> None:
         self.require_call(call_id)
         state = self.state.acknowledge()
-        if self.update(db, where="input_id=? AND state=?", parameters=(
-            self.input_id, self.state.declared_name,
-        ), state=state).rowcount != 1:
+        where, parameters = self._condition(
+            "1", (), {"input_id": self.input_id, "state": self.state},
+        )
+        if self.update(db, where=where, parameters=parameters, state=state).rowcount != 1:
             raise SelectedToolDenied("Selected tool terminal changed; outcome UNKNOWN; no retry")
 
     def verify(self, call_id: str) -> None:
diff --git a/src/agent_comms/native_tool_transport.py b/src/agent_comms/native_tool_transport.py
index af33eee5c5..ac50ab0c78 100644
--- a/src/agent_comms/native_tool_transport.py
+++ b/src/agent_comms/native_tool_transport.py
@@ -31,6 +31,7 @@
 from .private_path import PrivateDirectoryRole, PrivateSocketRole
 
 if TYPE_CHECKING:
+    from .backend import TurnSession
     from .pi_events import ToolExecutionEnd, ToolExecutionStart
 
 _LOG = logging.getLogger(__name__)
@@ -263,13 +264,14 @@ class ToolTransport(Generic[Call]):
     """A child-bound request policy. An instance always has its authenticated PID."""
 
     max_request = 1024 * 1024
+    turn: TurnSession
 
     @property
     def retainable(self) -> bool:
         return True
 
     @classmethod
-    async def open(cls, turn, ownership, *policy) -> UnboundToolTransport:
+    async def open(cls, turn: TurnSession, ownership, *policy) -> UnboundToolTransport:
         listener = ToolListener(turn.launch.session.directory, os.urandom(32).hex())
         ownership.push_async_callback(listener.close)
 
@@ -321,7 +323,7 @@ def tool_finished(self, event: ToolExecutionEnd, input_id: str) -> None:
         call.finish(
             FieldCodec.encode(event.tool_name),
             FieldCodec.encode(event.is_error),
-            CoordinationStore.path_in(self.turn.launch.session.root),
+            CoordinationStore.path_in(self.turn.launch.root),
             input_id,
             identity.round_number,
         )
diff --git a/src/agent_comms/selected_tool_broker.py b/src/agent_comms/selected_tool_broker.py
index caf9811d42..edc42da669 100644
--- a/src/agent_comms/selected_tool_broker.py
+++ b/src/agent_comms/selected_tool_broker.py
@@ -18,17 +18,15 @@
 from pathlib import Path
 from typing import ClassVar
 
-from agent_comms.coordinator import Coordination
-from .coordination_session import ReadNativeCoordinationSession
-
 from .comms import Comms
-from agent_comms.native_runtime_input import NativeRuntimeTable
 from .coordination_errors import CoordinationReadUnavailable
+from .coordination_session import ReadNativeCoordinationSession
 from .coordination_tables.participants import OwnerGenerations
+from .coordinator import Coordination
 from .envelope_claim_transitions import ExistingFileClaim, WakeClaim
 from .field_codec import FieldCodec, Id128Text, MalformedRecord
 from .native_input_record import FullNativeExecution, NativeInputIdentity
-from .native_runtime_input import NativeRuntimeInput, SelectedToolSlot
+from .native_runtime_input import NativeRuntimeInput, NativeRuntimeTable, SelectedToolSlot
 from .native_tool_call import (
     CoordinationBusy,
     NativeToolCall,
@@ -189,7 +187,7 @@ def consume_selected_slot(store: Coordination, input_id: str, call_id: str) -> N
 def record_selected_terminal(coordination_path: Path, input_id: str, call_id: str) -> None:
     """Only a committed terminal acknowledgement lets the tool return success."""
     try:
-        with Coordination(coordination_path) as store, store.session.writing() as db:
+        with Coordination(str(coordination_path)) as store, store.session.writing() as db:
             SelectedToolSlot.require(db, input_id=input_id).acknowledge(db, call_id)
     except sqlite3.Error as error:
         raise SelectedToolDenied("Selected tool terminal acknowledgement UNKNOWN; no retry") from error
diff --git a/tests/test_channel_coding_tools.py b/tests/test_channel_coding_tools.py
index b97bb57ed6..64ed5e47e1 100644
--- a/tests/test_channel_coding_tools.py
+++ b/tests/test_channel_coding_tools.py
@@ -224,7 +224,7 @@ async def allow(self, turn, round_number, call):
 
     owner = Owner()
     socket = CodingToolSocket(tool_listener(tmp_path, secrets.token_hex(32)), owner, pid=os.getpid())
-    socket.turn = SimpleNamespace(launch=SimpleNamespace(session=SimpleNamespace(root=tmp_path)))
+    socket.turn = SimpleNamespace(launch=SimpleNamespace(root=tmp_path))
     socket.announce((ToolCallContent(id="late", name="read", arguments={"path": "a"}),))
     socket.tool_started(ToolExecutionStart(
         tool_call_id=ReportedText("late"), tool_name=ReportedText("read"), args={"path": "a"}
diff --git a/tests/test_native_argument_refusal.py b/tests/test_native_argument_refusal.py
index 599daa6427..1fcc3be716 100644
--- a/tests/test_native_argument_refusal.py
+++ b/tests/test_native_argument_refusal.py
@@ -32,7 +32,7 @@ async def allow(self, round_number, call):
             raise AssertionError("invalid arguments must never acquire authority")
 
     socket = CodingToolSocket(tool_listener(tmp_path, "a" * 64), Owner(), pid=os.getpid())
-    socket.turn = SimpleNamespace(launch=SimpleNamespace(session=SimpleNamespace(root=tmp_path)))
+    socket.turn = SimpleNamespace(launch=SimpleNamespace(root=tmp_path))
     args = {"command": False}
     raw = (json.dumps({"token": socket.listener.token, "call_id": "bad", "name": "bash",
                        "arguments": {"command": "false"}}) + "\n").encode()
diff --git a/tests/test_selected_tool_broker.py b/tests/test_selected_tool_broker.py
index 84548670dc..421653afc3 100644
--- a/tests/test_selected_tool_broker.py
+++ b/tests/test_selected_tool_broker.py
@@ -230,7 +230,7 @@ def accept(request: broker.SelectedToolRequest) -> None:
         observed.append(request)
 
     server = broker.SelectedToolSocket(tool_listener(tmp_path, token), accept, pid=os.getpid())
-    server.turn = SimpleNamespace(launch=SimpleNamespace(session=SimpleNamespace(root=tmp_path)))
+    server.turn = SimpleNamespace(launch=SimpleNamespace(root=tmp_path))
     assert server.listener.path != previous.path
     assert server.listener.path.parent.stat().st_mode & 0o777 == 0o700
     await server.start()

From 44d53c76cd118e82f2913c65b8417d996a3c76e1 Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:21:39 -0400
Subject: [PATCH 11/13] Keep legacy slot interpretation in the carry and open
 test stores in their worker

---
 src/agent_comms/native_runtime_input.py    | 20 -----
 tests/test_selected_tool_broker.py         |  5 +-
 tools/cutover/carry_selected_tool_slots.py | 92 +++++++++++++++-------
 3 files changed, 66 insertions(+), 51 deletions(-)

diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py
index 7569888204..ec5e2c0d7c 100644
--- a/src/agent_comms/native_runtime_input.py
+++ b/src/agent_comms/native_runtime_input.py
@@ -344,10 +344,6 @@ def acknowledge(cls) -> type[SelectedSlotState]:
     def require_acknowledged(cls) -> None:
         raise SelectedToolDenied("Selected tool outcome UNKNOWN; no terminal acknowledgement")
 
-    @classmethod
-    def carried(cls, incoming: type[SelectedSlotState]) -> type[SelectedSlotState]:
-        return incoming
-
 
 class ConsumedSlot(SelectedSlotState):
     @classmethod
@@ -360,12 +356,6 @@ class AcknowledgedSlot(SelectedSlotState):
     def require_acknowledged(cls) -> None:
         pass
 
-    @classmethod
-    def carried(cls, incoming: type[SelectedSlotState]) -> type[SelectedSlotState]:
-        # A carry interrupted after deleting .done files must not erase the
-        # acknowledgement already committed in this store.
-        return cls
-
 
 @dataclass(frozen=True)
 class SelectedToolSlot(NativeRuntimeTable, WithoutRowid):
@@ -399,16 +389,6 @@ def verify(self, call_id: str) -> None:
         self.require_call(call_id)
         self.state.require_acknowledged()
 
-    def carry(self, db: sqlite3.Connection) -> None:
-        """Transfer a legacy slot, preserving an earlier committed acknowledgement."""
-        if self.insert_if_absent(db).rowcount == 1:
-            return
-        current = self.require(db, input_id=self.input_id)
-        current.require_call(self.call_id)
-        state = current.state.carried(self.state)
-        if state is not current.state:
-            current.acknowledge(db, self.call_id)
-
 
 @dataclass(frozen=True)
 class CurrentNativeCursor(
diff --git a/tests/test_selected_tool_broker.py b/tests/test_selected_tool_broker.py
index 421653afc3..f36856797b 100644
--- a/tests/test_selected_tool_broker.py
+++ b/tests/test_selected_tool_broker.py
@@ -225,8 +225,9 @@ async def test_listener_isolated_from_previous_turn_and_keeps_consumed_slot(
     observed: list[broker.SelectedToolRequest] = []
 
     def accept(request: broker.SelectedToolRequest) -> None:
-        broker.consume_selected_slot(slot_store, input_id, request.call_id)
-        broker.record_selected_terminal(slot_store.session.path, input_id, request.call_id)
+        with Coordination(slot_store.session.path) as store:
+            broker.consume_selected_slot(store, input_id, request.call_id)
+            broker.record_selected_terminal(store.session.path, input_id, request.call_id)
         observed.append(request)
 
     server = broker.SelectedToolSocket(tool_listener(tmp_path, token), accept, pid=os.getpid())
diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index 7c71ebccde..b0193f49a3 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -6,9 +6,11 @@
 
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
 from contextlib import closing
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from pathlib import Path
+from typing import ClassVar
 
 from agent_comms.child_process import Platform
 from agent_comms.coordination_database import CoordinationStore
@@ -18,6 +20,7 @@
 from agent_comms.native_runtime_input import (
     AcknowledgedSlot,
     ConsumedSlot,
+    SelectedSlotState,
     SelectedToolSlot,
 )
 from agent_comms.native_session_files import SavedSessionFile
@@ -25,31 +28,63 @@
 
 
 @dataclass(frozen=True)
-class LedgerSlot:
-    """One legacy consumed file and its optional matching acknowledgement."""
+class LedgerSlot(ABC):
+    """One legacy slot; its file kind owns validation and the declared state."""
 
     consumed: Path
-    acknowledgements: tuple[Path, ...]
+
+    @property
+    @abstractmethod
+    def state(self) -> type[SelectedSlotState]: ...
+
+    @classmethod
+    def at(cls, consumed: Path, acknowledgements: frozenset[Path]) -> LedgerSlot:
+        acknowledgement = consumed.with_name(consumed.name + ".done")
+        if acknowledgement in acknowledgements:
+            return AcknowledgedLedgerSlot(consumed, acknowledgement)
+        return ConsumedLedgerSlot(consumed)
 
     @property
     def files(self) -> tuple[Path, ...]:
-        return (self.consumed, *self.acknowledgements)
+        return (self.consumed,)
+
+    def require_acknowledgement(self, original: bytes) -> None:
+        """A consumed file does not claim a terminal acknowledgement."""
+
+    def retire_acknowledgement(self) -> None:
+        """A consumed file has no acknowledgement file to retire."""
 
-    def read(self) -> tuple[SelectedToolSlot, ...]:
+    def read(self) -> SelectedToolSlot:
         original = self.consumed.read_bytes()
-        consumed = SelectedToolSlot(
+        self.require_acknowledgement(original)
+        return SelectedToolSlot(
             self.consumed.name,
             original.decode("ascii").removesuffix("\n"),
-            ConsumedSlot,
+            self.state,
         )
-        records = [consumed]
-        for acknowledgement in self.acknowledgements:
-            if acknowledgement.read_bytes() != original:
-                raise RelationViolationError(
-                    f"Selected tool acknowledgement differs from {self.consumed}: {acknowledgement}"
-                )
-            records.append(replace(consumed, state=AcknowledgedSlot))
-        return tuple(records)
+
+
+class ConsumedLedgerSlot(LedgerSlot):
+    state: ClassVar[type[SelectedSlotState]] = ConsumedSlot
+
+
+@dataclass(frozen=True)
+class AcknowledgedLedgerSlot(LedgerSlot):
+    acknowledgement: Path
+    state: ClassVar[type[SelectedSlotState]] = AcknowledgedSlot
+
+    @property
+    def files(self) -> tuple[Path, ...]:
+        return (self.consumed, self.acknowledgement)
+
+    def require_acknowledgement(self, original: bytes) -> None:
+        if self.acknowledgement.read_bytes() != original:
+            raise RelationViolationError(
+                f"Selected tool acknowledgement differs from {self.consumed}: {self.acknowledgement}"
+            )
+
+    def retire_acknowledgement(self) -> None:
+        self.acknowledgement.unlink()
 
 
 class SelectedToolSlots(Carry):
@@ -82,13 +117,7 @@ def slots(self) -> tuple[LedgerSlot, ...]:
                     raise RelationViolationError(
                         f"Selected tool acknowledgement has no consumed file: {path}"
                     )
-            slots.extend(
-                LedgerSlot(
-                    path,
-                    tuple(ack for ack in (path.with_name(path.name + ".done"),) if ack in files),
-                )
-                for path in sorted(consumed)
-            )
+            slots.extend(LedgerSlot.at(path, acknowledgements) for path in sorted(consumed))
         return tuple(slots)
 
     def input_files(self) -> tuple[Path, ...]:
@@ -99,9 +128,9 @@ def carry(self, *, apply: bool) -> Carried:
         directories = self.directories()
         if not directories:
             return Carried(0)
-        rows = tuple(row for slot in slots for row in slot.read())
+        rows = tuple(slot.read() for slot in slots)
         files = tuple(path for slot in slots for path in slot.files)
-        acknowledged = sum(len(slot.acknowledgements) for slot in slots)
+        acknowledged = len(files) - len(slots)
         detail: dict[str, int | bool] = {
             "consumed_files": len(slots),
             "acknowledgement_files": acknowledged,
@@ -119,14 +148,19 @@ def carry(self, *, apply: bool) -> Carried:
         if rows:
             with closing(SQLiteFile.writer(path)) as db, SQLiteFile.write_transaction(db):
                 SelectedToolSlot.create(db, if_not_exists=True)
-                for row in rows:
-                    row.carry(db)
+                for slot, row in zip(slots, rows, strict=True):
+                    if row.insert_if_absent(db).rowcount == 0:
+                        current = SelectedToolSlot.require(db, input_id=row.input_id)
+                        if current.call_id != row.call_id or current.state is not row.state:
+                            raise RelationViolationError(
+                                f"Selected tool slot differs from committed row: {slot.consumed}"
+                            )
+                        current.require_call(row.call_id)
         # Every row is committed before the first unlink. Persist removal of
         # acknowledgements before removing consumed files: a killed carry can
         # leave a consumed file, but can never manufacture an orphan .done.
         for slot in slots:
-            for acknowledgement in slot.acknowledgements:
-                acknowledgement.unlink()
+            slot.retire_acknowledgement()
         for directory in directories:
             Platform.current().sync_directory(directory)
         for slot in slots:

From a7b47fe108cff9001d2260756696dab359172734 Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:26:20 -0400
Subject: [PATCH 12/13] Share ledger file reading and retirement through the
 declared file family

---
 tools/cutover/carry_selected_tool_slots.py | 22 +++++++++++-----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index b0193f49a3..c6b3f3e1b6 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -48,15 +48,15 @@ def at(cls, consumed: Path, acknowledgements: frozenset[Path]) -> LedgerSlot:
     def files(self) -> tuple[Path, ...]:
         return (self.consumed,)
 
-    def require_acknowledgement(self, original: bytes) -> None:
-        """A consumed file does not claim a terminal acknowledgement."""
+    def contents(self) -> bytes:
+        return self.consumed.read_bytes()
 
     def retire_acknowledgement(self) -> None:
-        """A consumed file has no acknowledgement file to retire."""
+        for acknowledgement in self.files[1:]:
+            acknowledgement.unlink()
 
     def read(self) -> SelectedToolSlot:
-        original = self.consumed.read_bytes()
-        self.require_acknowledgement(original)
+        original = self.contents()
         return SelectedToolSlot(
             self.consumed.name,
             original.decode("ascii").removesuffix("\n"),
@@ -77,14 +77,14 @@ class AcknowledgedLedgerSlot(LedgerSlot):
     def files(self) -> tuple[Path, ...]:
         return (self.consumed, self.acknowledgement)
 
-    def require_acknowledgement(self, original: bytes) -> None:
+    def contents(self) -> bytes:
+        original = super().contents()
         if self.acknowledgement.read_bytes() != original:
             raise RelationViolationError(
-                f"Selected tool acknowledgement differs from {self.consumed}: {self.acknowledgement}"
+                f"Selected tool acknowledgement differs from {self.consumed}: "
+                f"{self.acknowledgement}"
             )
-
-    def retire_acknowledgement(self) -> None:
-        self.acknowledgement.unlink()
+        return original
 
 
 class SelectedToolSlots(Carry):
@@ -101,7 +101,7 @@ def directories(self) -> tuple[Path, ...]:
         return tuple(sorted(directories))
 
     def slots(self) -> tuple[LedgerSlot, ...]:
-        slots = []
+        slots: list[LedgerSlot] = []
         for directory in self.directories():
             files = frozenset(directory.iterdir())
             for path in files:

From 81f6237143e290a19448b112b219d314a8c991aa Mon Sep 17 00:00:00 2001
From: Ratchet test <ratchet@example.invalid>
Date: Fri, 23 Oct 2026 08:29:17 -0400
Subject: [PATCH 13/13] Use the slot owner to check carried call identities

---
 tools/cutover/carry_selected_tool_slots.py | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
index c6b3f3e1b6..b603d005b6 100644
--- a/tools/cutover/carry_selected_tool_slots.py
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -24,6 +24,7 @@
     SelectedToolSlot,
 )
 from agent_comms.native_session_files import SavedSessionFile
+from agent_comms.native_tool_call import SelectedToolDenied
 from agent_comms.typed_table import HeldBusFile
 
 
@@ -151,11 +152,16 @@ def carry(self, *, apply: bool) -> Carried:
                 for slot, row in zip(slots, rows, strict=True):
                     if row.insert_if_absent(db).rowcount == 0:
                         current = SelectedToolSlot.require(db, input_id=row.input_id)
-                        if current.call_id != row.call_id or current.state is not row.state:
+                        try:
+                            current.require_call(row.call_id)
+                        except SelectedToolDenied as error:
                             raise RelationViolationError(
                                 f"Selected tool slot differs from committed row: {slot.consumed}"
+                            ) from error
+                        if current.state is not row.state:
+                            raise RelationViolationError(
+                                f"Selected tool state differs from committed row: {slot.consumed}"
                             )
-                        current.require_call(row.call_id)
         # Every row is committed before the first unlink. Persist removal of
         # acknowledgements before removing consumed files: a killed carry can
         # leave a consumed file, but can never manufacture an orphan .done.