fix(agent): address codex review — carry salvaged results, strict suspend persistence, propagate emitter failures

- HitlControlException gains partial_tool_results; both executors attach
  the already-persisted sibling ToolResultMessages, and the stateless
  loop HITL handlers append them to the returned message lists, so
  callers persisting run_agent_loop*'s return value keep completed
  siblings' results across a suspend.
- Sibling persistence before a suspend is no longer best-effort: an
  emit/checkpoint failure propagates (run fails rather than durably
  suspending with completed work missing). The cancel salvage path stays
  best-effort — CancelledError must reach the Agent backfill.
- Non-control exceptions reaching batch classification are framework
  failures (emit_fn raising in a worker), not tool failures: re-raised
  after all tasks settle instead of being masked as synthetic
  tool_results.

Author: claude

cubepi/agent/loop.py

@@ -203,17 +203,25 @@ async def run_agent_loop_resume(
             new_messages=new_messages,
             set_outcome=set_outcome,
         )
-    except HitlDetached:  # pragma: no cover — E2E tested
+    except HitlDetached as exc:  # pragma: no cover — E2E tested
         # The Agent caller (Agent.detach) emitted AgentSuspendedEvent already.
         # Loop exits silently — assistant message and pending state remain
-        # intact for the next respond() call.
+        # intact for the next respond() call. Sibling tool_results that
+        # completed before the detach were emitted by the executor; append
+        # them so callers persisting the returned messages keep them.
+        for msg in exc.partial_tool_results:
+            context.messages.append(msg)
+            new_messages.append(msg)
         if set_outcome is not None:
             set_outcome("suspended")
         return new_messages
-    except HitlAborted:  # pragma: no cover — E2E tested
+    except HitlAborted as exc:  # pragma: no cover — E2E tested
         # The Agent caller (Agent.abort_pending) emitted AgentAbortedEvent
         # already. Loop exits silently — synthetic deny + terminal aborted
         # assistant message already appended by abort_pending.
+        for msg in exc.partial_tool_results:
+            context.messages.append(msg)
+            new_messages.append(msg)
         if set_outcome is not None:
             set_outcome("abandoned")
         return new_messages
@@ -442,17 +450,26 @@ async def _run_loop(
             emit=emit,
             set_outcome=set_outcome,
         )
-    except HitlDetached:
+    except HitlDetached as exc:
         # AgentSuspendedEvent already emitted by Agent.detach() — exit silently
         # so AgentEndEvent doesn't double-signal termination. Assistant message
         # and pending state remain intact for the next respond() call.
+        # Sibling tool_results that completed before the detach were emitted
+        # (and checkpointed) by the executor; append them so callers reading
+        # current_context/new_messages keep them too.
+        for msg in exc.partial_tool_results:
+            current_context.messages.append(msg)
+            new_messages.append(msg)
         if set_outcome is not None:
             set_outcome("suspended")
         return
-    except HitlAborted:
+    except HitlAborted as exc:
         # AgentAbortedEvent already emitted by Agent.abort_pending() — exit
         # silently. Synthetic deny + terminal aborted assistant message are
         # already appended; conversation is closed.
+        for msg in exc.partial_tool_results:
+            current_context.messages.append(msg)
+            new_messages.append(msg)
         if set_outcome is not None:
             set_outcome("abandoned")
         return

cubepi/agent/tools.py

@@ -474,70 +474,79 @@ async def _execute_sequential(
     finalized_list: list[_FinalizedOutcome] = []
     messages: list[ToolResultMessage] = []
 
-    for idx, tc in enumerate(tool_calls):
-        if preresolved is not None:
-            rtc, was_resolved, resolver_error = preresolved[idx]
-        else:
-            # Lazy: resolve only when this call's turn comes, after every
-            # earlier call in the batch has fully executed.
-            rtc, was_resolved, resolver_error = await _resolve_tool_call(
-                tc, context, resolve_tool_call, signal
-            )
-
-        await emit_event(
-            emit_fn,
-            ToolExecutionStartEvent(
-                tool_call_id=rtc.id, tool_name=rtc.name, args=rtc.arguments
-            ),
-        )
-
-        preparation = resolver_error or await _prepare_tool_call(
-            context,
-            assistant_message,
-            rtc,
-            before_tool_call,
-            signal,
-            resolved=was_resolved,
-        )
+    try:
+        for idx, tc in enumerate(tool_calls):
+            if preresolved is not None:
+                rtc, was_resolved, resolver_error = preresolved[idx]
+            else:
+                # Lazy: resolve only when this call's turn comes, after every
+                # earlier call in the batch has fully executed.
+                rtc, was_resolved, resolver_error = await _resolve_tool_call(
+                    tc, context, resolve_tool_call, signal
+                )
 
-        if isinstance(preparation, _ImmediateOutcome):
-            finalized = _FinalizedOutcome(
-                tool_call=rtc,
-                result=preparation.result,
-                is_error=preparation.is_error,
-                blocked_by_hook=preparation.blocked_by_hook,
-                block_reason=preparation.block_reason,
-                hitl_trace=preparation.hitl_trace,
+            await emit_event(
+                emit_fn,
+                ToolExecutionStartEvent(
+                    tool_call_id=rtc.id, tool_name=rtc.name, args=rtc.arguments
+                ),
             )
-        else:
-            result, is_error = await _execute_prepared(preparation, signal, emit_fn)
-            finalized = await _finalize(
+
+            preparation = resolver_error or await _prepare_tool_call(
                 context,
                 assistant_message,
-                preparation,
-                result,
-                is_error,
-                after_tool_call,
+                rtc,
+                before_tool_call,
                 signal,
+                resolved=was_resolved,
             )
 
-        await emit_event(
-            emit_fn,
-            ToolExecutionEndEvent(
-                tool_call_id=rtc.id,
-                tool_name=rtc.name,
-                result=finalized.result,
-                is_error=finalized.is_error,
-                terminate=bool(finalized.result.terminate),
-                blocked_by_hook=finalized.blocked_by_hook,
-                block_reason=finalized.block_reason,
-            ),
-        )
-        tool_msg = _make_tool_result_message(finalized)
-        await emit_event(emit_fn, MessageStartEvent(message=tool_msg))
-        await emit_event(emit_fn, MessageEndEvent(message=tool_msg))
-        finalized_list.append(finalized)
-        messages.append(tool_msg)
+            if isinstance(preparation, _ImmediateOutcome):
+                finalized = _FinalizedOutcome(
+                    tool_call=rtc,
+                    result=preparation.result,
+                    is_error=preparation.is_error,
+                    blocked_by_hook=preparation.blocked_by_hook,
+                    block_reason=preparation.block_reason,
+                    hitl_trace=preparation.hitl_trace,
+                )
+            else:
+                result, is_error = await _execute_prepared(preparation, signal, emit_fn)
+                finalized = await _finalize(
+                    context,
+                    assistant_message,
+                    preparation,
+                    result,
+                    is_error,
+                    after_tool_call,
+                    signal,
+                )
+
+            await emit_event(
+                emit_fn,
+                ToolExecutionEndEvent(
+                    tool_call_id=rtc.id,
+                    tool_name=rtc.name,
+                    result=finalized.result,
+                    is_error=finalized.is_error,
+                    terminate=bool(finalized.result.terminate),
+                    blocked_by_hook=finalized.blocked_by_hook,
+                    block_reason=finalized.block_reason,
+                ),
+            )
+            tool_msg = _make_tool_result_message(finalized)
+            await emit_event(emit_fn, MessageStartEvent(message=tool_msg))
+            await emit_event(emit_fn, MessageEndEvent(message=tool_msg))
+            finalized_list.append(finalized)
+            messages.append(tool_msg)
+
+    except HitlControlException as exc:
+        # Completed calls' results were already emitted (and
+        # checkpointed) per-iteration; carry them on the exception so
+        # the stateless loop entry points can return them to callers
+        # that persist the loop's return value across the suspend.
+        exc.partial_tool_results = tuple(messages)
+        raise
 
     return ToolCallBatch(messages=messages, terminate=_should_terminate(finalized_list))
 
@@ -687,6 +696,7 @@ async def _run(prep: _PreparedToolCall) -> _FinalizedOutcome:
 
     finalized_list: list[_FinalizedOutcome] = []
     control_exc: BaseException | None = None
+    framework_exc: BaseException | None = None
     for entry, slot in zip(entries, scheduled):
         if isinstance(slot, _FinalizedOutcome):
             finalized_list.append(slot)
@@ -710,18 +720,23 @@ async def _run(prep: _PreparedToolCall) -> _FinalizedOutcome:
             if control_exc is None:
                 control_exc = exc
             continue
+        if not isinstance(exc, asyncio.CancelledError):
+            # Tool and hook failures are converted to error results at the
+            # source (_execute_prepared/_finalize), so anything else landing
+            # here is framework-side — typically emit_fn raising while
+            # processing a Start/End event. Emitter failures propagate at
+            # every other emit_event call site; synthesizing a bogus
+            # tool_result here would let the run continue with event
+            # processing broken. Deferred until every task has settled.
+            if framework_exc is None:
+                framework_exc = exc
+            continue
         # Per-task isolation: a stray CancelledError (tool self-cancel with
-        # no outer cancel — a tool bug) or any exception that slipped past
-        # _execute_prepared/_finalize degrades to an error result for THIS
-        # call only.
-        text = (
-            "[Tool execution cancelled]"
-            if isinstance(exc, asyncio.CancelledError)
-            else str(exc)
-        )
+        # no outer cancel — a tool bug) degrades to an error result for
+        # THIS call only.
         synthesized = _FinalizedOutcome(
             tool_call=entry.tool_call,
-            result=_error_result(text),
+            result=_error_result("[Tool execution cancelled]"),
             is_error=True,
             hitl_trace=entry.hitl_trace,
         )
@@ -739,17 +754,27 @@ async def _run(prep: _PreparedToolCall) -> _FinalizedOutcome:
         )
         finalized_list.append(synthesized)
 
+    if framework_exc is not None:
+        # Event processing is broken, so emitting sibling results below
+        # would fail too. Every task has already settled (no leaks) —
+        # propagate, taking precedence over a suspend: durably suspending
+        # a run whose event pipeline is failing is not safe.
+        raise framework_exc
+
     if control_exc is not None:
-        # Persist the siblings (each MessageEndEvent checkpoints
-        # immediately), best-effort, then let the control exception
-        # propagate exactly as the suspend/abort machinery expects.
-        try:
-            await _emit_tool_result_messages(finalized_list, emit_fn)
-        except Exception:
-            # Best-effort: sibling persistence must never swallow the
-            # control exception — the suspend/abort machinery depends on
-            # it propagating; unanswered ids are backfilled on resume.
-            pass
+        # Persist the siblings BEFORE propagating the suspend (each
+        # MessageEndEvent checkpoints immediately). Deliberately NOT
+        # best-effort: suspending after a sibling's persistence failed
+        # would durably record a batch whose completed work is missing —
+        # resume would re-run tools whose side effects already happened.
+        # A persistence failure propagates instead (consistent with every
+        # other MessageEndEvent site): the run fails rather than suspends.
+        emitted = await _emit_tool_result_messages(finalized_list, emit_fn)
+        if isinstance(control_exc, HitlControlException):
+            # Stateless loop entry points append these to the message
+            # lists they return, so callers persisting the return value
+            # keep the completed siblings' results across the suspend.
+            control_exc.partial_tool_results = tuple(emitted)
         raise control_exc
 
     messages = await _emit_tool_result_messages(finalized_list, emit_fn)

cubepi/hitl/exceptions.py

@@ -1,14 +1,29 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING, Sequence
+
+if TYPE_CHECKING:  # pragma: no cover
+    from cubepi.providers.base import ToolResultMessage
+
 
 class HitlControlException(BaseException):
     """Base for HITL control-flow exceptions.
 
     Inherits BaseException so existing `except Exception:` handlers in
     cubepi.agent.tools._prepare_tool_call and _execute_prepared do NOT
     swallow these — mirrors asyncio.CancelledError.
+
+    ``partial_tool_results``: ToolResultMessages of sibling tool calls in
+    the same batch that completed (and were emitted/checkpointed) before
+    this control exception suspended the run. The tool executors set it so
+    the stateless loop entry points can append those results to the
+    message lists they return — a caller persisting the loop's return
+    value must not lose a completed sibling's result just because another
+    call in the batch detached.
     """
 
+    partial_tool_results: Sequence[ToolResultMessage] = ()
+
 
 class HitlCancelled(HitlControlException):
     def __init__(self, reason: str):

dev/specs/2026-07-05-tool-batch-fault-isolation.md

@@ -121,20 +121,34 @@ the bare `await` loop. Every task settles before any outcome is processed;
 results stay in tool_call order. Then classify each slot:
 
 1. **`_FinalizedOutcome`** — success path, unchanged.
-2. **Ordinary failure** (any exception that is not `HitlControlException`,
-   `CancelledError`, `KeyboardInterrupt`, or `SystemExit`) — synthesize an
-   error outcome for that tool_call id:
-   `AgentToolResult(content=[TextContent(text=str(exc))], is_error=True)`,
-   plus a paired `ToolExecutionEndEvent` (the Start was emitted inside
-   `_run`; an unpaired Start would leave `state.pending_tool_calls` and trace
-   spans open — same invariant the prepare-phase comment protects).
-   With Part 2 in place this is defense-in-depth: tool-body and hook failures
-   are already converted inside the task.
+2. **Framework failure** (any exception that is not `HitlControlException`,
+   `CancelledError`, `KeyboardInterrupt`, or `SystemExit`) — with Part 2 in
+   place, tool-body and hook failures are converted to error results *at the
+   source*, inside the task. So an exception landing in the classification
+   stage is framework-side — in practice `emit_fn` raising while processing
+   a Start/End event. Emitter failures propagate at every other
+   `emit_event` call site, and synthesizing a bogus `tool_result` for one
+   would let the run continue with event processing broken; so it is
+   recorded and **re-raised after every task has settled** (no leaks, no
+   conversion). A framework failure takes precedence over a pending
+   suspend: durably suspending a run whose event pipeline is failing is not
+   safe.
 3. **`HitlControlException`** — the suspend contract requires propagation, so
    it cannot become an error result. Order of operations: first emit the
    `ToolResultMessage`s for every *other* settled slot (each `MessageEndEvent`
    checkpoints immediately via `Agent._process_event`, `agent.py:1207-1209`),
-   then re-raise the HITL exception. The HITL call's own tool_call
+   then re-raise the HITL exception. This emission is deliberately **not**
+   best-effort: if a sibling's persistence fails, suspending anyway would
+   durably record a batch whose completed work is missing (resume would
+   re-run side-effecting tools), so the persistence failure propagates and
+   the run fails rather than suspends. The emitted sibling results are also
+   attached to the exception as
+   `HitlControlException.partial_tool_results`; the stateless loop entry
+   points (`_run_loop` / `run_agent_loop_resume` HITL handlers) append them
+   to the message lists they return, so callers persisting the loop's
+   return value — who never see the raised-through batch — keep the
+   completed siblings' results. (The sequential executor attaches the same
+   attribute for its already-emitted prefix.) The HITL call's own tool_call
    deliberately stays unanswered and gets no End event — identical to today's
    sequential detach shape — and the existing detach/resume/abort paths
    answer it (synthetic-deny backfill scans unanswered ids of the last