fix(agent): isolate parallel tool-batch faults + typed checkpoint corruption

One escaping exception in a parallel tool batch (HitlControlException from
a tool body, CancelledError, or a BaseException from after_tool_call) used
to abort the bare-await collection loop: every ToolResultMessage in the
batch was dropped — including succeeded siblings — leaving dangling
tool_calls in the checkpoint (providers 400 on every later turn) and
leaking still-running sibling tasks whose side effects then get duplicated
on resume.

- _execute_parallel now settles every task via gather(return_exceptions=
  True), synthesizes per-call error results (with paired End events) for
  ordinary failures, and for HITL/outer-cancel emits all settled sibling
  results BEFORE re-raising, so the suspend/backfill contracts keep working
  on an intact transcript.
- _execute_prepared/_finalize catch BaseException (re-raising HITL/Cancel/
  KeyboardInterrupt/SystemExit) so hook and tool failures degrade to error
  results at the source.
- Checkpointer load paths (sqlite/postgres/mysql) wrap per-row
  deserialization in CheckpointCorruptionError carrying thread_id/backend/
  row_ref; unknown roles are now corruption in all three backends
  (sqlite previously passed raw dicts through silently).

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

Author: claude

cubepi/agent/tools.py

@@ -298,9 +298,17 @@ async def _execute_prepared(
             # AgentToolResult(is_error=True) without raising must surface here
             # (otherwise the model sees a successful result).
             return result, bool(result.is_error)
-        except HitlControlException:
+        except (
+            HitlControlException,
+            asyncio.CancelledError,
+            KeyboardInterrupt,
+            SystemExit,
+        ):
             raise
-        except Exception as exc:
+        except BaseException as exc:
+            # BaseException, not Exception: a tool body raising a bare
+            # BaseException subclass must degrade to an error result too —
+            # anything that escapes here detonates the whole batch.
             return _error_result(str(exc)), True
     finally:
         if token is not None:
@@ -350,7 +358,14 @@ async def _finalize(
                     if after_result.is_error is not None
                     else is_error
                 )
-        except Exception as exc:
+        except (
+            HitlControlException,
+            asyncio.CancelledError,
+            KeyboardInterrupt,
+            SystemExit,
+        ):
+            raise
+        except BaseException as exc:
             result = _error_result(str(exc))
             is_error = True
 
@@ -628,24 +643,120 @@ async def _run(prep: _PreparedToolCall) -> _FinalizedOutcome:
         return fin
 
     # Now that every prepare has succeeded, fan out the executions.
+    # `scheduled` stays index-aligned with `entries` so a failed task's
+    # _PreparedToolCall (tool_call id/name, hitl_trace) is recoverable when
+    # synthesizing its error outcome.
     scheduled: list[_FinalizedOutcome | asyncio.Task] = [
         entry
         if isinstance(entry, _FinalizedOutcome)
         else asyncio.create_task(_run(entry))
         for entry in entries
     ]
+    tasks = [s for s in scheduled if isinstance(s, asyncio.Task)]
+
+    # Settle EVERY task before processing any outcome: one failing tool must
+    # never drop its siblings' results (they were computed, and their side
+    # effects happened — losing the ToolResultMessage would leave dangling
+    # tool_calls in the checkpoint and duplicate the side effects on resume).
+    try:
+        await asyncio.gather(*tasks, return_exceptions=True)
+    except asyncio.CancelledError:
+        # Outer cancel: gather already propagated it to the children.
+        # Settle them, salvage the results of tools that completed before
+        # the cancel landed, then re-raise. The Agent layer backfills the
+        # genuinely-unanswered ids (_complete_cancelled_tool_calls); without
+        # the salvage it would stamp "[cancelled]" over real, side-effecting
+        # completions. Best-effort: never mask the CancelledError.
+        try:
+            await asyncio.gather(*tasks, return_exceptions=True)
+            salvaged = [
+                s if isinstance(s, _FinalizedOutcome) else s.result()
+                for s in scheduled
+                if isinstance(s, _FinalizedOutcome)
+                or (s.done() and not s.cancelled() and s.exception() is None)
+            ]
+            await _emit_tool_result_messages(salvaged, emit_fn)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            pass
+        raise
+
     finalized_list: list[_FinalizedOutcome] = []
-    for entry in scheduled:
-        if isinstance(entry, asyncio.Task):
-            finalized_list.append(await entry)
+    control_exc: BaseException | None = None
+    for entry, slot in zip(entries, scheduled):
+        if isinstance(slot, _FinalizedOutcome):
+            finalized_list.append(slot)
+            continue
+        if slot.cancelled():
+            # Task.exception() would re-raise the CancelledError; outer
+            # cancellation re-raised above, so this is a tool self-cancel.
+            exc: BaseException | None = asyncio.CancelledError()
         else:
-            finalized_list.append(entry)
+            exc = slot.exception()
+        if exc is None:
+            finalized_list.append(slot.result())
+            continue
+        assert isinstance(entry, _PreparedToolCall)
+        if isinstance(exc, (HitlControlException, KeyboardInterrupt, SystemExit)):
+            # Must propagate (suspend/interpreter contracts) — but only
+            # after every sibling result below has been emitted. The call
+            # itself deliberately stays unanswered, with no End event:
+            # same shape as a sequential detach; the HITL resume/abort
+            # paths backfill it. First raiser (batch order) wins.
+            if control_exc is None:
+                control_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)
+        )
+        synthesized = _FinalizedOutcome(
+            tool_call=entry.tool_call,
+            result=_error_result(text),
+            is_error=True,
+            hitl_trace=entry.hitl_trace,
+        )
+        # Pair the Start emitted inside _run — an unpaired Start leaves
+        # state.pending_tool_calls and trace spans open.
+        await emit_event(
+            emit_fn,
+            ToolExecutionEndEvent(
+                tool_call_id=entry.tool_call.id,
+                tool_name=entry.tool_call.name,
+                result=synthesized.result,
+                is_error=True,
+                terminate=False,
+            ),
+        )
+        finalized_list.append(synthesized)
 
