fix(tracing): classify durable HITL pauses as suspended

Co-Authored-By: Claude <noreply@anthropic.com>

Author: 2 people

CHANGELOG.md

@@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **Durable HITL pauses now finalize traces as suspended rather than aborted.**
+  `Agent.detach()` commits the pending request first, then the owning activation
+  emits a terminal suspension event after runtime-state cleanup. Open agent,
+  turn, chat, and tool spans end with `cubepi.run.outcome="suspended"`; observer
+  failures cannot hide the terminal event, cancellation still propagates, and
+  `respond()` opens a distinct correlated activation trace.
+
 ## [0.13.3] - 2026-08-02
 
 ### Fixed

cubepi/agent/agent.py

@@ -50,6 +50,13 @@
 
 TMessage = TypeVar("TMessage")
 
+
+def _consume_control_exception(future: asyncio.Future[StructuredValue]) -> None:
+    """Mark a detach control exception retrieved without changing awaiters."""
+    if not future.cancelled():
+        future.exception()
+
+
 if TYPE_CHECKING:
     from cubepi.deferred.types import DeferredStrategy, DeferredToolGroup
     from cubepi.providers.fallback import FallbackBoundModel
@@ -287,6 +294,7 @@ async def _composed_resolver(tool_call, *, context, signal=None):
             and hasattr(self.checkpointer, "mark_run_complete")
         )
         self._channel = channel
+        self._pending_suspension_event: HitlRequest | None = None
         # _bind_emit is a _BaseChannel internal, not part of the HitlChannel
         # protocol. Third-party channels that only implement the public
         # protocol won't have it — skip the wiring instead of crashing.
@@ -384,6 +392,19 @@ async def _dispatch_outcome(self, outcome: RunOutcome | None, run_id: str) -> No
                 cause=exc,
             ) from exc
 
+    async def _emit_suspended_event(self, outcome: RunOutcome) -> None:
+        """Publish suspension only after the run transition is committed."""
+        if outcome != "suspended":
+            self._pending_suspension_event = None
+            return
+        pending = self._pending_suspension_event
+        if pending is None:
+            return
+        self._pending_suspension_event = None
+        from cubepi.agent.types import AgentSuspendedEvent
+
+        await self._process_event(AgentSuspendedEvent(pending_request=pending))
+
     def _validate_hitl_bindings(self, run_id: str | None, *, caller: str) -> None:
         """Reject HITL-bound tools/middleware that disagree with `run_id`.
 
@@ -518,6 +539,7 @@ async def prompt(
                 await self._run_prompt(messages)
         except BaseException:
             # Spec §3.7: leave active_run_id SET on failure.
+            self._pending_suspension_event = None
             raise
         else:
             outcome: RunOutcome = self._state.last_outcome or "abandoned"
@@ -526,6 +548,7 @@ async def prompt(
             # the clear line below is unreachable on the exception path.
             await self._dispatch_outcome(outcome, effective_run_id)
             self._state.active_run_id = None
+            await self._emit_suspended_event(outcome)
             return effective_run_id
 
     async def fork(
@@ -723,11 +746,13 @@ async def resume(self, *, run_id: str | None = None) -> str:
         except BaseException:
             # Spec §3.7 parity: leave active_run_id SET on failure so callers
             # can observe which run failed.
+            self._pending_suspension_event = None
             raise
         else:
             outcome: RunOutcome = self._state.last_outcome or "abandoned"
             await self._dispatch_outcome(outcome, effective_run_id)
             self._state.active_run_id = None
+            await self._emit_suspended_event(outcome)
             return effective_run_id
 
     def _build_stream_options(self, signal: asyncio.Event) -> StreamOptions:
@@ -803,8 +828,6 @@ def _create_context_snapshot(self) -> AgentContext:
         )
 
     async def detach(self) -> None:
-        from cubepi.agent.types import AgentSuspendedEvent
-
         if self._channel is None:
             raise HitlError("agent has no channel bound")
         pending = self._channel.pending
@@ -814,11 +837,14 @@ async def detach(self) -> None:
             or self._channel._future.done()
         ):
             return  # nothing to detach
-        # Emit the suspended event BEFORE triggering the exception, so listeners
-        # see the real pending payload (codex pass 2 BLOCKING: previous draft
-        # emitted from the loop with pending=None — fundamentally wrong).
-        await self._process_event(AgentSuspendedEvent(pending_request=pending))
-        self._channel._future.set_exception(HitlDetached())
+        # Snapshot the payload before the channel clears its in-memory pending slot,
+        # then commit the control-flow transition. The owning prompt/resume task
+        # publishes AgentSuspendedEvent only after it records the suspended outcome
+        # and clears active_run_id, so observers cannot report an uncommitted pause.
+        self._pending_suspension_event = pending
+        future = self._channel._future
+        future.add_done_callback(_consume_control_exception)
+        future.set_exception(HitlDetached())
 
     async def load_pending_hitl_request(self) -> HitlRequest | None:
         if self.checkpointer is None or self.thread_id is None:
@@ -898,14 +924,16 @@ async def respond(
                 await self._run_hitl_resume()
             except BaseException:
                 # Spec §3.7: leave active_run_id SET on raise.
+                self._pending_suspension_event = None
                 raise
             else:
                 # Legacy guard: pending persisted without run_id (older
                 # save_pending_request callers) cannot drive dispatch.
+                outcome: RunOutcome = self._state.last_outcome or "abandoned"
                 if recovered_run_id is not None:
-                    outcome: RunOutcome = self._state.last_outcome or "abandoned"
                     await self._dispatch_outcome(outcome, recovered_run_id)
                 self._state.active_run_id = None
+                await self._emit_suspended_event(outcome)
 
     async def abort_pending(
         self, reason: str = "aborted by host"
@@ -1233,7 +1261,23 @@ async def _process_event(self, event: AgentEvent) -> None:
         await self._emit_to_listeners(event)
 
     async def _emit_to_listeners(self, event: AgentEvent) -> None:
-        for listener in self._listeners:
-            result = listener(event, self._active_signal)
-            if asyncio.iscoroutine(result):
-                await result
+        cancellation: asyncio.CancelledError | None = None
+        first_error: Exception | None = None
+        for listener in tuple(self._listeners):
+            try:
+                result = listener(event, self._active_signal)
+                if asyncio.iscoroutine(result):
+                    await result
+            except asyncio.CancelledError as exc:
+                if cancellation is None:
+                    cancellation = exc
+            except Exception as exc:
+                # One observer must not hide an event from later observers.
+                # Preserve existing error propagation after every listener had
+                # the chance to see the same committed state transition.
+                if first_error is None:
+                    first_error = exc
+        if cancellation is not None:
+            raise cancellation
+        if first_error is not None:
+            raise first_error

cubepi/tracing/recorder.py

@@ -34,6 +34,7 @@
 from cubepi.agent.types import (
     AgentEndEvent,
     AgentStartEvent,
+    AgentSuspendedEvent,
     MessageEndEvent,
     MessageStartEvent,
     TurnEndEvent,
@@ -66,6 +67,7 @@
     CUBEPI_LLM_THINKING_LEVEL,
     CUBEPI_OUTPUT_MESSAGES_COUNT,
     CUBEPI_RUN_ID,
+    CUBEPI_RUN_OUTCOME,
     CUBEPI_TOOL_BLOCK_REASON,
     CUBEPI_TOOL_BLOCKED_BY_HOOK,
     CUBEPI_TOOL_EXECUTION_MODE,
@@ -413,6 +415,8 @@ async def _on_agent_event(self, event: Any, signal: Any | None = None) -> None:
                 self._on_message_end(event)
             elif isinstance(event, TurnEndEvent):
                 self._on_turn_end(event)
+            elif isinstance(event, AgentSuspendedEvent):
+                self._on_agent_suspended()
             elif isinstance(event, AgentEndEvent):
                 self._on_agent_end(event)
             # MessageUpdateEvent / ToolExecutionUpdateEvent: IGNORED.
@@ -679,6 +683,49 @@ def _on_agent_start(self) -> None:
             if history:
                 self._run.transcript.extend(history)
 
+    def _on_agent_suspended(self) -> None:
+        """Finalize a durable HITL pause without classifying it as abort."""
+        run = self._run
+        if run is None:
+            return
+        for span in list(run.tool_spans.values()):
+            try:
+                span.set_attribute(CUBEPI_RUN_OUTCOME, "suspended")
+                span.end()
+            except Exception:
+                pass
+        run.tool_spans.clear()
+        if run.chat_span is not None:
+            try:
+                run.chat_span.set_attribute(CUBEPI_RUN_OUTCOME, "suspended")
+                run.chat_span.end()
+            except Exception:
+                pass
+            run.chat_span = None
+            run.chat_open_ns = None
+            run.chat_first_chunk_recorded = False
+        if run.turn_span is not None:
+            try:
+                run.turn_span.set_attribute(CUBEPI_RUN_OUTCOME, "suspended")
+                run.turn_span.end()
+            except Exception:
+                pass
+            run.turn_span = None
+        run.agent_span.set_attribute(CUBEPI_RUN_OUTCOME, "suspended")
+        run.agent_span.set_attribute(
+            CUBEPI_OUTPUT_MESSAGES_COUNT, len(run.output_messages)
+        )
+        run.agent_span.end()
+        self._sweep_tool_span_tokens(run)
+        if run.stream_file is not None:
+            try:
+                run.stream_file.close()
+            except Exception:
+                pass
+            run.stream_file = None
+        self._reset_active_run()
+        self._run = None
+
     def _on_agent_end(self, event: AgentEndEvent) -> None:
         run = self._run
         if run is None:
@@ -760,9 +807,8 @@ def _on_turn_end(self, event: TurnEndEvent) -> None:
         if run is None or run.turn_span is None:
             return
         msg = event.message
-        # Track output messages: assistant + any tool_results from this turn.
-        run.turn_output_messages.append(msg)
-        run.output_messages.append(msg)
+        # Assistant output is captured at MessageEnd so a HITL suspension before
+        # TurnEnd still records it. Add only this turn's tool results here.
         for tr in getattr(event, "tool_results", []) or []:
             run.turn_output_messages.append(tr)
             run.output_messages.append(tr)
@@ -948,6 +994,8 @@ def _on_message_end(self, event: MessageEndEvent) -> None:
         msg = event.message
         if getattr(msg, "role", None) == "assistant":
             run.transcript.append(msg)
+            run.turn_output_messages.append(msg)
+            run.output_messages.append(msg)
 
     # ------------------------------------------------------------------
     # Provider listeners — drive the chat span lifetime

cubepi/tracing/schema.py

@@ -104,6 +104,7 @@
 CUBEPI_AGENT_SYSTEM_PROMPT_SHA256 = "cubepi.agent.system_prompt.sha256"
 CUBEPI_INPUT_MESSAGES_COUNT = "cubepi.input.messages.count"
 CUBEPI_OUTPUT_MESSAGES_COUNT = "cubepi.output.messages.count"
+CUBEPI_RUN_OUTCOME = "cubepi.run.outcome"
 CUBEPI_ABORTED = "cubepi.aborted"
 
 # Turn span attributes (cubepi.turn span — no gen_ai.operation.name)

dev/plans/2026-08-04-tracing-hitl-suspension.md

@@ -0,0 +1,74 @@
+# HITL Suspension Trace Semantics Implementation Plan
+
+**Goal:** Export durable HITL pauses as committed `suspended` activations,
+never as aborted, while preserving cancellation behavior and tracing cleanup.
+
+**Architecture:** Snapshot the pending request in `Agent.detach()`, commit the
+`HitlDetached` channel transition, and publish `AgentSuspendedEvent` from the
+owning run task after outcome dispatch and active-run cleanup. The recorder then
+finalizes the activation with explicit suspended attributes.
+
+**Tech stack:** Python 3.13, asyncio, FauxProvider, CheckpointedChannel,
+OpenTelemetry SDK, pytest, Ruff, mypy, uv.
+
+## Task 1: Pin the production-shaped regression
+
+- Add `tests/tracing/test_hitl_suspension.py` using a real Agent, HITL channel,
+  ask-user tool, Tracer, and in-memory exporter.
+- Record RED for the original false-abort behavior.
+- Add adversarial RED assertions for:
+  - event observers seeing pre-commit runtime state;
+  - cross-task `_active_run` retention;
+  - assistant output count incorrectly remaining zero;
+  - an earlier regular/cancelled listener hiding suspension from tracing;
+  - detach followed by immediate owner-task cancellation retaining the pending
+    payload and emitting an unhandled-future warning;
+  - same-tick `reset()` deleting a committed suspension snapshot before the
+    owner task can publish it.
+
+## Task 2: Commit suspension at the owning task boundary
+
+- Snapshot the pending request in `Agent.detach()` before the channel clears its
+  in-memory slot.
+- Set `HitlDetached` without publishing the terminal event from the host task.
+- After `prompt()`, `resume()`, or `respond()` records `outcome=suspended`,
+  performs outcome dispatch, and clears `active_run_id`, publish the event from
+  that same owning task.
+- Fan out agent events across regular exceptions and `CancelledError`; after
+  fan-out, give cancellation priority or re-raise the first regular exception.
+- Clear the pending snapshot on every failed owning activation; do not let
+  `reset()` clear a snapshot before that activation publishes its terminal.
+- Mark the detach control exception retrieved via a future done callback without
+  changing what channel awaiters receive.
+
+## Task 3: Add explicit trace semantics
+
+- Add the recorder-owned schema attribute `cubepi.run.outcome`.
+- Handle committed `AgentSuspendedEvent` in `Recorder`.
+- End open tool/chat/turn/root spans with outcome `suspended`.
+- Sweep tool-span registrations, close stream state, and clear the active run.
+- Move assistant output accumulation to `MessageEndEvent` so suspension before
+  `TurnEndEvent` still records the partial output.
+- Do not change `_close_open_spans()` cancellation semantics.
+
+## Task 4: Document the lifecycle contract
+
+- Update tracing guidance to distinguish cancellation from durable suspension.
+- Update HITL reference/durable examples to describe post-commit event timing.
+- State that resume creates a new activation trace.
+
+## Task 5: Verify
+
+Run:
+
+```bash
+uv run pytest tests/tracing/test_hitl_suspension.py -q
+uv run pytest tests/tracing tests/hitl -q
+uv run pytest tests/
+uv run ruff check cubepi/ tests/
+uv run ruff format --check cubepi/ tests/
+uv run mypy cubepi
+```
+
+Inspect `git diff --check` and confirm the branch contains only the spec, plan,
+focused tests, lifecycle fix, recorder/schema fix, and user-facing docs.