diff --git a/src/agent_comms/backend.py b/src/agent_comms/backend.py index a95cb9c7e5..315730a22f 100644 --- a/src/agent_comms/backend.py +++ b/src/agent_comms/backend.py @@ -403,7 +403,8 @@ async def acquire_native(self, resources: AsyncExitStack, *, reuse: bool) -> PiS with self.startup.measurements.operation("open_transport"): unbound = await self.open_transport(transport) native = await self.open_native(reuse=reuse) - self.tool_socket = await unbound.bind_process(native.proc.pid) + with self.startup.measurements.operation("native_process_binding"): + self.tool_socket = await unbound.bind_process(native.proc.pid) native.transport = transport.pop_all() self.owner = asyncio.current_task() if self.owner is not None: @@ -414,9 +415,11 @@ async def acquire_native(self, resources: AsyncExitStack, *, reuse: bool) -> PiS async def open_native(self, *, reuse: bool) -> PiSessionChild: """Let the session manage child ownership for new or continued input.""" with self.startup.measurements.operation("native_open"): - return await self.native_session.open( + native = await self.native_session.open( self.launch, reuse=reuse, startup=self.startup, watchdog=self.watchdog, ) + self.startup.native_process = native.proc.identity + return native async def run(self) -> AsyncGenerator[events.AgentEvent, None]: self.loop = asyncio.get_running_loop() diff --git a/src/agent_comms/coordination_session.py b/src/agent_comms/coordination_session.py index 95e75a4509..91937b3c1d 100644 --- a/src/agent_comms/coordination_session.py +++ b/src/agent_comms/coordination_session.py @@ -17,7 +17,7 @@ CoordinationReadUnavailable, IntegrityViolationError, ) -from agent_comms.diagnostics import PublicationMeasurements +from agent_comms.publication_measurements import PublicationMeasurements from agent_comms.locked_store import RegistryFile from agent_comms.store_files import ( DEFAULT_STORE_CONTENTION, StoreContention, diff --git a/src/agent_comms/cursor_publication.py b/src/agent_comms/cursor_publication.py index 2de6706f9e..c7853172b4 100644 --- a/src/agent_comms/cursor_publication.py +++ b/src/agent_comms/cursor_publication.py @@ -26,7 +26,7 @@ from .coordination_errors import DeferredRead from .coordinator import Coordination from .cursor_owner import CommittedCursorOwner, CursorOwner, CursorState -from .diagnostics import PublicationMeasurements +from .publication_measurements import PublicationMeasurements from .field_codec import InputId from .message_bus import MessageBus from .native_entries import NativeEvidenceScope diff --git a/src/agent_comms/diagnostics.py b/src/agent_comms/diagnostics.py index c90b5d7106..c7bdf1a739 100644 --- a/src/agent_comms/diagnostics.py +++ b/src/agent_comms/diagnostics.py @@ -8,7 +8,7 @@ import time from contextlib import contextmanager from collections.abc import Mapping -from dataclasses import dataclass, field, replace +from dataclasses import replace from pathlib import Path from traceback import TracebackException from typing import TYPE_CHECKING @@ -19,12 +19,14 @@ ) -from .child_process import Platform +from .child_process import Platform, ProcessState from .field_codec import FieldCodec, InputId, ScratchName, TurnId from .locked_store import PatternFile, RuntimeFile from .native_failures import NativeRequestFailure +from .publication_measurements import PublicationMeasurements from .store_files import AtomicFile from .turn_failure import TurnFailure +from .turn_lease import TurnLeaseFence if TYPE_CHECKING: from .activity import DrainDiagnostic @@ -55,39 +57,6 @@ def open(cls, root: Path, thread: Thread): yield output -@dataclass -class PublicationMeasurements: - """Bounded transport timing counters, not a message or phase authority.""" - count: int = 0 - total_ns: int = 0 - maximum_ns: int = 0 - maximum_started_ns: int = 0 - maximum_finished_ns: int = 0 - operations: dict[str, PublicationMeasurements] = field(default_factory=dict) - - def operation(self, name: str): - """Borrow bounded counters for a declared acquisition/publication operation. - - These counters observe resources; they never decide readiness, acknowledgement - acceptance or replay. Owners call this with their fixed operation names. - """ - return self.operations.setdefault(name, PublicationMeasurements()).measuring() - - @contextmanager - def measuring(self): - started = time.monotonic_ns() - try: - yield - finally: - finished = time.monotonic_ns() - duration = finished - started - self.count += 1 - self.total_ns += duration - if duration > self.maximum_ns: - self.maximum_ns = duration - self.maximum_started_ns, self.maximum_finished_ns = started, finished - - def record_request_progress(root, lease, progress, native_process, *, publication=None): """Append original measurements with the exact existing turn/owner fence. @@ -104,15 +73,23 @@ def record_request_progress(root, lease, progress, native_process, *, publicatio RequestObservations.record(root, lease, record) -def record_acquisition_progress(root, lease, input_id, measurements): +def record_acquisition_progress( + root: Path, + lease: TurnLeaseFence, + input_id: InputId, + measurements: PublicationMeasurements, + native_process: ProcessState, +) -> None: """Report the parent's acquisition timings after its resources close. - The same diagnostic stream retains native and parent clocks with distinct - keys. There is no additional phase, readiness ledger or observation store. + The original input owns these counters, not the recipient's cursor lifetime. + NativeStartup observed the actual acquired child, or NoProcess if acquisition + never returned. This diagnostic grants no send, readiness or replay permission. """ RequestObservations.record(root, lease, { "turn": FieldCodec.encode(lease), "input_id": input_id, "acquisition": FieldCodec.encode(measurements), + "native_process": FieldCodec.encode(native_process), "recorded_monotonic_ns": time.monotonic_ns(), }) diff --git a/src/agent_comms/external_transcript.py b/src/agent_comms/external_transcript.py index b0800cb5f7..d3b5e5e2e6 100644 --- a/src/agent_comms/external_transcript.py +++ b/src/agent_comms/external_transcript.py @@ -286,7 +286,7 @@ def external_row(self): @classmethod @contextmanager - def publication(cls, root: Path, input_id: InputId): + def publication(cls, root: Path, input_id: InputId) -> Iterator[DraftPublication]: from .transcripts import TranscriptRead with TranscriptRead.publication_source(root) as connection: original = NativeRuntimeInput.one(connection, input_id=input_id) if connection is not None else None diff --git a/src/agent_comms/native_child.py b/src/agent_comms/native_child.py index 2e29ce1f8c..ee989c91e8 100644 --- a/src/agent_comms/native_child.py +++ b/src/agent_comms/native_child.py @@ -21,12 +21,12 @@ from .pi_events import PiEvent from .pi_rpc import PiRpcChannel from .private_path import PrivateSocketRole +from .publication_measurements import PublicationMeasurements from .session_revision import SessionRevision if TYPE_CHECKING: from typing import Self - from .diagnostics import PublicationMeasurements from .native_package import VerifiedNativePackage from .participant_identity import ParticipantIdentity from .pi_commands import ExtensionUiResponse, PiCommand diff --git a/src/agent_comms/native_prompt_send.py b/src/agent_comms/native_prompt_send.py index 8b7bfbc73b..5725a150bc 100644 --- a/src/agent_comms/native_prompt_send.py +++ b/src/agent_comms/native_prompt_send.py @@ -1,7 +1,7 @@ """One-use fenced raw prompt writes, independent of the owner's asyncio loop. Each successful os.write into the native pipe sends bytes irreversibly -while the input locks are held. No StreamWriter prompt buffer +after the original one-use grant. No StreamWriter prompt buffer exists to flush after that scope. Partial/failed/timed-out/cancelled sends grant no acceptance and are never retried. The reserved input remains unreplayable. """ @@ -18,7 +18,7 @@ from collections.abc import Callable from contextlib import AbstractContextManager, ExitStack from typing import TYPE_CHECKING -from .diagnostics import PublicationMeasurements +from .publication_measurements import PublicationMeasurements from .field_codec import AcpSessionId from .coordination_errors import CoordinationError from agent_comms.errors import RelationViolationError @@ -101,25 +101,28 @@ def _write_fenced( with authorized: deadline = time.monotonic() + timeout remaining = memoryview(payload) - while remaining: - budget = deadline - time.monotonic() - if cancelled.is_set() or budget <= 0: - raise PromptSendUnknown( - "Native prompt send is UNKNOWN after cancellation/deadline" - ) - try: - with measurements.operation("raw_pipe_write"): - count = os.write(fd, remaining) - except BlockingIOError: - # Bounded readiness wait, not input replay or provider retry. - select.select([], [fd], [], min(budget, 0.05)) - continue - except OSError as error: - raise PromptSendUnknown("Native prompt write is UNKNOWN; no retry") from error - if count <= 0: - raise PromptSendUnknown("Native prompt short write is UNKNOWN; no retry") - written += count - remaining = remaining[count:] + # One writer-thread interval includes partial writes and readiness + # waits. Exceptional exit still records time, never successful delivery. + with measurements.operation("prompt_write_loop"): + while remaining: + budget = deadline - time.monotonic() + if cancelled.is_set() or budget <= 0: + raise PromptSendUnknown( + "Native prompt send is UNKNOWN after cancellation/deadline" + ) + try: + with measurements.operation("raw_pipe_write"): + count = os.write(fd, remaining) + except BlockingIOError: + # Bounded readiness wait, not input replay or provider retry. + select.select([], [fd], [], min(budget, 0.05)) + continue + except OSError as error: + raise PromptSendUnknown("Native prompt write is UNKNOWN; no retry") from error + if count <= 0: + raise PromptSendUnknown("Native prompt short write is UNKNOWN; no retry") + written += count + remaining = remaining[count:] except (OSError, sqlite3.Error, CoordinationError) as error: if written: raise PromptSendUnknown( diff --git a/src/agent_comms/native_startup.py b/src/agent_comms/native_startup.py index c895bf0bd9..a84ef46eda 100644 --- a/src/agent_comms/native_startup.py +++ b/src/agent_comms/native_startup.py @@ -7,7 +7,8 @@ from pathlib import Path from typing import TYPE_CHECKING -from .diagnostics import PublicationMeasurements +from .child_process import NoProcess, ProcessState +from .publication_measurements import PublicationMeasurements from .field_codec import boundary_decoder if TYPE_CHECKING: @@ -94,10 +95,11 @@ def readiness_timeout( @dataclass class NativeStartup: - """Directory and measurements shared while opening one native child.""" + """Acquisition diagnostics shared by one operation, never send authority.""" root: Path measurements: PublicationMeasurements = field(default_factory=PublicationMeasurements, kw_only=True) + native_process: ProcessState = field(default_factory=NoProcess, init=False) @classmethod def for_launch( diff --git a/src/agent_comms/private_send_claim.py b/src/agent_comms/private_send_claim.py index 1e175de8c6..a127861037 100644 --- a/src/agent_comms/private_send_claim.py +++ b/src/agent_comms/private_send_claim.py @@ -37,6 +37,8 @@ from .native_startup import NativeStartup from .native_tool_call import NativeToolMode from .owned_send import OwnedSend +from .publication_measurements import PublicationMeasurements +from .registry_document import RegistrySnapshot from .text_digest import TextDigest from .tracked_turn import ( NO_FOLLOWING_INPUTS, @@ -63,6 +65,7 @@ class PrivateSendClaim(PromptBoundary): token_digest: str prompt: RenderedInput session: SelectedSession + measurements: PublicationMeasurements = field(repr=False, compare=False) _once: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) def following_sender(self, inputs: InputDrain, session_id: AcpSessionId, prompt: str) -> OwnedSend: @@ -119,20 +122,23 @@ async def reserve( A failed binding leaves the durable reservation intact; construction must never roll it back or make another input eligible for automatic replay. """ + measurements = PublicationMeasurements() digest = TextDigest.of(token).value def reserve(store): - input_id = stage.reserve(store, selected.identity, digest) - bind_expected_prompt( - store, - input_id=input_id, - stage=stage, - owner=selected.owner.thread, - generation=selected.identity.generation, - prompt=prompt.text, - ) - return input_id + with measurements.operation("input_reservation_binding_worker"): + input_id = stage.reserve(store, selected.identity, digest) + bind_expected_prompt( + store, + input_id=input_id, + stage=stage, + owner=selected.owner.thread, + generation=selected.identity.generation, + prompt=prompt.text, + ) + return input_id - input_id = await Coordination.run_async(selected.store.session.path, reserve) + with measurements.operation("input_reservation_binding"): + input_id = await Coordination.run_async(selected.store.session.path, reserve) return cls( selected=selected, stage=stage, @@ -140,6 +146,7 @@ def reserve(store): token_digest=digest, prompt=prompt, session=session, + measurements=measurements, ) def verify( @@ -174,8 +181,14 @@ async def execute( The raw writer keeps the single-use send token after the input is recorded. """ lease = self.owner.thread.require_turn_lease() - registry = await Coordination.run_worker(self.bus._registry.snapshot) + startup = NativeStartup(self.bus.bus.path.parent, measurements=self.measurements) + def snapshot_registry() -> RegistrySnapshot: + with self.measurements.operation("selected_registry_snapshot_worker"): + return self.bus._registry.snapshot() + try: + with self.measurements.operation("selected_registry_snapshot"): + registry = await Coordination.run_worker(snapshot_registry) result = await TrackedTurnSession.execute( execution.native_package, input_id=self.input_id, @@ -189,10 +202,7 @@ async def execute( self.bus.bus.path.parent, registry, ), ingress=MaintenanceBarrier(self.bus._registry.store.path), - startup=NativeStartup( - self.bus.bus.path.parent, - measurements=self.selected.cursor_delivery.measurements, - ), + startup=startup, selected_tool_mode=selected_tool_mode, observe_event=observe_event, request_observer=partial( @@ -220,7 +230,7 @@ def settle(store: Coordination, terminal: NativePiTerminalFailure = error): finally: await Coordination.run_worker(partial( record_acquisition_progress, self.bus.bus.path.parent, lease, - self.input_id, self.selected.cursor_delivery.measurements, + self.input_id, self.measurements, startup.native_process, )) async with self.selected.cursor_delivery.reading() as source_reads: await Coordination.run_read_async( @@ -241,29 +251,29 @@ def _start_once(self, identity: NativeSessionIdentity) -> None: subsequent owner loss permits replay. """ with ExitStack() as authority: - with self.selected.cursor_delivery.measurements.operation("coordinator_open"): + with self.measurements.operation("coordinator_open"): store = authority.enter_context(Coordination(str(self.store_path))) - with self.selected.cursor_delivery.measurements.operation("maintenance_ingress"): + with self.measurements.operation("maintenance_ingress"): authority.enter_context(MaintenanceBarrier( self.bus._registry.store.path ).allow_input()) - with self.selected.cursor_delivery.measurements.operation("registry_exclusion"): + with self.measurements.operation("registry_exclusion"): cut = authority.enter_context(self.bus._registry.store.current_snapshot()) registry = cut.snapshot - with self.selected.cursor_delivery.measurements.operation("prompt_binding"): + with self.measurements.operation("prompt_binding"): authority.enter_context(self.stage.bound_prompt( store, self.input_id, self.participant, self.wire_root_id, self.prompt.text, )) - with self.selected.cursor_delivery.measurements.operation("coordinator_exclusive"): + with self.measurements.operation("coordinator_exclusive"): db = authority.enter_context(store.session.writing()) - with self.selected.cursor_delivery.measurements.operation("selected_source_check"): + with self.measurements.operation("selected_source_check"): self.session.verify_start(identity) # From token consumption through COMMIT/fsync, every failure # remains uncertain and propagates without another send attempt. if not self._once.acquire(blocking=False): raise IdentityConflict("native send claim_scope cannot be reused") - with self.selected.cursor_delivery.measurements.operation("claim_checks"): + with self.measurements.operation("claim_checks"): self.owner.require_snapshot( registry, "recipient registry owner changed before native send" ) @@ -273,7 +283,7 @@ def _start_once(self, identity: NativeSessionIdentity) -> None: db, self.input_id, self.participant, self.token_digest ) self.stage.require_claim(store) - with self.selected.cursor_delivery.measurements.operation("native_claim_record"): + with self.measurements.operation("native_claim_record"): reserved.sent_owner_claimed_generation.record( reserved, db, self.owner.claimed_generation, identity ) diff --git a/src/agent_comms/publication_measurements.py b/src/agent_comms/publication_measurements.py new file mode 100644 index 0000000000..30685d8104 --- /dev/null +++ b/src/agent_comms/publication_measurements.py @@ -0,0 +1,50 @@ +"""Dependency-light resource timing values; never send or readiness authority.""" + +from __future__ import annotations + +import time +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager +from dataclasses import dataclass, field + + +@dataclass +class PublicationMeasurements: + """Bounded resource timings, not a message, phase or successful-send authority. + + First/last delimit observed bodies, including exceptional exits. Maxima and + totals remain counters; nested operations must not be added as disjoint work. + """ + count: int = 0 + total_ns: int = 0 + maximum_ns: int = 0 + maximum_started_ns: int = 0 + maximum_finished_ns: int = 0 + first_started_ns: int = 0 + last_finished_ns: int = 0 + operations: dict[str, PublicationMeasurements] = field(default_factory=dict) + + def operation(self, name: str) -> AbstractContextManager[None]: + """Borrow bounded counters for a declared acquisition/publication operation. + + These counters observe resources; they never decide readiness, acknowledgement + acceptance or replay. Owners call this with their fixed operation names. + """ + return self.operations.setdefault(name, PublicationMeasurements()).measuring() + + @contextmanager + def measuring(self) -> Iterator[None]: + started = time.monotonic_ns() + try: + yield + finally: + finished = time.monotonic_ns() + duration = finished - started + if self.count == 0 or started < self.first_started_ns: + self.first_started_ns = started + self.last_finished_ns = max(self.last_finished_ns, finished) + self.count += 1 + self.total_ns += duration + if duration > self.maximum_ns: + self.maximum_ns = duration + self.maximum_started_ns, self.maximum_finished_ns = started, finished diff --git a/src/agent_comms/tracked_turn.py b/src/agent_comms/tracked_turn.py index a0a117a2cd..b5156617e6 100644 --- a/src/agent_comms/tracked_turn.py +++ b/src/agent_comms/tracked_turn.py @@ -21,7 +21,7 @@ from .backend import TurnSession from .coordinator import Coordination from .errors import RelationViolationError -from .external_transcript import DraftRow +from .external_transcript import DraftPublication, DraftRow from .field_codec import InputId, MalformedRecord from .following_inputs import FollowingInputs from .maintenance_barrier import UNCOORDINATED_INGRESS, MaintenanceIngress @@ -33,6 +33,7 @@ NativeContextProof, NativeEmptyAnswer, NativePiPromptRejected, + NativePiRpcLaunch, NativeTurnResult, _verify_context, ) @@ -297,18 +298,21 @@ async def execute( # Independent execution acquires a fresh artifact. A selected stage # derives from its execution's original pre-claim factory; source/config and # child permissions remain fresh. Finish construction before acquiring the child. + def select_launch() -> NativePiRpcLaunch: + with measurements.operation("native_launch_selection_worker"): + return execution.tracked_launch( + package, + root=startup.root, + worktree=worktree, + session=session, + model=model, + thinking_level=thinking_level, + environment=environment, + selected_tool_mode=selected_tool_mode, + ) + with measurements.operation("native_launch_selection"): - launch = await Coordination.run_worker(partial( - execution.tracked_launch, - package, - root=startup.root, - worktree=worktree, - session=session, - model=model, - thinking_level=thinking_level, - environment=environment, - selected_tool_mode=selected_tool_mode, - )) + launch = await Coordination.run_worker(select_launch) turn = cls( launch, partial(commands.Prompt, @@ -344,9 +348,13 @@ async def complete(self) -> NativeTurnResult: async with AsyncExitStack() as transport: self.native = await self.acquire_native(transport, reuse=True) ownership.push_async_callback(self.retain_or_close) - self.draft_publication = await Coordination.run_worker(partial( - drafts.enter_context, DraftRow.publication( - self.startup.root, self.command.input_id))) + def acquire_draft() -> DraftPublication: + with self.startup.measurements.operation("draft_publication_worker"): + return drafts.enter_context(DraftRow.publication( + self.startup.root, self.command.input_id)) + + with self.startup.measurements.operation("draft_publication_acquisition"): + self.draft_publication = await Coordination.run_worker(acquire_draft) await ownership.enter_async_context(self.native.failures()) self.watchdog.reading() try: diff --git a/src/agent_comms/turn_progress.py b/src/agent_comms/turn_progress.py index ad4ebe1951..a148f9d1bf 100644 --- a/src/agent_comms/turn_progress.py +++ b/src/agent_comms/turn_progress.py @@ -24,7 +24,8 @@ from .comms import Comms from .context_store import UNREAD_CONTEXT, ContextPrior from .coordinator import Coordination -from .diagnostics import PublicationMeasurements, record_request_progress, record_terminal_failure +from .diagnostics import record_request_progress, record_terminal_failure +from .publication_measurements import PublicationMeasurements from .messages import AlertKind, NoticeRole from .send_result import SendResult from .thread_identity import ClaimedIdentity diff --git a/tests/test_native_send_admission.py b/tests/test_native_send_admission.py index 1425beabce..6305ee08ff 100644 --- a/tests/test_native_send_admission.py +++ b/tests/test_native_send_admission.py @@ -29,7 +29,7 @@ from agent_comms.coordination_database import CoordinationStore from agent_comms.coordinator import Coordination from agent_comms.declared_family import DeclaredFamily -from agent_comms.diagnostics import PublicationMeasurements +from agent_comms.publication_measurements import PublicationMeasurements from agent_comms.locked_store import RegistryFile from agent_comms.errors import RelationViolationError from agent_comms.native_prompt_send import (