+    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:
+            pass
+        raise control_exc
+
+    messages = await _emit_tool_result_messages(finalized_list, emit_fn)
+    return ToolCallBatch(messages=messages, terminate=_should_terminate(finalized_list))
+
+
+async def _emit_tool_result_messages(
+    finalized_list: list[_FinalizedOutcome], emit_fn: Callable
+) -> list[ToolResultMessage]:
     messages: list[ToolResultMessage] = []
     for finalized in finalized_list:
         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))
         messages.append(tool_msg)
-
-    return ToolCallBatch(messages=messages, terminate=_should_terminate(finalized_list))
+    return messages

cubepi/checkpointer/exceptions.py

@@ -85,6 +85,34 @@ class CheckpointerLockTimeoutError(CheckpointerError):
     (SQLite busy_timeout, etc.)."""
 
 
+class CheckpointCorruptionError(CheckpointerError):
+    """A persisted message row failed to deserialize during load.
+
+    Raised instead of the raw decoder/validation error so hosts can
+    distinguish data corruption from operational failures and locate the
+    bad row (``row_ref`` names the backend table + primary key). Covers
+    bad payload encoding, schema-invalid message data, and unknown roles.
+    """
+
+    def __init__(
+        self,
+        *,
+        thread_id: str,
+        backend: str,
+        row_ref: str,
+        cause: BaseException,
+    ) -> None:
+        super().__init__(
+            f"corrupt checkpoint row for thread {thread_id!r} "
+            f"({backend}, {row_ref}): {cause}"
+        )
+        self.thread_id = thread_id
+        self.backend = backend
+        self.row_ref = row_ref
+        self.__cause__ = cause
+        self.__suppress_context__ = True
+
+
 class CompletionMarkerFailedError(CheckpointerError):
     """mark_run_complete() failed AFTER the run's final append succeeded.
     Carries `run_id` so callers using prompt(run_id=None) can recover

cubepi/checkpointer/mysql/checkpointer.py

@@ -20,6 +20,7 @@
 
 from cubepi.checkpointer.base import CheckpointData
 from cubepi.checkpointer.exceptions import (
+    CheckpointCorruptionError,
     RunAlreadyClaimedError,
     RunAlreadyCompletedError,
     RunNotClaimedError,
@@ -64,6 +65,26 @@ def _role_of(msg: Message) -> str:
 }
 
 
+def _deserialize_row(
+    thread_id: str, seq: int, role: str, metadata: Any, payload: Any
+) -> Message:
+    """Deserialize one cubepi_messages row; corruption raises typed."""
+    try:
+        cls = _ROLE_TO_CLS.get(role)
+        if cls is None:
+            raise ValueError(f"unknown role in DB: {role!r}")
+        data = msgpack.unpackb(bytes(payload), raw=False)
+        data["metadata"] = _decode_json(metadata)
+        return cls.model_validate(data)
+    except Exception as exc:
+        raise CheckpointCorruptionError(
+            thread_id=thread_id,
+            backend="mysql",
+            row_ref=f"cubepi_messages.seq={seq}",
+            cause=exc,
+        ) from exc
+
+
 def _parse_dsn(dsn: str) -> dict[str, Any]:
     """Parse a mysql:// URL into aiomysql.create_pool kwargs."""
     u = urlparse(dsn)
@@ -209,14 +230,10 @@ async def load(self, thread_id: str) -> CheckpointData | None:
         if not msg_rows and extra_row is None:
             return None
 
