fix(email): don't cancel a retrying agent for a recoverable tool error (#2572)

When the model calls a batch tool with a bad argument (e.g.
`archive_message_batch` with a stray `mailbox` kwarg), the agent loop
correctly rejects it and starts retrying — but the SSE layer couldn't
tell that per-tool retry apart from a genuinely fatal failure, so it
ended the response and cancelled the still-retrying agent, dead-ending
the turn with no answer and no stats line. Reproduced 4/4 on-hardware
before this fix. Now a recoverable error surfaces as a non-terminal
status line (the user still sees the failure) and the retry can reach
completion.

Closes #2515

## Test plan

- [x] `python -m pytest tests/unit -k "sse or agent_error" -q` — 399
passed
- [x] `python -m pytest hub/agents/email/python/tests -k "sse or
translation or query_routes" -q` — 64 passed
- [x] New regression test drives the real `/v1/email/query` route + real
`SSEOutputHandler` + real translation layer with a fake agent that emits
a recoverable per-tool error then retries: asserts `run.cancel_event`
and `handler.cancelled` stay unset mid-run and the stream reaches
`final` with both the failed and retried tool calls streamed
(`test_recoverable_tool_error_does_not_terminate_stream_or_cancel_run`)
- [x] `python util/lint.py --all` — all blocking checks pass

Author: itomek

hub/agents/email/python/CHANGELOG.md

@@ -84,6 +84,16 @@ contract version is tracked separately as
 
 ### Fixed
 
