Batch round r2: nine ready heads, merged and tested once

src/agent_comms/active_route.py

@@ -19,6 +19,7 @@
 from pathlib import Path
 from typing import Annotated, ClassVar
 
+from .child_process import Platform
 from .coordination_errors import CoordinationError
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
@@ -193,15 +194,13 @@ def read_active_route(path: Path | None = None) -> ActiveRoute | None:
         parent = path.parent.lstat()
         if (
             not stat.S_ISDIR(parent.st_mode)
-            or parent.st_uid != os.geteuid()
-            or stat.S_IMODE(parent.st_mode) != 0o700
+            or not Platform.current().is_private(parent, 0o700)
         ):
             raise ValueError("active comms route directory must be owner-only")
         info = os.fstat(fd)
         if (
             not stat.S_ISREG(info.st_mode)
-            or info.st_uid != os.geteuid()
-            or stat.S_IMODE(info.st_mode) != 0o600
+            or not Platform.current().is_private(info, 0o600)
             or info.st_nlink != 1
             or not 0 < info.st_size <= 4096
         ):
@@ -244,12 +243,11 @@ def guard_default_route_write(expected_root: Path, *, blocking: bool = True) ->
         parent = path.parent.lstat()
         if (
             not stat.S_ISDIR(info.st_mode)
-            or info.st_uid != os.geteuid()
+            or not Platform.current().is_owned(info)
             or (parent.st_dev, parent.st_ino) != (info.st_dev, info.st_ino)
         ):
             raise ValueError("active comms route directory changed or is not owned")
-        if stat.S_IMODE(info.st_mode) != 0o700:
-            os.fchmod(directory, 0o700)
+        Platform.current().make_private_descriptor(directory, 0o700)
         route = CommsRoute.current(path)
         # Local roots may be created under this guard. ActiveRoute owns the
         # stronger observation of an already published root and its identity.
@@ -451,10 +449,10 @@ def route_directory(path: Path) -> Iterator[int]:
         acquire_exclusive_route_lock(directory)
         info = os.fstat(directory)
         parent = path.parent.lstat()
-        if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid()
+        if (not stat.S_ISDIR(info.st_mode) or not Platform.current().is_owned(info)
                 or (parent.st_dev, parent.st_ino) != (info.st_dev, info.st_ino)):
             raise ValueError("active comms route directory changed or is not owned")
-        os.fchmod(directory, 0o700)
+        Platform.current().make_private_descriptor(directory, 0o700)
         yield directory
     finally:
         os.close(directory)
@@ -475,12 +473,12 @@ def _publish_active_route_locked(
         parent = path.parent.lstat()
         if (
             not stat.S_ISDIR(info.st_mode)
-            or info.st_uid != os.geteuid()
+            or not Platform.current().is_owned(info)
             or parent.st_dev != info.st_dev
             or parent.st_ino != info.st_ino
         ):
             raise ValueError("active comms route directory changed or is not owned")
-        os.fchmod(directory, 0o700)
+        Platform.current().make_private_descriptor(directory, 0o700)
         if backup is None and read_active_route(path) is not None:
             raise ValueError("active comms route is already installed")
         payload = (json.dumps(route.to_record(), sort_keys=True) + "\n").encode()

src/agent_comms/agent_events.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from dataclasses import asdict, dataclass, field, replace
+from dataclasses import dataclass, field, replace
 from functools import partial
 from typing import TYPE_CHECKING, Any, ClassVar
 
@@ -22,7 +22,7 @@
 )
 from .turn_terminal import FailedGoalTurn
 from .turn_goal_account import OriginGoalSettlement, VerifiedGoalSettlement
-from .diagnostics import FailureReason, terminal_failure_reason
+from .turn_failure import BackendFailed, TurnFailure
 from .child_process import ProcessIdentity
 from .declared_family import DeclaredFamily
 from .field_codec import NullMember
@@ -258,7 +258,7 @@ def phase(self) -> str:
 class Done(AgentEvent, FailedGoalTurn):
     text: str
     ok: bool
