Move selected tool consumption into the input coordination store

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:

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,

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

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