Make declared failures own their reasons

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/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
 
@@ -545,7 +544,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
@@ -554,7 +552,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."
                     )
                 )
@@ -592,7 +590,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/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

src/agent_comms/diagnostics.py

@@ -7,17 +7,18 @@
 import os
 import time
 from contextlib import contextmanager
+from collections.abc import Mapping
 from dataclasses import dataclass, field, replace
-from enum import StrEnum
 from pathlib import Path
 from traceback import TracebackException
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING
 from uuid import uuid4
 
 from .child_process import Platform
-from .field_codec import FieldCodec, Id128Text, MalformedRecord
+from .field_codec import FieldCodec, Id128Text
 from .locked_store import DurableFile, PatternFile
 from .store_files import _atomic_write_text
+from .turn_failure import TurnFailure
 
 if TYPE_CHECKING:
     from .activity import DrainDiagnostic
@@ -43,20 +44,6 @@ def owner_process_output(root: Path, thread: Thread):
         yield output
 
 
-class FailureReason(StrEnum):
-    BACKEND_FAILED = "backend_failed"
-    PREFLIGHT_TIMEOUT = "native_preflight_timeout"
-    PREFLIGHT_EXIT = "native_preflight_exit"
-    INPUT_ID_UNAVAILABLE = "pi_input_id_unavailable"
-    IDENTITY_UNCERTAIN = "session_identity_uncertain"
-    AUTHORITY_CHANGED = "input_authority_changed"
-    FOLLOWUP_UNRECOGNIZED = "unrecognized_followup_input"
-    INPUT_MISSING = "current_prompt_input_missing"
-    MODEL_REQUEST_FAILED = "model_request_failed"
-    FINAL_STOP_MISSING = "assistant_final_stop_missing"
-    QUEUED_INPUT_MISSING = "queued_input_start_missing"
-
-
 @dataclass
 class PublicationMeasurements:
     """Bounded transport timing counters, not a message or phase authority."""
@@ -133,34 +120,20 @@ def record(cls, root, lease, record):
             output.write(json.dumps(record) + "\n")
 
 
-def terminal_failure_reason(event: dict) -> FailureReason:
-    """Allowlisted backend terminal classification; never diagnostic prose."""
-    measurements = event.get("diagnostic", {})
-    if not isinstance(measurements, dict):
-        measurements = {}
-    try:
-        reason = measurements.get("reason") or event.get("reason_code")
-        return FieldCodec.decode(FailureReason, reason)
-    except MalformedRecord:
-        return FailureReason.BACKEND_FAILED
-
-
 def record_terminal_failure(
     root: Path,
     *,
     turn_id: str,
     thread: str,
-    event: dict,
+    failure: type[TurnFailure],
     sequences: tuple[int, ...],
+    measurements: Mapping[str, object] | None = None,
     native_response: Response | None = None,
     source_error: BaseException | None = None,
 ) -> Path:
     """Persist before publishing the failure notice; this record grants no retry authority."""
     Id128Text.decode(turn_id)
-    measurements = event.get("diagnostic", {})
-    if not isinstance(measurements, dict):
-        measurements = {}
-    reason = terminal_failure_reason(event)
+    measurements = measurements or {}
     safe = {
         key: value
         for key in (
@@ -176,7 +149,7 @@ def record_terminal_failure(
         "version": 1,
         "turn_id": turn_id,
         "thread": thread,
-        "reason": reason.value,
+        "failure": failure.declared_name,
         "sequences": list(sequences),
         "measurements": safe,
         "outcome": "failed; inputs must not be replayed automatically",
@@ -187,40 +160,6 @@ def record_terminal_failure(
     return DiagnosticDocument.write(root, turn_id, document)
 
 
-@dataclass(frozen=True)
-class NativeTerminalFailureRecord:
-    """The document ``record_terminal_failure`` writes for a native terminal failure.
-
-    Core writes it only after verifying the failed input's live context.
-    """
-
-    version: Literal[1]
-    turn_id: str
-    thread: str
-    reason: FailureReason
-    sequences: tuple[int, ...]
-    measurements: dict[str, int]
-    outcome: str
-    source_error: str
-    native: dict[str, str]
-
-    @classmethod
-    def read(cls, root: Path, input_id: str) -> NativeTerminalFailureRecord:
-        path = DiagnosticDocument.path_in(root, name=input_id)
-        record = FieldCodec.decode(cls, json.loads(path.read_text()))
-        if record.turn_id != input_id or not record.raised.startswith(
-            "agent_comms.native_pi.NativePiTerminalFailure: "
-        ):
-            raise ValueError("Diagnostic does not record a native terminal failure for this input")
-        return record
-
-    @property
-    def raised(self) -> str:
-        """The first unindented line of the last traceback is the raised exception."""
-        final = self.source_error.rpartition("Traceback (most recent call last):\n")[2]
-        return next((line for line in final.splitlines() if not line.startswith(" ")), "")
-
-
 def record_drain_failure(
     root: Path, *, thread: str, diagnostic: DrainDiagnostic, source_error: BaseException,
 ) -> Path: