From 42eba6a47a5a13a0a042a1d712499e3272a45063 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:07:05 -0400 Subject: [PATCH 1/4] Move pinned route privacy and socket OS choices onto Platform --- src/agent_comms/active_route.py | 14 ++++++------- src/agent_comms/child_process.py | 35 +++++++++++++++++++++++++++++++- src/agent_comms/private_path.py | 14 +++---------- 3 files changed, 44 insertions(+), 19 deletions(-) diff --git a/src/agent_comms/active_route.py b/src/agent_comms/active_route.py index dca5e83a8d..71a91cb590 100644 --- a/src/agent_comms/active_route.py +++ b/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 @@ -200,7 +201,7 @@ def read_active_route(path: Path | None = None) -> ActiveRoute | None: info = os.fstat(fd) if ( not stat.S_ISREG(info.st_mode) - or info.st_uid != os.geteuid() + or not Platform.current().is_owned(info) or stat.S_IMODE(info.st_mode) != 0o600 or info.st_nlink != 1 or not 0 < info.st_size <= 4096 @@ -248,8 +249,7 @@ def guard_default_route_write(expected_root: Path, *, blocking: bool = True) -> 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 +451,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 +475,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() diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py index 88068921f6..e2a9472a92 100644 --- a/src/agent_comms/child_process.py +++ b/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 diff --git a/src/agent_comms/private_path.py b/src/agent_comms/private_path.py index e14127e3f0..83acd733fb 100644 --- a/src/agent_comms/private_path.py +++ b/src/agent_comms/private_path.py @@ -8,15 +8,14 @@ import os import stat -import sys from abc import ABC, abstractmethod from collections.abc import Callable, Iterator from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path -from tempfile import TemporaryDirectory from typing import BinaryIO, ClassVar +from .child_process import Platform from .declared_family import DeclaredFamily from .errors import RelationViolationError from .field_codec import NullMember, RecordMember, SubclassUnion @@ -285,15 +284,8 @@ def address(cls, path: Path) -> Iterator[Path]: try: if not os.path.samestat(observed, os.fstat(fd)): raise ValueError("Socket directory changed during acquisition") - if sys.platform.startswith("linux"): - yield Path(f"/proc/{os.getpid()}/fd/{fd}") / path.name - else: - 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(fd)): - raise ValueError("Socket directory changed during address acquisition") - yield parent / path.name + with Platform.current().socket_address(path, fd) as address: + yield address finally: os.close(fd) From e9aa104cb44fa946ea136415d0447ae5d19dd21f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:09:51 -0400 Subject: [PATCH 2/4] Carry declared failures through terminals, goals and diagnostics --- src/agent_comms/agent_events.py | 14 ++-- src/agent_comms/backend.py | 10 ++- src/agent_comms/diagnostics.py | 77 +++------------------ src/agent_comms/goal_failure_observation.py | 10 ++- src/agent_comms/goal_presentation.py | 2 +- src/agent_comms/selected_result.py | 3 +- src/agent_comms/turn_failure.py | 29 +++++--- src/agent_comms/turn_output.py | 4 +- src/agent_comms/turn_progress.py | 3 +- src/agent_comms/turn_terminal.py | 12 ++-- src/agent_comms/turn_watchdog.py | 6 +- 11 files changed, 58 insertions(+), 112 deletions(-) diff --git a/src/agent_comms/agent_events.py b/src/agent_comms/agent_events.py index 0cafba4a34..909bf63e48 100644 --- a/src/agent_comms/agent_events.py +++ b/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) diff --git a/src/agent_comms/backend.py b/src/agent_comms/backend.py index 6437186226..ac5f2d6475 100644 --- a/src/agent_comms/backend.py +++ b/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(), )) diff --git a/src/agent_comms/diagnostics.py b/src/agent_comms/diagnostics.py index 7b46814031..c6bf024288 100644 --- a/src/agent_comms/diagnostics.py +++ b/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: diff --git a/src/agent_comms/goal_failure_observation.py b/src/agent_comms/goal_failure_observation.py index 035bd0bdcc..5c061dac4c 100644 --- a/src/agent_comms/goal_failure_observation.py +++ b/src/agent_comms/goal_failure_observation.py @@ -12,7 +12,7 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Annotated -from .diagnostics import FailureReason +from .turn_failure import TurnFailure from .field_codec import Id128Text, WireName from .goal_attempt_identity import GoalAttemptIdentity from .thread_identity import ( @@ -62,7 +62,7 @@ class FailedTurnEvidence(GoalLedgerTable, TypedTable, GoalFailureEvidence): turn_generation: Annotated[int, Column(check="turn_generation>0")] goal_revision: Annotated[int, Column(check="goal_revision>=0")] turn_id: Annotated[str, Column(unique=True, check="length(turn_id)=32")] - reason: FailureReason + failure: type[TurnFailure] def __post_init__(self) -> None: # These are the durable observation's constraints, independent of any @@ -74,8 +74,6 @@ def __post_init__(self) -> None: if not math.isfinite(self.owner_created_at): raise ValueError("Observation incarnation must be finite") Id128Text.from_text(self.turn_id) - if not isinstance(self.reason, FailureReason): - raise ValueError("Observation reason must be decoded") def execution(self, goal_id: str) -> GoalExecution: from .goal_presentation import AttemptFailedExecution @@ -132,7 +130,7 @@ def from_terminal( claim_scope: int, current_owner: Thread | None, current_claim: int | None, - reason: FailureReason, + failure: type[TurnFailure], ) -> FailedTurnObservation | None: # Observation is passive: a changed owner, goal or lease yields no # observation. The evidence is built from already-trusted values, so a @@ -140,7 +138,7 @@ def from_terminal( evidence = FailedTurnEvidence( reservation.attempt_id, reservation.goal_id, reservation.generation, owner.name, owner.created_at, owner.worktree, claim_scope, - lease.identity.generation, goal.revision, turn_id, reason, + lease.identity.generation, goal.revision, turn_id, failure, ) if owner.goal != goal or reservation.goal_id != goal.id: return None diff --git a/src/agent_comms/goal_presentation.py b/src/agent_comms/goal_presentation.py index 9dbb269f5f..13426e100d 100644 --- a/src/agent_comms/goal_presentation.py +++ b/src/agent_comms/goal_presentation.py @@ -269,7 +269,7 @@ class AttemptFailedExecution(BlockedExecution, RetryableExecution): @property def reason(self) -> str: - return self.evidence.reason.value + return self.evidence.failure.declared_name class GoalPublicationSnapshot(ABC): diff --git a/src/agent_comms/selected_result.py b/src/agent_comms/selected_result.py index c9aef68e53..ff91b72f82 100644 --- a/src/agent_comms/selected_result.py +++ b/src/agent_comms/selected_result.py @@ -12,6 +12,7 @@ from .diagnostics import record_terminal_failure from .messages import AlertKind, NoticeRole from .send_result import SendResult +from .turn_failure import BackendFailed @dataclass(frozen=True, slots=True) @@ -54,7 +55,7 @@ def publish_native_failure( participant.comms.root, turn_id=input_id, thread=participant.owner.thread.name, - event={}, + failure=BackendFailed, sequences=tuple(source.delivery.message.seq for source in participant.batch.sources), native_response=native_response, source_error=source_error, diff --git a/src/agent_comms/turn_failure.py b/src/agent_comms/turn_failure.py index d25ecf679e..23aaec190f 100644 --- a/src/agent_comms/turn_failure.py +++ b/src/agent_comms/turn_failure.py @@ -7,7 +7,6 @@ from typing import TYPE_CHECKING, ClassVar from .declared_family import DeclaredFamily -from .diagnostics import FailureReason from .native_arguments import SelectedModel if TYPE_CHECKING: @@ -18,7 +17,6 @@ @dataclass(frozen=True) class TurnFailure(DeclaredFamily): text: str - code: ClassVar[str | None] = None # A failure of the model request itself; Pi re-sending that request ends it. ends_with_request_retry: ClassVar[bool] = False @@ -54,19 +52,26 @@ class InputIdUnavailable(TurnFailure): def with_startup_diagnostics(self, stderr: str) -> TurnFailure: return self.with_text(self.text + "\nBackend startup reported:\n" + stderr) - code = FailureReason.INPUT_ID_UNAVAILABLE + + +class NativePreflightTimeout(InputIdUnavailable): + """Input-ID preflight expired before the native child attested.""" + + +class NativePreflightExit(InputIdUnavailable): + """The native child exited before input-ID preflight attested.""" class IdentityUncertain(TurnFailure): - code = FailureReason.IDENTITY_UNCERTAIN + pass class AuthorityChanged(TurnFailure): - code = FailureReason.AUTHORITY_CHANGED + pass class FollowupUnrecognized(TurnFailure): - code = FailureReason.FOLLOWUP_UNRECOGNIZED + pass class TerminalFailure(TurnFailure): @@ -84,7 +89,6 @@ def explanation(cls, output: TurnOutput) -> str: class InputMissing(TerminalFailure): - code = FailureReason.INPUT_MISSING default_text = "Pi RPC run ended without this prompt's user message start." @classmethod @@ -112,7 +116,6 @@ def raise_native(self, proof, model: SelectedModel) -> None: raise NativePiTerminalFailure(self.text, proof, model, self) - code = FailureReason.MODEL_REQUEST_FAILED ends_with_request_retry = True @@ -141,7 +144,6 @@ class ProviderTransportFailed(RetryableProviderFailure, ModelRequestFailed): class FinalStopMissing(TerminalFailure): - code = FailureReason.FINAL_STOP_MISSING default_text = "Pi RPC run ended without an authoritative final assistant stop." @classmethod @@ -154,7 +156,6 @@ def explanation(cls, output: TurnOutput) -> str: class QueuedInputMissing(TerminalFailure): - code = FailureReason.QUEUED_INPUT_MISSING default_text = "Pi RPC run ended with an unstarted queued input; delivery is uncertain." @classmethod @@ -176,3 +177,11 @@ class ExtensionUiFailed(TurnFailure): class BackendDidNotExit(TurnFailure): pass + + +class BackendFailed(TurnFailure): + """A failed terminal without a more specific native failure declaration.""" + + +class NativeLaunchInvalid(BackendFailed): + """The configured native launch could not be acquired.""" diff --git a/src/agent_comms/turn_output.py b/src/agent_comms/turn_output.py index 24cbd3ba06..d79a5c0c14 100644 --- a/src/agent_comms/turn_output.py +++ b/src/agent_comms/turn_output.py @@ -30,7 +30,6 @@ class TurnOutput: default_factory=UnrecordedFailureSource, init=False, repr=False) final_assistant_stop: bool = False diagnostic: dict[str, int] = field(default_factory=dict) - preflight_failure: str | None = None # The live turn owns the one privacy-safe decoded retry result. Completion # and failure presentation retain this immutable value; they do not decode it. retry_result: ProviderRetryResult | None = None @@ -123,11 +122,10 @@ def done(self, session: TurnSession, stderr: str) -> Iterator[events.AgentEvent] terminal = events.Done( text=self.terminal_text(success, stderr, session.native.proc.returncode), ok=success and session.native.attestation.trustworthy, - reason_code=self.failure.code if self.failure else None, + failure=self.declared_failure if self.failure else None, retry_result=self.retry_result, diagnostic={ **self.diagnostic, - **({"reason": self.preflight_failure} if self.preflight_failure else {}), **( {"exit_code": session.native.proc.returncode} if session.native.proc.returncode is not None diff --git a/src/agent_comms/turn_progress.py b/src/agent_comms/turn_progress.py index 2489c74684..65667e5d16 100644 --- a/src/agent_comms/turn_progress.py +++ b/src/agent_comms/turn_progress.py @@ -322,7 +322,8 @@ def publish_checkpoint(self, terminal: events.Done) -> tuple[SendResult, ...]: self.comms.root, turn_id=self.turn_lease.turn_id, thread=self.thread_name, - event=vars(terminal), + failure=terminal.failure_kind, + measurements=terminal.diagnostic, sequences=(), ) results = [] diff --git a/src/agent_comms/turn_terminal.py b/src/agent_comms/turn_terminal.py index 22406c8b81..9f5d2923f2 100644 --- a/src/agent_comms/turn_terminal.py +++ b/src/agent_comms/turn_terminal.py @@ -7,7 +7,7 @@ from contextlib import suppress from typing import TYPE_CHECKING -from .diagnostics import FailureReason, terminal_failure_reason +from .turn_failure import BackendFailed, TurnFailure from .goal_failure_observation import FailedTurnObservation from .goal_attempts import StaleAttemptError @@ -77,12 +77,12 @@ def failure_observation(self, account, permit, captured): claim_scope=account.lease.claimed_generation, current_owner=snapshot.threads.get(account.thread_name), current_claim=snapshot.claim_generation(account.thread_name), - reason=self.failure_reason, + failure=self.failure_kind, ) @property @abstractmethod - def failure_reason(self) -> FailureReason: ... + def failure_kind(self) -> type[TurnFailure]: ... def settle_launched(self, account, permit, store, captured, current) -> None: if permit.has_verified_progress(store): @@ -90,15 +90,15 @@ def settle_launched(self, account, permit, store, captured, current) -> None: observation = self.failure_observation(account, permit, captured) with suppress(StaleAttemptError): permit.reservation.fail( - store, self.failure_reason.value, + store, self.failure_kind.declared_name, observation=observation, ) class EndedWithoutResult(FailedGoalTurn): @property - def failure_reason(self) -> FailureReason: - return terminal_failure_reason({}) + def failure_kind(self) -> type[TurnFailure]: + return BackendFailed class OperatorStopped(TurnTerminal): diff --git a/src/agent_comms/turn_watchdog.py b/src/agent_comms/turn_watchdog.py index 80c0c03498..a49d75960a 100644 --- a/src/agent_comms/turn_watchdog.py +++ b/src/agent_comms/turn_watchdog.py @@ -12,7 +12,6 @@ from . import turn_failure as failures from . import turn_phase as phases from .declared_family import DeclaredFamily -from .diagnostics import FailureReason from .mro_dispatch import MroDispatch, handles from .native_startup import SessionLoad @@ -76,7 +75,7 @@ async def expire(self, session: TurnSession) -> AsyncIterator[events.AgentEvent] class PreflightWait(TurnWait): - reason = "native_preflight_timeout" + reason = failures.NativePreflightTimeout.declared_name phase = "await_get_state" async def expire(self, session: TurnSession) -> AsyncIterator[events.AgentEvent]: @@ -84,7 +83,6 @@ async def expire(self, session: TurnSession) -> AsyncIterator[events.AgentEvent] elapsed_ms = round((watchdog.clock() - watchdog.launch_started_at) * 1000) wait_ms = round(self.elapsed * 1000) budget_ms = round(self.budget * 1000) - session.output.preflight_failure = FailureReason.PREFLIGHT_TIMEOUT session.output.diagnostic = { "elapsed_ms": elapsed_ms, "wait_ms": wait_ms, @@ -94,7 +92,7 @@ async def expire(self, session: TurnSession) -> AsyncIterator[events.AgentEvent] session.output.diagnostic.update(watchdog.session_load.diagnostic) session_size = watchdog.session_load.summary session.output.record_failure( - failures.InputIdUnavailable( + failures.NativePreflightTimeout( "Pi native input-ID capability preflight timed out " f"(phase={self.phase}, elapsed_ms={elapsed_ms}, " f"wait_ms={wait_ms}, budget_ms={budget_ms}, " From 0969fa7aa8442f0733f0c60277fdab1d92243ca1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:12:16 -0400 Subject: [PATCH 3/4] Carry stored failure names and adapt consumers to declared kinds --- tests/test_agent_events.py | 2 +- tests/test_goal_failure_observation.py | 6 +- tests/test_headless_diagnostics.py | 11 +- tests/test_pi_rpc_nominal.py | 2 +- tests/test_preflight_feedback.py | 17 ++-- tests/test_turn_failure_privacy.py | 2 +- tools/cutover/carry_declared_failures.py | 124 +++++++++++++++++++++++ 7 files changed, 142 insertions(+), 22 deletions(-) create mode 100644 tools/cutover/carry_declared_failures.py diff --git a/tests/test_agent_events.py b/tests/test_agent_events.py index ecb0535a0e..7fd12718c2 100644 --- a/tests/test_agent_events.py +++ b/tests/test_agent_events.py @@ -173,7 +173,7 @@ def test_payload_fields_and_frozen_multiple_inheritance(): with pytest.raises(FrozenInstanceError): value.extra = "not a declared field" done = events.Done("failed", False, "input_authority_changed", {"elapsed_ms": 3}) - assert done.reason_code == "input_authority_changed" + assert done.reason_code == "authority_changed" assert done.diagnostic == {"elapsed_ms": 3} diff --git a/tests/test_goal_failure_observation.py b/tests/test_goal_failure_observation.py index b5c90ec931..0dc933eeb7 100644 --- a/tests/test_goal_failure_observation.py +++ b/tests/test_goal_failure_observation.py @@ -8,7 +8,7 @@ import pytest from agent_comms.child_process import ProcessIdentity -from agent_comms.diagnostics import FailureReason +from agent_comms.turn_failure import BackendFailed, FinalStopMissing from agent_comms.goal_attempts import ( AttemptRecord, GoalAttemptStore, @@ -43,7 +43,7 @@ def bound(tmp_path, *, thread_registration): observation = FailedTurnObservation.from_terminal( permit.reservation, owner=owner, goal=goal, lease=lease, turn_id=lease.turn_id, claim_scope=3, current_owner=owner, current_claim=3, - reason=FailureReason.FINAL_STOP_MISSING, + failure=FinalStopMissing, ) assert observation is not None return store, owner, lease, observation @@ -81,7 +81,7 @@ def test_mismatched_turn_lease_is_not_bound(bound, mutation): rejected = FailedTurnObservation.from_terminal( observation.reservation, owner=owner, goal=owner.goal, lease=replace(lease, **mutation), turn_id=lease.turn_id, claim_scope=3, - current_owner=owner, current_claim=3, reason=FailureReason.BACKEND_FAILED, + current_owner=owner, current_claim=3, failure=BackendFailed, ) assert rejected is None observation.reservation.fail(store, "failed anyway", observation=rejected) diff --git a/tests/test_headless_diagnostics.py b/tests/test_headless_diagnostics.py index 68428f81db..0677ae0397 100644 --- a/tests/test_headless_diagnostics.py +++ b/tests/test_headless_diagnostics.py @@ -4,6 +4,7 @@ import os from agent_comms.diagnostics import record_terminal_failure +from agent_comms.turn_failure import NativePreflightTimeout def test_terminal_diagnostic_excludes_untrusted_payloads(tmp_path): @@ -12,22 +13,18 @@ def test_terminal_diagnostic_excludes_untrusted_payloads(tmp_path): turn_id="a" * 32, thread="worker", sequences=(6943,), - event={ - "reason_code": "native_preflight_timeout", - "text": "SECRET stderr data:image/png;base64,ABC123", - "prompt": "private user text", - "diagnostic": { + failure=NativePreflightTimeout, + measurements={ "wait_ms": 5000, "session_bytes": 88255627, "exit_code": -15, "token": "SECRET", "spawn_ms": "SECRET", - }, }, ) raw = path.read_text() value = json.loads(raw) - assert value["reason"] == "native_preflight_timeout" + assert value["failure"] == NativePreflightTimeout.declared_name assert value["measurements"] == {"wait_ms": 5000, "session_bytes": 88255627, "exit_code": -15} assert value["sequences"] == [6943] assert all(fragment not in raw for fragment in ("SECRET", "private user", "base64")) diff --git a/tests/test_pi_rpc_nominal.py b/tests/test_pi_rpc_nominal.py index a94dcd3c93..0cde17a8b3 100644 --- a/tests/test_pi_rpc_nominal.py +++ b/tests/test_pi_rpc_nominal.py @@ -119,7 +119,7 @@ def test_failure_precedence_and_text_have_one_owner(left, right): expected = min((left, right), key=lambda cls: cls.declared_position()) assert type(session.output.failure) is expected assert session.output.failure_text == expected.__name__ - assert session.output.failure.code == expected.code + assert session.output.failure.declared_name == expected.declared_name def test_external_formats_are_derived_and_pin_wire_spellings(): diff --git a/tests/test_preflight_feedback.py b/tests/test_preflight_feedback.py index 0593970bda..daff7af890 100644 --- a/tests/test_preflight_feedback.py +++ b/tests/test_preflight_feedback.py @@ -2,22 +2,21 @@ import pytest from agent_comms.input_attempt import ReservedInput, NotSentInput, MissingInput from agent_comms.input_disposition import InputDocument -from agent_comms.turn_failure import InputIdUnavailable, PromptSendFailed +from agent_comms.turn_failure import NativePreflightExit, NativePreflightTimeout, PromptSendFailed from agent_comms.turn_output import TurnOutput -from agent_comms.diagnostics import FailureReason from stored_inputs import stored_inputs -@pytest.mark.parametrize('reason', [FailureReason.PREFLIGHT_EXIT, FailureReason.PREFLIGHT_TIMEOUT]) -def test_preflight_diagnostics_keep_original_phase(reason): - output = TurnOutput(preflight_failure=reason) - output.record_failure(InputIdUnavailable('Original preflight phase and timing')) +@pytest.mark.parametrize('kind', [NativePreflightExit, NativePreflightTimeout]) +def test_preflight_diagnostics_keep_original_phase(kind): + output = TurnOutput() + output.record_failure(kind('Original preflight phase and timing')) output.startup_error('Native startup cause\n') assert output.failure_text == 'Original preflight phase and timing\nBackend startup reported:\nNative startup cause' - output = TurnOutput(sensitive=True, preflight_failure=reason) - output.record_failure(InputIdUnavailable('Private phase')) + output = TurnOutput(sensitive=True) + output.record_failure(kind('Private phase')) output.startup_error('private image bytes') assert output.failure_text == 'Image prompt failed; backend diagnostics withheld.' - assert output.preflight_failure is reason + assert type(output.failure) is kind assert 'Private phase' not in repr(output) assert 'private image bytes' not in repr(output) ordinary = TurnOutput(failure=PromptSendFailed('Transport cause')) diff --git a/tests/test_turn_failure_privacy.py b/tests/test_turn_failure_privacy.py index cd2c95b4bc..2565965fdc 100644 --- a/tests/test_turn_failure_privacy.py +++ b/tests/test_turn_failure_privacy.py @@ -22,7 +22,7 @@ def test_recorded_failure_has_one_safe_text_and_keeps_its_behavior(kind, sensiti assert original.text == 'private provider response body' assert stored.text == output.failure_text == expected assert output.terminal_text(False, 'private stderr', 1) == expected - assert stored.code == original.code + assert type(stored) is type(original) assert stored.declared_position() == original.declared_position() assert stored.ends_with_request_retry == original.ends_with_request_retry if sensitive: diff --git a/tools/cutover/carry_declared_failures.py b/tools/cutover/carry_declared_failures.py new file mode 100644 index 0000000000..3b6e336860 --- /dev/null +++ b/tools/cutover/carry_declared_failures.py @@ -0,0 +1,124 @@ +"""Carry obsolete reason labels into declared failure classes and the failure key. + +Only installation reads the retired spellings. Delete after every retained root +has been carried; running readers have no compatibility decoder. +""" + +from __future__ import annotations + +import json +from contextlib import closing +from pathlib import Path +from typing import ClassVar + +from agent_comms import turn_failure as failures +from agent_comms.diagnostics import DiagnosticDocument +from agent_comms.errors import RelationViolationError +from agent_comms.goal_attempts import AttemptRecord, GoalAttemptStore, GoalLedgerTable +from agent_comms.goal_failure_observation import FailedTurnEvidence +from agent_comms.installer.carry import Carried, Carry +from agent_comms.locked_store import SQLiteFile +from agent_comms.store_files import _atomic_write_text +from agent_comms.typed_table import HeldBusFile, sql_literal + +from .carry_bus_history import CarryCatalogSource + + +class DeclaredFailures(Carry): + old_names: ClassVar[dict[str, type[failures.TurnFailure]]] = { + "backend_failed": failures.BackendFailed, + "native_preflight_timeout": failures.NativePreflightTimeout, + "native_preflight_exit": failures.NativePreflightExit, + "pi_input_id_unavailable": failures.InputIdUnavailable, + "session_identity_uncertain": failures.IdentityUncertain, + "input_authority_changed": failures.AuthorityChanged, + "unrecognized_followup_input": failures.FollowupUnrecognized, + "current_prompt_input_missing": failures.InputMissing, + "model_request_failed": failures.ModelRequestFailed, + "assistant_final_stop_missing": failures.FinalStopMissing, + "queued_input_start_missing": failures.QueuedInputMissing, + } + + def roots(self) -> tuple[Path, ...]: + return tuple(dict.fromkeys(( + self.root, + *(source.mapped_root(self.root, self.original_root) + for source in CarryCatalogSource.read(self.root, self.original_root)), + ))) + + def backup_files(self) -> tuple[Path, ...]: + paths = tuple( + path for root in self.roots() + for path in ( + *GoalAttemptStore.files(GoalAttemptStore.path_in(root)), + *DiagnosticDocument.files_in(root), + *GoalAttemptStore.source_paths(root), + ) if path.exists() + ) + for path in paths: + self.require_confined(self.root, path) + return paths + + def carry(self, *, apply: bool) -> Carried: + self.backup_files() + if apply: + # The installer's root hold also covers its retained, inactive snapshots. + HeldBusFile(GoalAttemptStore, GoalAttemptStore.path_in(self.root)).require() + results = tuple(self.carry_root(root, apply=apply) for root in self.roots()) + return Carried( + sum(result.changed for result in results), + rewritten=tuple(name for result in results for name in result.rewritten), + ) + + def carry_root(self, root: Path, *, apply: bool) -> Carried: + changed = 0 + rewritten: list[str] = [] + path = GoalAttemptStore.path_in(root) + if path.exists(): + with closing(SQLiteFile.writer(":memory:")) as prepared: + with closing(SQLiteFile.read_snapshot(path)) as original: + original.backup(prepared) + columns = FailedTurnEvidence.stored_columns(prepared) + if "reason" in columns: + for (reason,) in prepared.execute( + f'SELECT DISTINCT reason FROM "{FailedTurnEvidence.declared_name}"' + ): + if reason not in self.old_names: + raise RelationViolationError(f"Unknown stored failure at {path}: {reason!r}") + cases = " ".join( + f"WHEN {sql_literal(old)} THEN {sql_literal(kind)}" + for old, kind in self.old_names.items() + ) + FailedTurnEvidence.rebuild(prepared, {"failure": f"CASE reason {cases} END"}) + changed += 1 + if AttemptRecord.stored_columns(prepared): + for old, kind in self.old_names.items(): + if old != kind.declared_name: + changed += prepared.execute( + f'UPDATE "{AttemptRecord.declared_name}" SET resolution=? ' + "WHERE resolution=? AND json_extract(phase,'$.kind')='failed'", + (kind.declared_name, old), + ).rowcount + if changed: + GoalLedgerTable.require_schema(prepared) + if apply: + prepared.commit() + with closing(SQLiteFile.writer(path)) as destination: + prepared.backup(destination) + GoalAttemptStore.publish_change(path) + rewritten.extend(str(file.relative_to(self.root)) for file in ( + path, *GoalAttemptStore.source_paths(root), + )) + for diagnostic in DiagnosticDocument.matching_in(root): + document = json.loads(diagnostic.read_text()) + if "turn_id" not in document or "reason" not in document: + continue + if "failure" in document: + raise RelationViolationError(f"Both failure formats are present at {diagnostic}") + old = document.pop("reason") + document["failure"] = self.old_names[old].declared_name + if apply: + _atomic_write_text(diagnostic, json.dumps(document, indent=2) + "\n") + rewritten.append(str(diagnostic.relative_to(self.root))) + changed += 1 + return Carried(changed, rewritten=tuple(rewritten)) From f5b83f5202d7c4b88d0b01bc756d793986724d55 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:13:29 -0400 Subject: [PATCH 4/4] Close remaining route privacy decisions and terminal fixture inputs --- src/agent_comms/active_route.py | 8 +++----- src/agent_comms/turn_failure.py | 1 - tests/test_agent_events.py | 4 +++- tests/test_headless_diagnostics.py | 10 +++++----- tests/test_turn_failure_privacy.py | 2 +- tools/cutover/carry_declared_failures.py | 4 ++-- 6 files changed, 14 insertions(+), 15 deletions(-) diff --git a/src/agent_comms/active_route.py b/src/agent_comms/active_route.py index 71a91cb590..b95a5b2b4a 100644 --- a/src/agent_comms/active_route.py +++ b/src/agent_comms/active_route.py @@ -194,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 not Platform.current().is_owned(info) - 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 ): @@ -245,7 +243,7 @@ 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") diff --git a/src/agent_comms/turn_failure.py b/src/agent_comms/turn_failure.py index 23aaec190f..fdeafae6a8 100644 --- a/src/agent_comms/turn_failure.py +++ b/src/agent_comms/turn_failure.py @@ -53,7 +53,6 @@ def with_startup_diagnostics(self, stderr: str) -> TurnFailure: return self.with_text(self.text + "\nBackend startup reported:\n" + stderr) - class NativePreflightTimeout(InputIdUnavailable): """Input-ID preflight expired before the native child attested.""" diff --git a/tests/test_agent_events.py b/tests/test_agent_events.py index 7fd12718c2..823008c17c 100644 --- a/tests/test_agent_events.py +++ b/tests/test_agent_events.py @@ -172,7 +172,9 @@ def test_payload_fields_and_frozen_multiple_inheritance(): for value in (start, events.OriginalInputStarted(), events.ModelChanged("id", True)): with pytest.raises(FrozenInstanceError): value.extra = "not a declared field" - done = events.Done("failed", False, "input_authority_changed", {"elapsed_ms": 3}) + from agent_comms.turn_failure import AuthorityChanged + + done = events.Done("failed", False, AuthorityChanged("failed"), {"elapsed_ms": 3}) assert done.reason_code == "authority_changed" assert done.diagnostic == {"elapsed_ms": 3} diff --git a/tests/test_headless_diagnostics.py b/tests/test_headless_diagnostics.py index 0677ae0397..466eb3463d 100644 --- a/tests/test_headless_diagnostics.py +++ b/tests/test_headless_diagnostics.py @@ -15,11 +15,11 @@ def test_terminal_diagnostic_excludes_untrusted_payloads(tmp_path): sequences=(6943,), failure=NativePreflightTimeout, measurements={ - "wait_ms": 5000, - "session_bytes": 88255627, - "exit_code": -15, - "token": "SECRET", - "spawn_ms": "SECRET", + "wait_ms": 5000, + "session_bytes": 88255627, + "exit_code": -15, + "token": "SECRET", + "spawn_ms": "SECRET", }, ) raw = path.read_text() diff --git a/tests/test_turn_failure_privacy.py b/tests/test_turn_failure_privacy.py index 2565965fdc..8cae43e5c6 100644 --- a/tests/test_turn_failure_privacy.py +++ b/tests/test_turn_failure_privacy.py @@ -22,7 +22,7 @@ def test_recorded_failure_has_one_safe_text_and_keeps_its_behavior(kind, sensiti assert original.text == 'private provider response body' assert stored.text == output.failure_text == expected assert output.terminal_text(False, 'private stderr', 1) == expected - assert type(stored) is type(original) + assert stored.declared_name == original.declared_name assert stored.declared_position() == original.declared_position() assert stored.ends_with_request_retry == original.ends_with_request_retry if sensitive: diff --git a/tools/cutover/carry_declared_failures.py b/tools/cutover/carry_declared_failures.py index 3b6e336860..830e1e2b22 100644 --- a/tools/cutover/carry_declared_failures.py +++ b/tools/cutover/carry_declared_failures.py @@ -14,8 +14,8 @@ from agent_comms import turn_failure as failures from agent_comms.diagnostics import DiagnosticDocument from agent_comms.errors import RelationViolationError -from agent_comms.goal_attempts import AttemptRecord, GoalAttemptStore, GoalLedgerTable -from agent_comms.goal_failure_observation import FailedTurnEvidence +from agent_comms.goal_attempts import AttemptRecord, GoalAttemptStore +from agent_comms.goal_failure_observation import FailedTurnEvidence, GoalLedgerTable from agent_comms.installer.carry import Carried, Carry from agent_comms.locked_store import SQLiteFile from agent_comms.store_files import _atomic_write_text