+- **A batch-tool retry no longer gets killed mid-recovery by the streaming
+  layer (#2515).** When the model called a batch tool with a spurious extra
+  argument (e.g. `archive_message_batch` with a stray `mailbox` kwarg), the
+  agent loop correctly rejected it and started retrying — but the SSE layer
+  couldn't tell that per-tool error apart from a genuinely fatal failure, so
+  it ended the response and cancelled the still-retrying agent, dead-ending
+  the turn with no answer and no stats line. `print_error` now carries a
+  `recoverable` flag through to the wire; a recoverable error folds to a
+  non-terminal status line instead of a terminal `error`, so the retry can
+  reach completion and the user still sees the failure as it happens.
 - **A failed memory startup is now visible in chat, and blames the right
   cause (#2519).** When the embedding model wasn't reachable, memory quietly
   disabled itself: a log line and a REST field said so, but the agent's

hub/agents/email/python/gaia_agent_email/sse_translation.py

@@ -29,10 +29,13 @@
 - **Buffer ``tool_start`` + ``tool_args`` into one ``tool_call``** (spec §6.3): the
   handler emits the name first and the arguments separately; the canonical
   ``tool_call`` carries ``{tool, args}`` together.
-- **Fail loudly, never silently.** ``agent_error`` and a governance
-  ``policy_alert`` map to a terminal ``error`` with an actionable ``detail`` — never
-  a placeholder. The ``None`` queue sentinel is *stream close*, handled by the
-  drain loop, not a wire event.
+- **Fail loudly, never silently.** A fatal top-level ``agent_error`` and a
+  governance ``policy_alert`` map to a terminal ``error`` with an actionable
+  ``detail`` — never a placeholder. A **recoverable** ``agent_error`` (the
+  source event's ``recoverable`` flag, set by ``agent.py``'s
+  ``STATE_ERROR_RECOVERY`` retry path) is explicitly NOT terminal — it folds
+  to a ``status`` line so the run continues (#2515). The ``None`` queue
+  sentinel is *stream close*, handled by the drain loop, not a wire event.
 
 Spec open questions surfaced in this file (do not block #2016):
 - **Q2** — ``policy_alert`` maps to ``error`` (status 403). A governance block is
@@ -240,6 +243,14 @@ def _on_answer(self, event: Dict[str, Any]) -> List[Dict[str, Any]]:
 
     def _on_agent_error(self, event: Dict[str, Any]) -> List[Dict[str, Any]]:
         detail = str(event.get("content") or "Unknown agent error")
+        if event.get("recoverable"):
+            # A per-tool error the agent loop is retrying (agent.py's
+            # STATE_ERROR_RECOVERY path, e.g. a bad tool argument) is not
+            # terminal — the run continues on this same stream. Fold it to a
+            # status line, same pattern as ``_on_tool_confirm_denied``, so
+            # the user still SEES the failure without the stream (and the
+            # still-retrying agent) being cut out from under it (#2515).
+            return [{"type": "status", "message": f"Tool call failed, retrying: {detail}"}]
         return [{"type": "error", "detail": detail, "status": _ERROR_STATUS_AGENT}]
 
     def _on_policy_alert(self, event: Dict[str, Any]) -> List[Dict[str, Any]]:

hub/agents/email/python/tests/test_query_route.py

@@ -15,6 +15,7 @@
 
 import json
 import threading
+import time
 import uuid
 
 import pytest
@@ -189,6 +190,48 @@ def process_query(self, query, max_steps=None):
         raise ValueError("triage produced malformed JSON at row 4")
 
 
+class _RecoverableRetryFakeAgent:
+    """Reproduces #2515: a per-tool error the agent loop is retrying (e.g. the
+    live repro — ``archive_message_batch`` called with a spurious ``mailbox``
+    kwarg), NOT a fatal top-level failure. Pauses right after emitting the
+    recoverable error so the test can inspect ``run.cancel_event`` /
+    ``handler.cancelled`` BEFORE the retry step runs — proving the streaming
+    layer didn't cut the response and cancel the still-retrying agent out
+    from under it.
+    """
+
+    def __init__(self):
+        self.conversation_history = []
+        self.console = None
+        self._cancel_event = None
+        self.error_emitted = threading.Event()
+
+    def process_query(self, query, max_steps=None):
+        self.console.print_processing_start(query, 20, "fake-model")
+        self.console.print_step_header(1, 20)
+        self.console.print_tool_usage("archive_message_batch")
+        self.console.print_error(
+            "Unexpected argument(s) for archive_message_batch: mailbox. "
+            "Accepted argument(s): message_ids.",
+            recoverable=True,
+        )
+        self.error_emitted.set()
+        # Give the streaming layer a beat to process the queued event (and,
+        # pre-fix, cut the stream + cancel this run) before the retry.
+        if self._cancel_event is not None:
+            self._cancel_event.wait(timeout=2)
+            if self._cancel_event.is_set():
+                self.console.print_final_answer("Cancelled.", streaming=False)
+                return {"answer": "Cancelled."}
+        self.console.print_step_header(2, 20)
+        self.console.print_tool_usage("archive_message_batch")
+        self.console.pretty_print_json({"message_ids": ["m1"]}, title="Arguments")
+        self.console.pretty_print_json({"archived": 1})
+        self.console.print_tool_complete()
+        self.console.print_final_answer("Archived 1 message.", streaming=False)
+        return {"answer": "Archived 1 message."}
+
+
 class _InternalErrorFakeAgent:
     """Mimics the base agent's Lemonade-down branch: it sets an actionable
     ``final_answer`` and returns a failed result WITHOUT calling
@@ -347,6 +390,51 @@ def test_cancel_unknown_run_id_is_404(app_client):
     assert resp.status_code == 404
 
 
+# ---------------------------------------------------------------------------
+# #2515 — a recoverable per-tool error must not end the stream or cancel the
+# still-retrying agent
+# ---------------------------------------------------------------------------
+
+
+def test_recoverable_tool_error_does_not_terminate_stream_or_cancel_run(monkeypatch):
+    fake = _RecoverableRetryFakeAgent()
+    monkeypatch.setattr(query_routes, "build_query_agent", lambda **k: fake)
+    client = TestClient(export_openapi.build_app())
+    run_id = str(uuid.uuid4())
+    collected = {}
+
+    def _stream():
+        resp = client.post(
+            "/v1/email/query",
+            json={"query": "archive stuff", "run_id": run_id, "context": []},
+        )
+        collected["text"] = resp.text
+
+    t = threading.Thread(target=_stream, daemon=True)
+    t.start()
+
+    assert fake.error_emitted.wait(timeout=10), "recoverable error never emitted"
+    # Give the async stream generator a moment to drain the queued
+    # ``agent_error`` event through the translator before asserting nothing
+    # tore the run down in response to it.
+    time.sleep(0.3)
+    run = query_routes.registry.get(run_id)
+    assert run is not None, "run ended prematurely — was cancelled before the retry"
+    assert not run.cancel_event.is_set(), "recoverable error set the cancel event"
+    assert not run.handler.cancelled.is_set(), "recoverable error cancelled the handler"
+
+    t.join(timeout=10)
+    events = _parse_sse(collected["text"])
+    types = _types(events)
+    # Both the failed attempt (step 1) AND the retried attempt (step 2) got
+    # their tool_call streamed — proving the loop was not cut off after the
+    # recoverable error and reached completion (#2515).
+    assert types.count("tool_call") == 2
+    assert types.count("error") == 0
+    assert types[-1] == "final"
+    assert events[-1]["answer"] == "Archived 1 message."
+
+
 # ---------------------------------------------------------------------------
 # Error path — a failed run ends with a terminal error event
 # ---------------------------------------------------------------------------

hub/agents/email/python/tests/test_sse_translation.py

@@ -222,6 +222,29 @@ def test_agent_error_maps_to_terminal_error():
     assert out == [{"type": "error", "detail": "boom", "status": 500}]
 
 
+def test_recoverable_agent_error_maps_to_non_terminal_status():
+    """#2515: a per-tool error the agent loop is retrying (agent.py's
+    STATE_ERROR_RECOVERY path) is NOT terminal — the two layers must agree
+    that "recoverable" means the run continues, not that the wire-level
+    terminal contract gets loosened for every agent_error."""
+    out = _tr().translate(
+        {"type": "agent_error", "content": "boom", "recoverable": True}
+    )
+    assert out[0]["type"] not in TERMINAL_TYPES
+    assert out[0]["type"] == "status"
+    # The user must still SEE the failure — never silently swallowed.
+    assert "boom" in out[0]["message"]
+
+
+def test_recoverable_false_agent_error_is_still_terminal():
+    # An explicit False (as well as the field's absence, covered above) keeps
+    # the existing terminal contract — this is not a blanket downgrade.
+    out = _tr().translate(
+        {"type": "agent_error", "content": "boom", "recoverable": False}
+    )
+    assert out == [{"type": "error", "detail": "boom", "status": 500}]
+
+
 def test_policy_alert_maps_to_error_with_tail():
     out = _tr().translate(
         {

src/gaia/agents/base/agent.py

@@ -2200,7 +2200,9 @@ def _execute_tool(self, tool_name: str, tool_args: Dict[str, Any]) -> Any:
             )
             logger.error(error_msg)
             self.error_history.append(error_msg)
-            self.console.print_error(error_msg)
+            # The main loop's is_error/STATE_ERROR_RECOVERY handling retries
+            # this — not a fatal top-level failure (#2515).
+            self.console.print_error(error_msg, recoverable=True)
             return {
                 "status": "error",
                 "error": error_msg,
@@ -2226,8 +2228,10 @@ def _execute_tool(self, tool_name: str, tool_args: Dict[str, Any]) -> Any:
             logger.error(f"Error executing tool {tool_name}: {e}")
             self.error_history.append(str(e))  # Store brief error, not formatted
 
-            # Print to console immediately so user sees it
-            self.console.print_error(formatted_error)
+            # Print to console immediately so user sees it. The caller's
+            # is_error/STATE_ERROR_RECOVERY handling retries this — not a
+            # fatal top-level failure (#2515).
+            self.console.print_error(formatted_error, recoverable=True)
 
             return {
                 "status": "error",
@@ -3116,7 +3120,9 @@ def _process_query_impl(
                         )
                         # Only print if error wasn't already displayed by _execute_tool
                         if not tool_result.get("error_displayed"):
-                            self.console.print_error(last_error)
+                            # STATE_ERROR_RECOVERY below retries this — not a
+                            # fatal top-level failure (#2515).
+                            self.console.print_error(last_error, recoverable=True)
 
                         # Switch to error recovery state
                         self.execution_state = self.STATE_ERROR_RECOVERY
@@ -3887,9 +3893,12 @@ def _process_query_impl(
                         f"Invalid plan format: expected list, got {type(parsed['plan']).__name__}. "
                         f"Plan content: {parsed['plan']}"
                     )
+                    # The "continue" below asks the LLM to correct itself —
+                    # not a fatal top-level failure (#2515).
                     self.console.print_error(
                         f"LLM returned invalid plan format (expected array, got {type(parsed['plan']).__name__}). "
-                        "Asking for correction..."
+                        "Asking for correction...",
+                        recoverable=True,
                     )
 
                     # Create error recovery prompt
@@ -3923,8 +3932,11 @@ def _process_query_impl(
 
                 if invalid_steps:
                     logger.error(f"Invalid plan steps found: {invalid_steps}")
+                    # The "continue" below asks the LLM to correct itself —
+                    # not a fatal top-level failure (#2515).
                     self.console.print_error(
-                        f"Plan contains {len(invalid_steps)} invalid step(s). Asking for correction..."
+                        f"Plan contains {len(invalid_steps)} invalid step(s). Asking for correction...",
+                        recoverable=True,
                     )
 
                     # Create detailed error message
@@ -4141,7 +4153,10 @@ def _process_query_impl(
                             last_error,
                         )
                         if not tool_result.get("error_displayed"):
-                            self.console.print_error(last_error)
+                            # any_error below switches to STATE_ERROR_RECOVERY
+                            # and retries — not a fatal top-level failure
+                            # (#2515, the archive_message_batch repro).
+                            self.console.print_error(last_error, recoverable=True)
                         any_error = True
 
                 if fanout_repeat_break:
@@ -4365,7 +4380,9 @@ def _process_query_impl(
                     )
                     # Only print if error wasn't already displayed by _execute_tool
                     if not tool_result.get("error_displayed"):
-                        self.console.print_error(last_error)
+                        # STATE_ERROR_RECOVERY below retries this — not a
+                        # fatal top-level failure (#2515).
+                        self.console.print_error(last_error, recoverable=True)
 
                     # Switch to error recovery state
                     self.execution_state = self.STATE_ERROR_RECOVERY

src/gaia/agents/base/console.py

@@ -261,8 +261,18 @@ def pretty_print_json(self, data: Dict[str, Any], title: str = None):
     # === Status Messages (Required) ===
 
     @abstractmethod
-    def print_error(self, error_message: str):
-        """Print error message."""
+    def print_error(self, error_message: str, recoverable: bool = False):
+        """Print error message.
+
+        Args:
+            error_message: The error to display.
+            recoverable: True when the agent loop is retrying past this error
+                (e.g. a per-tool argument-validation failure that enters
+                ``STATE_ERROR_RECOVERY``), False for a genuinely fatal
+                top-level failure that ends the run. Handlers that translate
+                this into a downstream wire event (SSE, etc.) use the flag to
+                decide terminal vs. non-terminal — see #2515.
+        """
         ...
 
     @abstractmethod
@@ -1221,12 +1231,14 @@ def print_tool_complete(self) -> None:
         else:
             print("✅ Tool execution complete")
 
-    def print_error(self, error_message: str) -> None:
+    def print_error(self, error_message: str, recoverable: bool = False) -> None:
         """
         Print an error message with appropriate styling.
 
         Args:
             error_message: The error message to display
+            recoverable: Unused by the CLI console (styling is the same
+                either way); accepted for interface parity with SSE handlers.
         """
         # Handle None error messages
         if error_message is None:
@@ -2594,7 +2606,7 @@ def print_tool_complete(self):
     def pretty_print_json(self, data: Dict[str, Any], title: str = None):
         """No-op implementation."""
 
-    def print_error(self, error_message: str):
+    def print_error(self, error_message: str, recoverable: bool = False):
         """No-op implementation."""
 
     def print_warning(self, warning_message: str):

src/gaia/api/sse_handler.py

@@ -144,9 +144,9 @@ def pretty_print_json(self, data: Dict[str, Any], title: str = None):
 
     # === Status Messages (Required) ===
 
-    def print_error(self, error_message: str):
+    def print_error(self, error_message: str, recoverable: bool = False):
         """Print error message."""
-        self._add_event("error", {"message": error_message})
+        self._add_event("error", {"message": error_message, "recoverable": recoverable})
 
     def print_warning(self, warning_message: str):
         """Print warning message."""

src/gaia/ui/sse_handler.py

@@ -443,13 +443,17 @@ def pretty_print_json(self, data: Dict[str, Any], title: str = None):
 
     # === Status Messages ===
 
-    def print_error(self, error_message: str):
-        self._emit(
-            {
-                "type": "agent_error",
-                "content": str(error_message) if error_message else "Unknown error",
-            }
-        )
+    def print_error(self, error_message: str, recoverable: bool = False):
+        event: Dict[str, Any] = {
+            "type": "agent_error",
+            "content": str(error_message) if error_message else "Unknown error",
+        }
+        # Only set when True: keeps the wire shape unchanged for every
+        # existing (fatal) caller, and lets a downstream translator treat a
+        # missing/False flag as the pre-#2515 terminal default.
+        if recoverable:
+            event["recoverable"] = True
+        self._emit(event)
 
     def print_warning(self, warning_message: str):
         self._emit(

tests/unit/agents/test_console_tool_confirmation.py

@@ -126,7 +126,7 @@ def print_tool_complete(self):
     def pretty_print_json(self, data, title=None):
         pass
 
-    def print_error(self, error_message):
+    def print_error(self, error_message, recoverable=False):
         pass
 
     def print_warning(self, warning_message):

tests/unit/chat/ui/test_sse_handler.py

@@ -625,6 +625,25 @@ def test_non_string_error_is_converted(self, handler):
         events = _drain(handler)
         assert "bad value" in events[0]["content"]
 
+    def test_recoverable_flag_defaults_to_omitted(self, handler):
+        # No "recoverable" key at all when the caller doesn't pass it — keeps
+        # the wire shape unchanged for every existing (fatal) caller (#2515).
+        handler.print_error("Something went wrong")
+        events = _drain(handler)
+        assert "recoverable" not in events[0]
+
+    def test_recoverable_true_is_carried_onto_the_event(self, handler):
+        # A per-tool error the agent loop is retrying (agent.py's
+        # STATE_ERROR_RECOVERY path) must be distinguishable from a fatal
+        # top-level failure downstream (#2515).
+        handler.print_error("retryable failure", recoverable=True)
+        events = _drain(handler)
+        assert events[0] == {
+            "type": "agent_error",
+            "content": "retryable failure",
+            "recoverable": True,
+        }
+
 
 # ===========================================================================
 # SSEOutputHandler.print_warning