-    reason_code: str | None = None
+    failure: TurnFailure | None = None
     diagnostic: dict[str, Any] | None = None
     outcome: NativeTurnOutcome = field(default_factory=UnresolvedNativeTurnOutcome)
     # Immutable completion snapshot of TurnOutput's already decoded result.
@@ -272,8 +272,12 @@ def resolve(
         return replace(self, outcome=outcome)
 
     @property
-    def failure_reason(self) -> FailureReason:
-        return terminal_failure_reason(asdict(self))
+    def failure_kind(self) -> type[TurnFailure]:
+        return BackendFailed if self.failure is None else type(self.failure)
+
+    @property
+    def reason_code(self) -> str | None:
+        return None if self.ok else self.failure_kind.declared_name
 
     def failure_observation(self, account, permit, captured):
         return None if self.ok else super().failure_observation(account, permit, captured)

src/agent_comms/attempt_store.py

@@ -19,12 +19,7 @@
 from agent_comms.coordination_snapshot import RecoverySnapshot
 from agent_comms.coordination_tables.assignments import WakeAssignment
 from agent_comms.coordination_tables.attempts import AttemptRecord, ReplayFact
-from agent_comms.coordination_tables.recovery import (
-    ACPClientConnectivity,
-    ConnectivityFacet,
-    OwnerConnectivity,
-    RecoveryAudit,
-)
+from agent_comms.coordination_tables.recovery import RecoveryAudit
 from agent_comms.owner_fence import OwnerFence
 from agent_comms.turn_failure import TurnFailure
 from agent_comms.recovery_reader import RecoveryReader
@@ -208,34 +203,6 @@ def fail_unknown(
             )
             return Applied(self.get(fence.execution_id))
 
-    def observe_connectivity(
-        self,
-        fence: OwnerFence,
-        *,
-        expected_pointer_revision: int,
-        expected_revision: int | None,
-        owner: OwnerConnectivity,
-        acp_client: ACPClientConnectivity,
-    ) -> Applied[RecoverySnapshot]:
-        owner, acp_client = OwnerConnectivity(owner), ACPClientConnectivity(acp_client)
-        with self.session.writing() as db:
-            snapshot, _ = self.require_fence(fence)
-            before = snapshot.connectivity
-            if (
-                snapshot.pointer_revision != expected_pointer_revision
-                or (before.revision if before else None) != expected_revision
-            ):
-                raise StaleRevision("connectivity or pointer revision changed")
-            now = self.session.now(before.observed_at_ms if before else 0)
-            ConnectivityFacet(
-                fence.execution_id,
-                owner,
-                acp_client,
-                before.revision + 1 if before else 1,
-                now,
-            ).upsert(db)
-            return Applied(self.get(fence.execution_id))
-
     def append_recovery_audit(
         self,
         fence: OwnerFence,

src/agent_comms/backend.py

@@ -47,7 +47,6 @@
 from .acp_failure import UnrecordedFailureSource
 from .child_process import TimedOutOutcome
 from .coordinator import Coordination
-from .diagnostics import FailureReason
 from .extension_ui import ExtensionUiSession
 from .field_codec import Id128Text, MalformedRecord, boundary_decoder
 from .image_inputs import ImageInput
@@ -158,8 +157,8 @@ async def stream_agent_events(
                 session_file=session_file,
             ))
         except NativePiUnavailable as error:
-            yield events.Error(text=str(error), reason_code="native_launch_invalid")
-            yield events.Done(ok=False, reason_code="native_launch_invalid", text=str(error))
+            yield events.Error(text=str(error), reason_code=failures.NativeLaunchInvalid.declared_name)
+            yield events.Done(ok=False, failure=failures.NativeLaunchInvalid(str(error)), text=str(error))
             return
         from .session_fence import session_writer_fence
 
@@ -539,7 +538,6 @@ async def receive_record(self) -> AsyncIterator[events.AgentEvent]:
             return
         if not self.line:
             if self.awaiting_native_attestation:
-                self.output.preflight_failure = FailureReason.PREFLIGHT_EXIT
                 self.output.diagnostic = {
                     "elapsed_ms": round(
                         (self.loop.time() - self.watchdog.launch_started_at) * 1000
@@ -548,7 +546,7 @@ async def receive_record(self) -> AsyncIterator[events.AgentEvent]:
                 }
                 self.output.diagnostic.update(self.watchdog.session_load.diagnostic)
                 self.output.record_failure(
-                    failures.InputIdUnavailable(
+                    failures.NativePreflightExit(
                         "Pi native input-ID capability preflight ended before attestation."
                     )
                 )
@@ -586,7 +584,7 @@ async def invalidate_identity(self) -> tuple[events.AgentEvent, ...]:
         self.output.discard_text()
         updates = (self.context_info(), events.TurnState(
             state="failed",
-            reason_code="session_identity_uncertain",
+            reason_code=failures.IdentityUncertain.declared_name,
             elapsed_ms=0,
             phase=ShutdownPhase(),
         ))

src/agent_comms/bus_tables.py

@@ -26,6 +26,7 @@
 from agent_comms.field_codec import FieldCodec, WireName, WireOrder
 from agent_comms.message_reference import MessageReference
 from agent_comms.typed_table import (
+    TableReference,
     AppendOnlyTables,
     Column,
     ComposedSchemas,
@@ -303,7 +304,7 @@ class ClaimOperations(TypedTable, BusProjectionTables):
 
     @classmethod
     def references(cls) -> tuple[ForeignKey, ...]:
-        return (ForeignKey(("seq",), messages.Claimed.table, ("seq",)),)
+        return (ForeignKey(columns=("seq",), target=messages.Claimed.table, target_columns=("seq",)),)
 
     @classmethod
     def used(cls, db: sqlite3.Connection, operation_id: str) -> bool:
@@ -319,7 +320,7 @@ class HeldResources(TypedTable, BusProjectionTables):
 
     @classmethod
     def references(cls) -> tuple[ForeignKey, ...]:
-        return (ForeignKey(("claim_seq",), messages.Claimed.table, ("seq",)),)
+        return (ForeignKey(columns=("claim_seq",), target=messages.Claimed.table, target_columns=("seq",)),)
 
     @classmethod
     def owner_for(cls, db: sqlite3.Connection, resource: str) -> ResourceClaim:
@@ -429,7 +430,7 @@ def advance(self, removed: tuple[str, ...], projected: ClaimProjection) -> None:
 class CapabilityRow(TypedTable, BusTables, declared_name=None):
     """Every generated capability row belongs to one original bus sequence."""
 
-    seq: Annotated[int, Column(primary_key=True, references=(BusRow, "seq"))]
+    seq: Annotated[int, Column(primary_key=True, references=TableReference(BusRow, ("seq",)))]
 
 
 class _CapabilityTable:

src/agent_comms/child_process.py

@@ -37,6 +37,7 @@
 from dataclasses import dataclass
 from functools import partial
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING, Any, ClassVar, NoReturn
 
 from .declared_family import DeclaredFamily
@@ -428,9 +429,24 @@ def sync_directory(self, path: Path) -> None: ...
     @abstractmethod
     def is_private(self, info: os.stat_result, permissions: int) -> bool: ...
 
+    @abstractmethod
+    def is_owned(self, info: os.stat_result) -> bool: ...
+
     @abstractmethod
     def make_private(self, path: Path, permissions: int) -> None: ...
 
+    @abstractmethod
+    def make_private_descriptor(self, descriptor: int, permissions: int) -> None: ...
+
+    @contextmanager
+    def socket_address(self, path: Path, descriptor: int) -> Generator[Path, None, None]:
+        with TemporaryDirectory(prefix="ac-socket-", dir="/tmp") as directory:
+            parent = Path(directory) / "d"
+            parent.symlink_to(path.parent, target_is_directory=True)
+            if not os.path.samestat(os.stat(parent), os.fstat(descriptor)):
+                raise ValueError("Socket directory changed during address acquisition")
+            yield parent / path.name
+
     @abstractmethod
     def try_store_lock(self, descriptor: int, *, mode: LockMode) -> None:
         """One nonblocking native attempt; busy is BlockingIOError."""
@@ -507,13 +523,20 @@ def sync_directory(self, path: Path) -> None:
             os.close(descriptor)
 
     def is_private(self, info: os.stat_result, permissions: int) -> bool:
-        return stat.S_IMODE(info.st_mode) == permissions and info.st_uid == os.geteuid()
+        return stat.S_IMODE(info.st_mode) == permissions and self.is_owned(info)
+
+    def is_owned(self, info: os.stat_result) -> bool:
+        return info.st_uid == os.geteuid()
 
     def make_private(self, path: Path, permissions: int) -> None:
         with suppress(FileNotFoundError):
             if stat.S_IMODE(path.lstat().st_mode) != permissions:
                 os.chmod(path, permissions, follow_symlinks=False)
 
+    def make_private_descriptor(self, descriptor: int, permissions: int) -> None:
+        if stat.S_IMODE(os.fstat(descriptor).st_mode) != permissions:
+            os.fchmod(descriptor, permissions)
+
     def acquire_store_lock(self, descriptor: int, *, mode: LockMode, blocking: bool) -> None:
         import fcntl
 
@@ -559,6 +582,10 @@ class LinuxProcessRecord:
 
 
 class LinuxPlatform(NamespaceContainment, PidfdHandles, PosixPlatform):
+    @contextmanager
+    def socket_address(self, path: Path, descriptor: int) -> Generator[Path, None, None]:
+        yield Path(f"/proc/{os.getpid()}/fd/{descriptor}") / path.name
+
     def parent_death_binding(self) -> int:
         """The exec gate arms PR_SET_PDEATHSIG and binds to this launcher's PID."""
         return os.getpid()
@@ -728,9 +755,15 @@ def is_private(self, info: os.stat_result, permissions: int) -> bool:
         # NTFS ACLs, not POSIX mode bits, determine privacy.
         return True
 
+    def is_owned(self, info: os.stat_result) -> bool:
+        return True
+
     def make_private(self, path: Path, permissions: int) -> None:
         pass
 
+    def make_private_descriptor(self, descriptor: int, permissions: int) -> None:
+        pass
+
     def try_store_lock(self, descriptor: int, *, mode: LockMode) -> None:
         import errno
         import msvcrt
@@ -1858,24 +1891,24 @@ def __init__(self, process: subprocess.Popen[bytes], identity: ProcessIdentity,
         self.process = process
 
     @classmethod
-    def launch(
+    @contextmanager
+    def launching(
         cls,
         command: tuple[str, ...],
         *,
         cwd: str | Path | None = None,
         env: Mapping[str, str] = INHERITED_ENVIRONMENT,
         output: Any = subprocess.DEVNULL,
-        before_start: Callable[[ProcessIdentity], None] | None = None,
         runtime: LaunchRuntime = LaunchRuntime.running(),
         lifetime: type[ChildLifetime] = DiesWithLauncherLifetime,
-    ) -> SpawnedProcess:
+    ) -> Generator[SpawnedProcess]:
+        """Yield the child before it executes; leaving the body releases it, raising reaps it."""
         if not command:
             raise ValueError("A child command is required")
         with Platform.current().launch(command, (), runtime, lifetime) as launch:
             child = launch.spawn(cwd=cwd, env=env, stdout=output, stderr=output)
             try:
-                if before_start is not None:
-                    before_start(child.identity)
+                yield child
             except BaseException:
                 launch.cancel_before_release(child.identity)
                 child.reap()
@@ -1887,7 +1920,12 @@ def launch(
             except BaseException:
                 child.close()
                 raise
-        return child
+
+    @classmethod
+    def launch(cls, command: tuple[str, ...], **options: Any) -> SpawnedProcess:
+        """Launch with nothing to reserve before the child executes."""
+        with cls.launching(command, **options) as child:
+            return child
 
     def collect(self, *, timeout: float) -> subprocess.CompletedProcess[bytes]:
         """Collect within the caller's owned context; its exit retires descendants."""