-        messages: list[Message] = []
-        for _seq, role, metadata, payload in msg_rows:
-            cls = _ROLE_TO_CLS.get(role)
-            if cls is None:
-                raise ValueError(f"unknown role in DB: {role!r}")
-            data = msgpack.unpackb(bytes(payload), raw=False)
-            data["metadata"] = _decode_json(metadata)
-            messages.append(cls.model_validate(data))
+        messages: list[Message] = [
+            _deserialize_row(thread_id, seq, role, metadata, payload)
+            for seq, role, metadata, payload in msg_rows
+        ]
 
         parent_thread_id: str | None = None
         if extra_row is not None:
@@ -472,15 +489,10 @@ async def snapshot(self, thread_id: str, *, after_run_id: str) -> list[Message]:
                     (thread_id, thread_id, cutoff),
                 )
                 rows = await cur.fetchall()
-        messages: list[Message] = []
-        for _seq, role, metadata, payload in rows:
-            cls = _ROLE_TO_CLS.get(role)
-            if cls is None:
-                raise ValueError(f"unknown role in DB: {role!r}")
-            data = msgpack.unpackb(bytes(payload), raw=False)
-            data["metadata"] = _decode_json(metadata)
-            messages.append(cls.model_validate(data))
-        return messages
+        return [
+            _deserialize_row(thread_id, seq, role, metadata, payload)
+            for seq, role, metadata, payload in rows
+        ]
 
     async def fork(
         self,

cubepi/checkpointer/postgres/checkpointer.py

@@ -16,6 +16,7 @@
 
 from cubepi.checkpointer.base import CheckpointData
 from cubepi.checkpointer.exceptions import (
+    CheckpointCorruptionError,
     RunAlreadyClaimedError,
     RunAlreadyCompletedError,
     RunNotClaimedError,
@@ -88,6 +89,29 @@ def _role_of(msg: Message) -> str:
 }
 
 
+def _deserialize_row(thread_id: str, r: Any) -> Message:
+    """Deserialize one cubepi_messages row; corruption raises typed."""
+    try:
+        cls = _ROLE_TO_CLS.get(r["role"])
+        if cls is None:
+            raise ValueError(f"unknown role in DB: {r['role']!r}")
+        data = msgpack.unpackb(bytes(r["payload"]), raw=False)
+        # The DB metadata column is the source of truth for Message.metadata.
+        # (payload also contains it, but column is the canonical view for querying.)
+        raw_meta = r["metadata"]
+        data["metadata"] = (
+            json.loads(raw_meta) if isinstance(raw_meta, str) else (raw_meta or {})
+        )
+        return cls.model_validate(data)
+    except Exception as exc:
+        raise CheckpointCorruptionError(
+            thread_id=thread_id,
+            backend="postgres",
+            row_ref=f"cubepi_messages.seq={r['seq']}",
+            cause=exc,
+        ) from exc
+
+
 class PostgresCheckpointer:
     """Checkpointer backed by PostgreSQL.
 
@@ -168,19 +192,7 @@ async def load(self, thread_id: str) -> CheckpointData | None:
         if not msg_rows and extra_row is None:
             return None
 
-        messages: list[Message] = []
-        for r in msg_rows:
-            cls = _ROLE_TO_CLS.get(r["role"])
-            if cls is None:
-                raise ValueError(f"unknown role in DB: {r['role']!r}")
-            data = msgpack.unpackb(bytes(r["payload"]), raw=False)
-            # The DB metadata column is the source of truth for Message.metadata.
-            # (payload also contains it, but column is the canonical view for querying.)
-            raw_meta = r["metadata"]
-            data["metadata"] = (
-                json.loads(raw_meta) if isinstance(raw_meta, str) else (raw_meta or {})
-            )
-            messages.append(cls.model_validate(data))
+        messages: list[Message] = [_deserialize_row(thread_id, r) for r in msg_rows]
 
         parent_thread_id: str | None = None
         if extra_row is not None:
@@ -388,18 +400,7 @@ async def snapshot(self, thread_id: str, *, after_run_id: str) -> list[Message]:
                 thread_id,
                 cutoff,
             )
-        messages: list[Message] = []
-        for r in rows:
-            cls = _ROLE_TO_CLS.get(r["role"])
-            if cls is None:
-                raise ValueError(f"unknown role in DB: {r['role']!r}")
-            data = msgpack.unpackb(bytes(r["payload"]), raw=False)
-            raw_meta = r["metadata"]
-            data["metadata"] = (
-                json.loads(raw_meta) if isinstance(raw_meta, str) else (raw_meta or {})
-            )
-            messages.append(cls.model_validate(data))
-        return messages
+        return [_deserialize_row(thread_id, r) for r in rows]
 
     async def fork(
         self,