Measure selected native acquisition in its own input lifetime

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

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,

src/agent_comms/cursor_publication.py

@@ -1,3 +1,3 @@
 """Read durable native evidence and order its informational ACP publication.
 
 Delivery bookkeeping belongs to the connection, never to input permission or
@@ -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

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

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

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

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 @@
         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(

src/agent_comms/native_startup.py

@@ -1,3 +1,3 @@
 """Native child startup resources and readiness bounds."""
 
 from __future__ import annotations
@@ -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 @@
 
 @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(

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,27 +122,31 @@ 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,
             input_id=input_id,
             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
                 )