fix(kanban): make worker termination and retries fail closed

agent/conversation_loop.py

@@ -38,7 +38,7 @@
 )
 from agent.display import KawaiiSpinner
 from agent.error_classifier import FailoverReason, classify_api_error
-from agent.iteration_budget import IterationBudget
+from agent.iteration_budget import IterationBudget, effective_iteration_state
 from agent.turn_context import (
     _compression_warrants_another_preflight_pass,
     build_turn_context,
@@ -753,6 +753,23 @@ def run_conversation(
     # Commentary deduplication spans all provider continuations and tool calls
     # within one user turn, but must not suppress the same phrase next turn.
     agent._delivered_interim_texts = set()
+    agent._kanban_terminal_tool = None
+    agent._kanban_ownership_lost_reason = None
+    _kanban_worker = bool(os.environ.get("HERMES_KANBAN_TASK"))
+    if not _kanban_worker:
+        os.environ.pop("HERMES_ITERATIONS_REMAINING", None)
+        os.environ.pop("HERMES_MAX_ITERATIONS", None)
+
+    def _sync_kanban_iteration_env() -> None:
+        if not _kanban_worker:
+            return
+        _, effective_remaining, effective_max = effective_iteration_state(
+            agent.iteration_budget,
+            agent.max_iterations,
+            api_call_count,
+        )
+        os.environ["HERMES_ITERATIONS_REMAINING"] = str(effective_remaining)
+        os.environ["HERMES_MAX_ITERATIONS"] = str(effective_max)
 
     # Main conversation loop counters (pure locals consumed by the loop below).
     api_call_count = 0
@@ -812,6 +829,19 @@ def run_conversation(
         )
 
     while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call:
+        _sync_kanban_iteration_env()
+        from tools.kanban_tools import current_worker_ownership_error
+
+        ownership_error = current_worker_ownership_error()
+        if ownership_error:
+            agent._kanban_ownership_lost_reason = ownership_error
+            _turn_exit_reason = "kanban_ownership_lost"
+            final_response = (
+                "This Kanban worker stopped because it no longer owns the "
+                f"active task run: {ownership_error}."
+            )
+            break
+
         _redirect_text = agent._drain_pending_redirect()
         if _redirect_text:
             _apply_active_turn_redirect(agent, messages, _redirect_text)
@@ -843,10 +873,12 @@ def run_conversation(
         if agent._budget_grace_call:
             agent._budget_grace_call = False
         elif not agent.iteration_budget.consume():
+            _sync_kanban_iteration_env()
             _turn_exit_reason = "budget_exhausted"
             if not agent.quiet_mode:
                 agent._safe_print(f"\n⚠️  Iteration budget exhausted ({agent.iteration_budget.used}/{agent.iteration_budget.max_total} iterations used)")
             break
+        _sync_kanban_iteration_env()
 
         # Fire step_callback for gateway hooks (agent:step event)
         if agent.step_callback is not None:
@@ -5316,6 +5348,28 @@ def _perform_api_call(next_api_kwargs):
 
                 agent._execute_tool_calls(assistant_message, messages, effective_task_id, api_call_count)
 
+                terminal_tool = getattr(agent, "_kanban_terminal_tool", None)
+                if terminal_tool:
+                    _turn_exit_reason = f"kanban_terminal({terminal_tool})"
+                    final_response = (
+                        "Kanban task closed successfully via "
+                        f"`{terminal_tool}`."
+                    )
+                    break
+
+                ownership_error = getattr(
+                    agent,
+                    "_kanban_ownership_lost_reason",
+                    None,
+                )
+                if ownership_error:
+                    _turn_exit_reason = "kanban_ownership_lost"
+                    final_response = (
+                        "This Kanban worker stopped because it no longer owns "
+                        f"the active task run: {ownership_error}."
+                    )
+                    break
+
                 if agent._tool_guardrail_halt_decision is not None:
                     decision = agent._tool_guardrail_halt_decision
                     _turn_exit_reason = "guardrail_halt"

agent/iteration_budget.py

@@ -59,4 +59,21 @@ def remaining(self) -> int:
             return max(0, self.max_total - self._used)
 
 
-__all__ = ["IterationBudget"]
+def effective_iteration_state(
+    budget: IterationBudget,
+    max_iterations: int,
+    api_call_count: int,
+) -> tuple[int, int, int]:
+    """Return effective ``(used, remaining, maximum)`` iteration limits."""
+    maximum = max(0, min(int(budget.max_total), int(max_iterations)))
+    remaining = max(
+        0,
+        min(
+            int(budget.remaining),
+            int(max_iterations) - int(api_call_count),
+        ),
+    )
+    return maximum - remaining, remaining, maximum
+
+
+__all__ = ["IterationBudget", "effective_iteration_state"]

agent/kanban_stop.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 from typing import Any, Iterable, Optional
 
@@ -35,34 +36,30 @@ def kanban_stop_nudge_enabled() -> bool:
     return bool(task)
 
 
-def _tool_call_name(tc: Any) -> str:
-    if isinstance(tc, dict):
-        fn = tc.get("function")
-        if isinstance(fn, dict):
-            return str(fn.get("name") or "")
-        return str(tc.get("name") or "")
-    fn = getattr(tc, "function", None)
-    if fn is not None:
-        return str(getattr(fn, "name", "") or "")
-    return str(getattr(tc, "name", "") or "")
+def kanban_terminal_succeeded(tool_name: str, result: Any) -> bool:
+    """Return whether a terminal Kanban tool confirmed its state change."""
+    if tool_name not in _TERMINAL_KANBAN_TOOLS:
+        return False
+    if isinstance(result, str):
+        try:
+            result = json.loads(result)
+        except (TypeError, ValueError):
+            return False
+    return isinstance(result, dict) and result.get("ok") is True
 
 
 def session_called_kanban_terminal(messages: Iterable[dict] | None) -> bool:
-    """True if this conversation already invoked a terminal kanban tool."""
+    """True if this conversation successfully closed its Kanban task."""
     if not messages:
         return False
     for msg in messages:
         if not isinstance(msg, dict):
             continue
-        role = msg.get("role")
-        if role == "assistant":
-            for tc in msg.get("tool_calls") or []:
-                if _tool_call_name(tc) in _TERMINAL_KANBAN_TOOLS:
-                    return True
-        elif role == "tool":
-            name = str(msg.get("name") or "")
-            if name in _TERMINAL_KANBAN_TOOLS:
-                return True
+        if msg.get("role") != "tool":
+            continue
+        name = str(msg.get("name") or "")
+        if kanban_terminal_succeeded(name, msg.get("content")):
+            return True
     return False
 
 
@@ -103,6 +100,7 @@ def build_kanban_stop_nudge(
 
 __all__ = [
     "build_kanban_stop_nudge",
+    "kanban_terminal_succeeded",
     "kanban_stop_nudge_enabled",
     "session_called_kanban_terminal",
 ]

agent/tool_dispatch_helpers.py

@@ -38,9 +38,13 @@
 
 logger = logging.getLogger(__name__)
 
-# Tools that must never run concurrently (interactive / user-facing).
+# Tools that must never run concurrently (interactive or terminal barriers).
 # When any of these appear in a batch, we fall back to sequential execution.
-_NEVER_PARALLEL_TOOLS = frozenset({"clarify"})
+_NEVER_PARALLEL_TOOLS = frozenset({
+    "clarify",
+    "kanban_block",
+    "kanban_complete",
+})
 
 # Read-only tools with no shared mutable session state.
 _PARALLEL_SAFE_TOOLS = frozenset({

agent/tool_executor.py

@@ -209,6 +209,50 @@ def _cancelled_tool_result(reason: str = "user interrupt") -> str:
     )
 
 
+def _kanban_worker_ownership_error() -> Optional[str]:
+    """Return the current worker ownership error, if this is a Kanban worker."""
+    from tools.kanban_tools import current_worker_ownership_error
+
+    return current_worker_ownership_error()
+
+
+def _skip_tool_calls(
+    agent,
+    messages: list,
+    tool_calls,
+    *,
+    reason: str,
+) -> None:
+    """Append one non-effect result for each tool call that was not started."""
+    for tool_call in tool_calls:
+        name = tool_call.function.name
+        messages.append(
+            make_tool_result_message(
+                name,
+                f"[Tool execution skipped — {name} was not started. {reason}]",
+                tool_call.id,
+                effect_disposition="none",
+            )
+        )
+        _flush_session_db_after_tool_progress(
+            agent,
+            messages,
+            stage=f"skipped tool result {name}",
+        )
+
+
+def _terminal_barrier_succeeded(agent, function_name: str, result: Any) -> bool:
+    """Mark a successful terminal tool for this worker's own task."""
+    if not os.environ.get("HERMES_KANBAN_TASK"):
+        return False
+    from agent.kanban_stop import kanban_terminal_succeeded
+
+    if not kanban_terminal_succeeded(function_name, result):
+        return False
+    agent._kanban_terminal_tool = function_name
+    return True
+
+
 def _emit_cancelled_terminal_post_tool_call(
     agent,
     *,
@@ -363,6 +407,17 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
     # avoids rebuilding it per result inside the loop below).
     _tool_budget = _budget_for_agent(agent)
 
+    ownership_error = _kanban_worker_ownership_error()
+    if ownership_error:
+        agent._kanban_ownership_lost_reason = ownership_error
+        _skip_tool_calls(
+            agent,
+            messages,
+            tool_calls,
+            reason=f"Kanban worker ownership was lost: {ownership_error}",
+        )
+        return
+
     # ── Pre-flight: interrupt check ──────────────────────────────────
     if agent._interrupt_requested:
         print(f"{agent.log_prefix}⚡ Interrupt: skipping {num_tools} tool call(s)")
@@ -625,6 +680,23 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
         # submit site below (GHSA-qg5c-hvr5-hjgr, #13617).
         start = time.time()
         try:
+            ownership_error = _kanban_worker_ownership_error()
+            if ownership_error:
+                agent._kanban_ownership_lost_reason = ownership_error
+                result = (
+                    f"[Tool execution skipped — {function_name} was not started. "
+                    f"Kanban worker ownership was lost: {ownership_error}]"
+                )
+                results[index] = (
+                    function_name,
+                    function_args,
+                    result,
+                    0.0,
+                    True,
+                    True,
+                    middleware_trace,
+                )
+                return
             try:
                 result = agent._invoke_tool(
                     function_name,
@@ -659,6 +731,7 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 logger.error("_invoke_tool raised for %s: %s", function_name, tool_error, exc_info=True)
             duration = time.time() - start
             is_error, _ = _detect_tool_failure(function_name, result)
+            _terminal_barrier_succeeded(agent, function_name, result)
             if is_error:
                 logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200])
             else:
@@ -1059,6 +1132,18 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
     # Resolve the context-scaled tool-output budget once per turn.
     _tool_budget = _budget_for_agent(agent)
     for i, tool_call in enumerate(assistant_message.tool_calls, 1):
+        ownership_error = _kanban_worker_ownership_error()
+        if ownership_error:
+            agent._kanban_ownership_lost_reason = ownership_error
+            remaining_calls = assistant_message.tool_calls[i - 1:]
+            _skip_tool_calls(
+                agent,
+                messages,
+                remaining_calls,
+                reason=f"Kanban worker ownership was lost: {ownership_error}",
+            )
+            break
+
         # SAFETY: check interrupt BEFORE starting each tool.
         # If the user sent "stop" during a previous tool's execution,
         # do NOT start any more tools -- skip them all immediately.
@@ -1595,6 +1680,14 @@ def _execute(next_args: dict) -> Any:
         # Log tool errors to the persistent error log so [error] tags
         # in the UI always have a corresponding detailed entry on disk.
         _is_error_result, _ = _detect_tool_failure(function_name, function_result)
+        _terminal_succeeded = (
+            not _execution_blocked
+            and _terminal_barrier_succeeded(
+                agent,
+                function_name,
+                function_result,
+            )
+        )
         # The agent-runtime tools above (todo, session_search, memory,
         # context-engine, memory-manager, clarify, delegate_task) are
         # dispatched inline — they never reach handle_function_call, so the
@@ -1719,6 +1812,19 @@ def _execute(next_args: dict) -> Any:
         # entire batch.  The model sees it on the next API iteration.
         agent._apply_pending_steer_to_tool_results(messages, 1)
 
+        if _terminal_succeeded and i < len(assistant_message.tool_calls):
+            remaining_calls = assistant_message.tool_calls[i:]
+            _skip_tool_calls(
+                agent,
+                messages,
+                remaining_calls,
+                reason=(
+                    f"{function_name} closed the Kanban task; later calls in "
+                    "this response were not started"
+                ),
+            )
+            break
+
         if not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off":
             if agent.verbose_logging:
                 print(f"  ✅ Tool {i} completed in {tool_duration:.2f}s")
@@ -1793,7 +1899,7 @@ def execute_tool_calls_segmented(agent, assistant_message, messages: list, effec
         _exec_cwd = Path(_active_env.cwd) if _active_env is not None and _active_env.cwd else None
         segments = _plan_tool_batch_segments(assistant_message.tool_calls, execution_cwd=_exec_cwd)
 
-    for kind, calls in segments:
+    for segment_index, (kind, calls) in enumerate(segments):
         segment_message = SimpleNamespace(tool_calls=list(calls))
         if kind == "parallel":
             execute_tool_calls_concurrent(
@@ -1805,6 +1911,28 @@ def execute_tool_calls_segmented(agent, assistant_message, messages: list, effec
                 agent, segment_message, messages, effective_task_id, api_call_count,
                 finalize=False,
             )
+        terminal_tool = getattr(agent, "_kanban_terminal_tool", None)
+        ownership_error = getattr(agent, "_kanban_ownership_lost_reason", None)
+        if terminal_tool or ownership_error:
+            later_calls = [
+                tool_call
+                for _kind, segment_calls in segments[segment_index + 1:]
+                for tool_call in segment_calls
+            ]
+            if later_calls:
+                reason = (
+                    f"{terminal_tool} closed the Kanban task; later calls in "
+                    "this response were not started"
+                    if terminal_tool
+                    else f"Kanban worker ownership was lost: {ownership_error}"
+                )
+                _skip_tool_calls(
+                    agent,
+                    messages,
+                    later_calls,
+                    reason=reason,
+                )
+            break
 
     # ── Whole-turn finalize (budget + /steer) ─────────────────────────
     total_tools = len(assistant_message.tool_calls)