diff --git a/src/agent_comms/channel_coding_tools.py b/src/agent_comms/channel_coding_tools.py
index 81f44818bf..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(turn.launch.session.directory, slot, slot)
+            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 4d6d23623c..ec5e2c0d7c 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,
@@ -331,6 +333,63 @@ 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(NativeRuntimeTable, WithoutRowid):
+    """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()]
+    state: type[SelectedSlotState] = ConsumedSlot
+
+    @classmethod
+    def consume(
+        cls, db: sqlite3.Connection, input_id: str, call_id: str,
+    ) -> None:
+        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:
+        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()
+        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:
+        self.require_call(call_id)
+        self.state.require_acknowledged()
+
+
 @dataclass(frozen=True)
 class CurrentNativeCursor(
     MonotoneColumns,
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..ac50ab0c78 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 (
@@ -30,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__)
@@ -262,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)
 
@@ -320,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),
-            self.listener.session_directory,
+            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 dbf814fecd..edc42da669 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,17 +18,15 @@
 from pathlib import Path
 from typing import ClassVar
 
-from agent_comms.child_process import Platform
-from agent_comms.coordinator import Coordination
-
 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
+from .native_runtime_input import NativeRuntimeInput, NativeRuntimeTable, SelectedToolSlot
 from .native_tool_call import (
     CoordinationBusy,
     NativeToolCall,
@@ -41,7 +39,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
@@ -73,7 +70,6 @@ def mode(self, owner):
             owner.store,
             owner.resource_scope,
             owner.owner_name,
-            owner.session_dir,
             owner.input_id,
         )
 
@@ -136,11 +132,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 +169,27 @@ 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) -> 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)
+    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(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
 
 
@@ -234,7 +198,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.
@@ -257,7 +220,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)
@@ -306,18 +269,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(
@@ -325,12 +284,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(session_dir, input_id, request.call_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(
@@ -339,7 +297,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]):
diff --git a/tests/test_channel_coding_tools.py b/tests/test_channel_coding_tools.py
index 423539392e..64ed5e47e1 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(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..1fcc3be716 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(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..f36856797b 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,13 @@ 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)
+        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())
+    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()
@@ -255,7 +269,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()
diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py
new file mode 100644
index 0000000000..b603d005b6
--- /dev/null
+++ b/tools/cutover/carry_selected_tool_slots.py
@@ -0,0 +1,181 @@
+"""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 abc import ABC, abstractmethod
+from contextlib import closing
+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
+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,
+    SelectedSlotState,
+    SelectedToolSlot,
+)
+from agent_comms.native_session_files import SavedSessionFile
+from agent_comms.native_tool_call import SelectedToolDenied
+from agent_comms.typed_table import HeldBusFile
+
+
+@dataclass(frozen=True)
+class LedgerSlot(ABC):
+    """One legacy slot; its file kind owns validation and the declared state."""
+
+    consumed: 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,)
+
+    def contents(self) -> bytes:
+        return self.consumed.read_bytes()
+
+    def retire_acknowledgement(self) -> None:
+        for acknowledgement in self.files[1:]:
+            acknowledgement.unlink()
+
+    def read(self) -> SelectedToolSlot:
+        original = self.contents()
+        return SelectedToolSlot(
+            self.consumed.name,
+            original.decode("ascii").removesuffix("\n"),
+            self.state,
+        )
+
+
+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 contents(self) -> bytes:
+        original = super().contents()
+        if self.acknowledgement.read_bytes() != original:
+            raise RelationViolationError(
+                f"Selected tool acknowledgement differs from {self.consumed}: "
+                f"{self.acknowledgement}"
+            )
+        return original
+
+
+class SelectedToolSlots(Carry):
+    opened_files = (CoordinationStore,)
+
+    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 directories:
+            self.require_confined(self.root, directory)
+        return tuple(sorted(directories))
+
+    def slots(self) -> tuple[LedgerSlot, ...]:
+        slots: list[LedgerSlot] = []
+        for directory in self.directories():
+            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}"
+                    )
+            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}"
+                    )
+            slots.extend(LedgerSlot.at(path, acknowledgements) for path in sorted(consumed))
+        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()
+        directories = self.directories()
+        if not directories:
+            return Carried(0)
+        rows = tuple(slot.read() for slot in slots)
+        files = tuple(path for slot in slots for path in slot.files)
+        acknowledged = len(files) - len(slots)
+        detail: dict[str, int | bool] = {
+            "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:
+            return Carried(changed, detail)
+        path = CoordinationStore.path_in(self.root)
+        self.require_confined(self.root, path)
+        HeldBusFile(CoordinationStore, path).require()
+        if rows:
+            with closing(SQLiteFile.writer(path)) as db, SQLiteFile.write_transaction(db):
+                SelectedToolSlot.create(db, if_not_exists=True)
+                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)
+                        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}"
+                            )
+        # 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:
+            slot.retire_acknowledgement()
+        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(
+            changed,
+            detail,
+            (path.relative_to(self.root), *(file.relative_to(self.root) for file in files)),
+        )