fix(tool-executor): emit tool results on hard interrupt to keep alternation

The sequential executor's KeyboardInterrupt handlers emitted a cancelled
post-tool-call event for the current tool, called agent.interrupt(), then
re-raised — WITHOUT appending a tool result message for the interrupted call
or any remaining calls in the batch. The assistant tool-call turn was left
with no matching tool results, a message-role alternation violation that
malforms the next provider request (relying on downstream repair passes to
patch it, which don't run on every path).

The cooperative-interrupt block (_interrupt_requested) and the concurrent
executor already emit a result for every call_id; this brings the two hard-
interrupt handlers into line via a shared _append_cancelled_tool_results
helper that appends a cancelled result for the current + remaining calls
before re-raising.

Verified live before/after (0 tool results -> 3 for a 3-call batch
interrupted on the first tool) and with a sabotage-checked regression test.
52 interrupt/executor tests pass.

agent/tool_executor.py

@@ -1312,6 +1312,26 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
 
 
 
+def _append_cancelled_tool_results(messages: list, tool_calls, *, reason: str) -> None:
+    """Append a cancelled ``tool`` result for each call in ``tool_calls``.
+
+    Used when a hard interrupt (KeyboardInterrupt / BaseException) aborts the
+    sequential executor mid-batch. Without this, the loop re-raises leaving the
+    assistant tool-call turn with no matching tool results — a message-role
+    alternation violation that malforms the next provider request. Mirrors the
+    cooperative-interrupt skip block and the concurrent path, both of which
+    already emit a result for every call_id.
+    """
+    for tc in tool_calls:
+        name = getattr(getattr(tc, "function", None), "name", "") or "tool"
+        messages.append(make_tool_result_message(
+            name,
+            f"[Tool execution cancelled — {name} was skipped due to {reason}]",
+            getattr(tc, "id", "") or "",
+            effect_disposition="none",
+        ))
+
+
 def execute_tool_calls_sequential(agent, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0, *, finalize: bool = True) -> None:
     """Execute tool calls sequentially (original behavior). Used for single calls or interactive tools.
 
@@ -1723,6 +1743,14 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                     agent.interrupt("keyboard interrupt")
                 except Exception:
                     pass
+                # Emit a tool result for THIS call and every remaining call in
+                # the batch before re-raising, so the assistant tool-call turn
+                # is never left without matching tool results (alternation).
+                _append_cancelled_tool_results(
+                    messages,
+                    assistant_message.tool_calls[i - 1:],
+                    reason="keyboard interrupt",
+                )
                 raise
             except Exception as tool_error:
                 function_result = f"Error executing tool '{function_name}': {tool_error}"
@@ -1791,6 +1819,13 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                     agent.interrupt("keyboard interrupt")
                 except Exception:
                     pass
+                # Emit a tool result for THIS call and every remaining call in
+                # the batch before re-raising (see interactive branch above).
+                _append_cancelled_tool_results(
+                    messages,
+                    assistant_message.tool_calls[i - 1:],
+                    reason="keyboard interrupt",
+                )
                 raise
             except Exception as tool_error:
                 function_result = f"Error executing tool '{function_name}': {tool_error}"

tests/run_agent/test_tool_call_incremental_persistence.py

@@ -299,6 +299,48 @@ def test_execute_tool_calls_sequential_flushes_each_tool_result_before_next_disp
     ]
 
 
+def test_sequential_keyboard_interrupt_emits_results_for_all_calls():
+    """A KeyboardInterrupt mid-batch must not leave dangling tool_calls.
+
+    When a tool handler raises KeyboardInterrupt, the sequential executor
+    re-raises to abort the turn — but it must first append a tool result for
+    the interrupted call AND every remaining call, or the assistant tool-call
+    turn is left without matching tool results (a message-role alternation
+    violation that malforms the next provider request). Mirrors the
+    cooperative-interrupt and concurrent paths, which already do this.
+    """
+    agent = _make_agent()
+    tool_calls = [
+        _mock_tool_call(name="web_search", call_id="c1"),
+        _mock_tool_call(name="web_search", call_id="c2"),
+        _mock_tool_call(name="web_search", call_id="c3"),
+    ]
+    messages: list = []
+    assistant_message = SimpleNamespace(content="", tool_calls=tool_calls)
+
+    def _interrupt_dispatch(function_name, function_args, effective_task_id, **kwargs):
+        # First tool raises a hard interrupt mid-batch.
+        raise KeyboardInterrupt()
+
+    agent._flush_messages_to_session_db = MagicMock()
+
+    with (
+        patch("run_agent.handle_function_call", side_effect=_interrupt_dispatch),
+        patch(
+            "agent.tool_executor.maybe_persist_tool_result",
+            side_effect=lambda **kwargs: kwargs["content"],
+        ),
+        pytest.raises(KeyboardInterrupt),
+    ):
+        agent._execute_tool_calls_sequential(assistant_message, messages, "task-1")
+
+    # Every call_id has a matching tool result — alternation preserved.
+    tool_results = [m for m in messages if m.get("role") == "tool"]
+    assert [m["tool_call_id"] for m in tool_results] == ["c1", "c2", "c3"]
+    # The results are marked as cancelled, not fabricated successes.
+    assert all("cancelled" in m["content"].lower() for m in tool_results)
+
+
 @pytest.mark.parametrize("executor_mode", ["sequential", "concurrent"])
 def test_tool_result_is_durable_before_ui_completion_on_abnormal_exit(
     tmp_path,