fix(kanban): retry a failed worker turn in place; carry the recovered response

Same-session recovery for kanban workers whose turn dies on a retryable
provider failure: the one-shot CLI used to return rc=0 with no terminal
kanban call, the dispatcher booked a protocol violation and cold-restarted
the card, discarding the whole session context. This retries the turn IN
PLACE (same session, bounded budget, 15/45/90s backoff) and exits honestly
when recovery is exhausted or refused.

Retry authority is positive and typed — never inferred from an outcome
shape: only a failed turn whose classifier marked it retryable; interrupts
are never retried (already persisted/cleared by agent/turn_recovery.py);
the #87096 bounded terminal settlement is never retried; partial /
completed=False are not authority at all. Before EVERY attempt the worker
proves it still owns the exact, unexpired run/claim lease: the dispatcher-
pinned carrier (HERMES_KANBAN_DB + HERMES_KANBAN_RUN_ID +
HERMES_KANBAN_CLAIM_LOCK) is required and compared exactly — no ambient
board resolution — and claim_expires must be unexpired on both the task
row and the open run row, mirroring release_stale_claims()'s stale
predicate and heartbeat_claim()'s extend-then-mirror contract. Fail-closed:
no proof, no retry. The loop never extends its own lease (claim-lifetime
policy remains #95318).

Rebased onto current main (28aceb3451): the one-shot machinery moved out of
cli.py into hermes_cli/cli_single_query.py, and the carrier re-derives
there — the shared exit mapper keeps ONE decision point (the worker
predicate inside it is the STRIPPED kanban_task_id(), at both the
credentials and the turn sites, so a whitespace-only value is not a worker
to the exit mapping either), and both one-shot drivers (`-Q` quiet and `-q`
non-quiet) run the recovery before their goal gates, which continue only on
a settled, authorized turn (post-recovery exit == 0).

Re-review P2 repaired: the non-quiet driver's recovery callback now
CAPTURES the recovered response (nonlocal) and goal continuation receives
the latest settled turn's deliverable — never the stale provider error of
the turn recovery replaced. The quiet driver already derived its response
post-recovery; the non-quiet side now matches. The non-quiet fake mirrors
production (chat() returns the settled result's response) and the positive
control asserts the judge receives "done"; the discarded-capture mutation
reddens that test with the exact stale-error string from the review's
reproduction.

Tests: 63 in tests/agent/test_kanban_turn_recovery.py (retry authority,
carrier/lease proof incl. expiry-during-backoff, goal composition negatives
on both drivers, positive controls, whitespace predicate pin), 133 green
across the 12 focused files incl. main's test_single_query_exit_contract.py.
Mutation receipts: worker predicate un-stripped -> 1 red; non-quiet goal
gate removed -> 1 red; response capture discarded -> 1 red (the review's
repro); quiet goal gate removed -> 8 red; restores byte-identical.

agent/kanban_turn_recovery.py

@@ -0,0 +1,380 @@
+"""In-place recovery for kanban worker turns that die on a retryable failure.
+
+A worker spawned by the kanban dispatcher used to end its run silently when a
+provider call failed: the one-shot CLI returned, the process exited ``rc=0``
+with no terminal ``kanban_complete`` / ``kanban_block`` call, the dispatcher
+booked a *protocol violation* and respawned the task from scratch — discarding
+the whole session context the worker had accumulated. Under a provider stream
+storm that turns one slow morning into several cold restarts per card.
+
+This module retries such a turn IN PLACE: same session, same conversation
+history, bounded budget, explicit backoff, and a continuation nudge that tells
+the model to finish the task with a real terminal call. When the budget is
+exhausted (or the retry is not authorised) the caller exits NON-ZERO so the run
+is booked honestly instead of masquerading as a clean exit.
+
+Retry authority is POSITIVE and TYPED — never inferred from an outcome shape:
+
+* only a ``failed`` turn whose classifier marked the failure ``retryable`` is
+  eligible, and quota/billing walls are left to the dispatcher's
+  cooldown/breaker accounting (``KANBAN_RATE_LIMIT_EXIT_CODE``, see
+  ``hermes_cli/cli_single_query.py::_single_query_exit_code``);
+* ``interrupted=True`` is never retried (the interrupt was persisted and
+  cleared by ``agent/turn_recovery.py::abort_turn_on_interrupt`` — re-entering
+  the model would resurrect explicitly cancelled work);
+* a turn that ended through the bounded terminal settlement is never retried
+  (``agent/turn_finalizer.py::_record_kanban_budget_exhausted`` already recorded
+  ``outcome="timed_out"`` and released the claim — #87096);
+* before EVERY attempt the worker proves it still owns the exact live run and
+  unexpired claim lease (``worker_claim_is_live``). The proof is the carrier the
+  dispatcher pinned at spawn — ``HERMES_KANBAN_DB`` + ``HERMES_KANBAN_RUN_ID`` +
+  ``HERMES_KANBAN_CLAIM_LOCK`` are REQUIRED and compared exactly, and the
+  task/run ``claim_expires`` lease must be unexpired; any missing coordinate or
+  expired lease means no proof, no retry (fail closed).
+
+Incomplete-but-not-failed results (``partial`` / ``completed=False``) are
+deliberately NOT retry authority: truncation and compression repair are owned
+inside the conversation loop (#89289) and by the dispatcher. They still fail the
+process exit code (``hermes_cli/cli_single_query.py::_single_query_exit_code``) so such a run is booked
+honestly instead of ending as a silent ``rc=0``.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sqlite3
+import time
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+from agent.error_classifier import FailoverReason
+
+logger = logging.getLogger(__name__)
+
+#: Recovery attempts when ``HERMES_KANBAN_TURN_RECOVERY`` is unset.
+DEFAULT_MAX_RECOVERY_ATTEMPTS = 3
+
+#: Backoff before recovery attempt N (1-based). The last entry repeats.
+RECOVERY_DELAYS_SECONDS: tuple[float, ...] = (15.0, 45.0, 90.0)
+
+#: Failure reasons that must NOT be retried in place: quota walls and billing
+#: blocks need the dispatcher's cooldown/breaker accounting, not another call.
+#: Aligned with the exit side of main's ``cli._TRANSIENT_PROVIDER_REASONS``
+#: (which also releases ``timeout`` / ``overloaded`` / ``server_error`` runs as
+#: neutral 75s AFTER in-place recovery has been exhausted). The two sets answer
+#: different questions: this one says who may re-enter the model here; main's
+#: says how the process exit is booked.
+_NON_RECOVERABLE_REASONS = frozenset({
+    FailoverReason.rate_limit.value,
+    FailoverReason.billing.value,
+    FailoverReason.upstream_rate_limit.value,
+})
+
+#: ``turn_exit_reason`` prefixes that mean the turn already went through a
+#: durable terminal settlement (the run may be closed and its claim released):
+#: ``agent/turn_finalizer.py`` writes ``max_iterations_reached(N/M)`` right after
+#: recording ``outcome="timed_out"`` for an out-of-budget kanban worker (#87096).
+_TERMINAL_TURN_EXIT_PREFIXES = ("max_iterations_reached",)
+
+_OFF_VALUES = frozenset({"0", "false", "no", "off"})
+
+
+def kanban_task_id() -> Optional[str]:
+    """The dispatcher-set kanban task id, or ``None`` when this is not a worker run.
+
+    Single source of truth for every kanban-worker predicate. The env value is
+    stripped exactly once, here, so a whitespace-only value can never read as
+    "worker" to one caller and "not a worker" to another (the exit-code guard and
+    the recovery gate must agree).
+    """
+    task_id = (os.environ.get("HERMES_KANBAN_TASK") or "").strip()
+    return task_id or None
+
+
+def kanban_turn_recovery_enabled() -> bool:
+    """On when ``HERMES_KANBAN_TASK`` is set and the attempt budget is non-zero."""
+    if kanban_task_id() is None:
+        return False
+    return max_recovery_attempts() > 0
+
+
+def max_recovery_attempts() -> int:
+    """``HERMES_KANBAN_TURN_RECOVERY`` parsed as an attempt count (0 disables).
+
+    Unset/blank -> :data:`DEFAULT_MAX_RECOVERY_ATTEMPTS`; explicit 0/false/no/off
+    -> 0; anything unparseable -> the default. Clamped to [0, 10] so a bad value
+    can never create an unbounded loop.
+    """
+    raw = (os.environ.get("HERMES_KANBAN_TURN_RECOVERY") or "").strip()
+    if not raw:
+        return DEFAULT_MAX_RECOVERY_ATTEMPTS
+    if raw.lower() in _OFF_VALUES:
+        return 0
+    try:
+        value = int(raw)
+    except (TypeError, ValueError):
+        return DEFAULT_MAX_RECOVERY_ATTEMPTS
+    return max(0, min(value, 10))
+
+
+def recovery_delay_seconds(attempt: int) -> float:
+    """Backoff before recovery attempt ``attempt`` (1-based); last entry repeats."""
+    if attempt < 1:
+        attempt = 1
+    index = min(attempt - 1, len(RECOVERY_DELAYS_SECONDS) - 1)
+    return RECOVERY_DELAYS_SECONDS[index]
+
+
+def _turn_exit_is_terminal(result: dict) -> bool:
+    reason = str(result.get("turn_exit_reason") or "")
+    return reason.startswith(_TERMINAL_TURN_EXIT_PREFIXES)
+
+
+def should_recover_turn(result: Any, *, attempt: int) -> bool:
+    """True when a settled worker turn grants in-place retry authority.
+
+    ``attempt`` is the number of recovery attempts ALREADY made. Authority is
+    positive and typed: an enabled recovery within budget, a FAILED turn whose
+    classifier marked the failure ``retryable``, no quota/billing wall, no
+    explicit cancellation, and no durable terminal settlement. Everything else —
+    including ``partial`` / ``completed=False`` — is not authority to re-enter
+    the model. See the module docstring for the full rule set.
+    """
+    if not kanban_turn_recovery_enabled():
+        return False
+    if attempt >= max_recovery_attempts():
+        return False
+    if not isinstance(result, dict):
+        return False
+    if result.get("interrupted") is True:
+        return False
+    if _turn_exit_is_terminal(result):
+        return False
+    if result.get("failed") is not True:
+        return False
+    if result.get("failure_retryable") is not True:
+        return False
+    if str(result.get("failure_reason") or "") in _NON_RECOVERABLE_REASONS:
+        return False
+    return True
+
+
+def _kanban_db_path() -> Optional[str]:
+    """The dispatcher-pinned board DB — no ambient resolution, ever.
+
+    The dispatcher pins ``HERMES_KANBAN_DB`` (plus run id + claim lock) at spawn
+    (``hermes_cli/kanban_db_dispatch.py``); that pin IS the exact authority
+    carrier the retry proof must re-check. If the pin is absent there is no
+    exact carrier to re-prove, so this returns ``None`` and the caller fails
+    closed: resolving whatever board is ambient *now* would silently move the
+    proof onto a different board — precisely the weaker proof this fail-closed
+    contract exists to prevent.
+    """
+    pinned = (os.environ.get("HERMES_KANBAN_DB") or "").strip()
+    return pinned or None
+
+
+def _now() -> int:
+    """Wall-clock seconds — one seam so tests can advance the clock across a backoff."""
+    return int(time.time())
+
+
+def worker_claim_is_live() -> bool:
+    """True when THIS process still owns a live, unexpired run/claim lease.
+
+    Verified read-only against the dispatcher-pinned board before EVERY retry.
+    All three pinned coordinates are REQUIRED — ``HERMES_KANBAN_DB``,
+    ``HERMES_KANBAN_RUN_ID``, ``HERMES_KANBAN_CLAIM_LOCK`` — and compared with
+    exact equality; a missing pin means there is no exact authority carrier to
+    re-prove, never a fallback to ambient board state.
+
+    The lease itself is part of the liveness contract: the canonical stale-claim
+    selector (``hermes_cli/kanban_db.py::release_stale_claims``) treats
+    ``status='running' AND claim_expires < now`` as stale and reclaims the task,
+    and the worker-liveness path treats an expired claim as non-live. So both
+    the task row AND the open run row must carry an UNEXPIRED ``claim_expires``
+    (``heartbeat_claim`` extends the task expiry and mirrors it onto the active
+    run row). An expired lease means mutation authority has lapsed even before
+    the reconciler processes the row; a NULL expiry is not a provable live
+    lease either and fails closed.
+
+    FAIL-CLOSED: any missing coordinate, missing/unreadable DB, mismatched
+    run/lock/pid, closed run, or expired lease returns False and the caller
+    does not re-enter the model — an honest non-zero exit is always available,
+    whereas mutating state under a released or expired claim is not recoverable.
+    """
+    task_id = kanban_task_id()
+    if task_id is None:
+        return False
+    db_path = _kanban_db_path()
+    if not db_path or not Path(db_path).exists():
+        logger.warning("kanban claim check: pinned board db %r not found — not retrying in place", db_path)
+        return False
+    run_id_env = (os.environ.get("HERMES_KANBAN_RUN_ID") or "").strip()
+    lock_env = (os.environ.get("HERMES_KANBAN_CLAIM_LOCK") or "").strip()
+    if not run_id_env or not lock_env:
+        logger.warning(
+            "kanban claim check: missing dispatcher-pinned run-id/claim-lock carrier "
+            "— no exact authority to re-prove, not retrying in place"
+        )
+        return False
+    now = _now()
+    try:
+        conn = sqlite3.connect(f"{Path(db_path).absolute().as_uri()}?mode=ro", uri=True)
+        try:
+            conn.row_factory = sqlite3.Row
+            row = conn.execute(
+                "SELECT status, worker_pid, current_run_id, claim_lock, claim_expires "
+                "FROM tasks WHERE id = ?",
+                (task_id,),
+            ).fetchone()
+            if row is None:
+                return False
+            if (row["status"] or "") != "running":
+                return False
+            worker_pid = row["worker_pid"]
+            if worker_pid is not None and int(worker_pid) != os.getpid():
+                return False
+            # A running task with no run pointer is not proof of a live run: the
+            # dispatcher writes it on claim and clears it on release/end.
+            current_run_id = row["current_run_id"]
+            if current_run_id is None:
+                return False
+            if str(current_run_id) != run_id_env:
+                return False
+            if (row["claim_lock"] or "") != lock_env:
+                return False
+            if row["claim_expires"] is None or int(row["claim_expires"]) < now:
+                return False
+            run = conn.execute(
+                "SELECT ended_at, worker_pid, claim_expires FROM task_runs WHERE id = ?",
+                (int(current_run_id),),
+            ).fetchone()
+            if run is None:
+                return False
+            if run["ended_at"] is not None:
+                return False
+            run_pid = run["worker_pid"]
+            if run_pid is not None and int(run_pid) != os.getpid():
+                return False
+            if run["claim_expires"] is None or int(run["claim_expires"]) < now:
+                return False
+            return True
+        finally:
+            conn.close()
+    except Exception:
+        logger.warning("kanban claim check failed — not retrying in place", exc_info=True)
+        return False
+
+
+def _truncate(text: str, limit: int = 300) -> str:
+    text = (text or "").strip().replace("\n", " ")
+    return text[: limit - 1] + "…" if len(text) > limit else text
+
+
+def build_recovery_nudge(result: Any, *, attempt: int, max_attempts: int) -> str:
+    """The synthetic user turn sent after a failed worker turn.
+
+    Mirrors the stop-nudge contract (agent/kanban_stop.py): plain text is not a
+    terminal state; the worker must finish with ``kanban_complete`` /
+    ``kanban_block``. Emphasises "same session, do not start over" so the model
+    reuses everything it already read/wrote instead of re-doing the work.
+    """
+    error = ""
+    if isinstance(result, dict):
+        error = _truncate(str(result.get("error") or result.get("final_response") or ""))
+    task_id = kanban_task_id() or "this task"
+    return (
+        "[System: the previous turn ended UNFINISHED — the API call failed after "
+        f"all retries and the turn was interrupted (recovery attempt {attempt}/{max_attempts}). "
+        f"Error: {error or 'provider stream failure'}.\n\n"
+        f"Task `{task_id}` is still `running`. This is the SAME session, with your full "
+        "context — nothing you already read, wrote, or computed is lost. Do NOT start over.\n\n"
+        "Do this immediately:\n"
+        "1. If the interrupted turn was mid tool-call, re-check the on-disk state of "
+        "whatever it was writing before continuing (the action may not have executed).\n"
+        "2. Continue the task from where it stopped and finish any remaining deliverable.\n"
+        "3. End with a terminal `kanban_complete(summary=..., artifacts=[...])` if the work "
+        "is done, or `kanban_block(reason=...)` if you are blocked.]"
+    )
+
+
+def _emit_recovery_skipped(emit: Optional[Callable[[str], None]]) -> None:
+    """The one "no proof, no retry" line — fail-closed is worth saying out loud."""
+    message = (
+        f"[kanban] in-place recovery skipped for {kanban_task_id() or 'task'}: this worker no "
+        "longer holds a live run/claim (already settled, reclaimed, or unverifiable)"
+    )
+    logger.warning("%s", message)
+    if emit is not None:
+        try:
+            emit(message)
+        except Exception:
+            logger.debug("kanban turn-recovery emit failed", exc_info=True)
+
+
+def recover_failed_kanban_turns(
+    turn_fn: Callable[[str], Any],
+    get_result: Callable[[], Any],
+    *,
+    sleep_fn: Callable[[float], None] = time.sleep,
+    emit: Optional[Callable[[str], None]] = None,
+    claim_check: Optional[Callable[[], bool]] = None,
+) -> int:
+    """Retry an authorised kanban worker turn in place; returns attempts made.
+
+    ``turn_fn(nudge)`` runs one more turn with the given synthetic user message.
+    ``get_result()`` returns the LATEST settled turn result (``None`` stops the
+    loop). Before EVERY attempt the worker must still own its live run/claim
+    (``claim_check``, default :func:`worker_claim_is_live`); the loop is bounded
+    by the attempt budget even when ``get_result`` never changes. The loop
+    deliberately does NOT extend its own claim lease during the backoff —
+    self-holding a claim is claim-lifetime policy (#95318), not this module's;
+    if the lease expires while sleeping, the re-proof fails and the attempt is
+    not made.
+    """
+    attempts = 0
+    check = claim_check or worker_claim_is_live
+    while True:
+        result = get_result()
+        if not should_recover_turn(result, attempt=attempts):
+            return attempts
+        if not check():
+            _emit_recovery_skipped(emit)
+            return attempts
+        attempts += 1
+        delay = recovery_delay_seconds(attempts)
+        message = (
+            f"[kanban] provider failure on {kanban_task_id() or 'task'} — retrying the turn in place "
+            f"(attempt {attempts}/{max_recovery_attempts()}) after {int(delay)}s; session context preserved"
+        )
+        logger.warning("%s", message)
+        if emit is not None:
+            try:
+                emit(message)
+            except Exception:
+                logger.debug("kanban turn-recovery emit failed", exc_info=True)
+        sleep_fn(delay)
+        # Re-prove ownership AFTER the backoff and immediately before model re-entry: the
+        # dispatcher can reclaim or settle this run while we sleep, and the pre-sleep proof
+        # alone would let that one attempt run without live ownership (round-5 finding F1 —
+        # TOCTOU window). No proof, no retry; the caller's honest exit still applies.
+        if not check():
+            _emit_recovery_skipped(emit)
+            return attempts
+        turn_fn(build_recovery_nudge(result, attempt=attempts, max_attempts=max_recovery_attempts()))
+
+
+__all__ = [
+    "DEFAULT_MAX_RECOVERY_ATTEMPTS",
+    "RECOVERY_DELAYS_SECONDS",
+    "build_recovery_nudge",
+    "kanban_task_id",
+    "kanban_turn_recovery_enabled",
+    "max_recovery_attempts",
+    "recover_failed_kanban_turns",
+    "recovery_delay_seconds",
+    "should_recover_turn",
+    "worker_claim_is_live",
+]

hermes_cli/cli_single_query.py

@@ -159,13 +159,19 @@ def _single_query_exit_code(result, *, credentials_rate_limited: bool = False,
     (EX_CONFIG): the dispatcher blocks the card at once. The same code applies
     before a turn when credential resolution explicitly requires re-authentication;
     unknown startup failures retain exit 1.
+
+    The worker predicate is the STRIPPED ``kanban_task_id()`` (from
+    ``agent.kanban_turn_recovery``) so a whitespace-only value is not a worker here
+    either — the exit mapping and the recovery gate must agree on what a worker is.
     """
     from cli import _TERMINAL_PROVIDER_REASONS, _TRANSIENT_PROVIDER_REASONS
+    from agent.kanban_turn_recovery import kanban_task_id
+
     if not isinstance(result, dict):
-        if credentials_rate_limited and os.environ.get("HERMES_KANBAN_TASK"):
+        if credentials_rate_limited and kanban_task_id():
             from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
             return KANBAN_RATE_LIMIT_EXIT_CODE
-        if credentials_terminal and os.environ.get("HERMES_KANBAN_TASK"):
+        if credentials_terminal and kanban_task_id():
             from hermes_cli.kanban_db import KANBAN_TERMINAL_PROVIDER_EXIT_CODE
             return KANBAN_TERMINAL_PROVIDER_EXIT_CODE
         return 1
@@ -173,7 +179,7 @@ def _single_query_exit_code(result, *, credentials_rate_limited: bool = False,
         return 130
     if not (result.get("failed") or result.get("partial") or result.get("completed") is False):
         return 0
-    if os.environ.get("HERMES_KANBAN_TASK"):
+    if kanban_task_id():
         reason = result.get("failure_reason")
         if reason in _TRANSIENT_PROVIDER_REASONS:
             from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
@@ -220,6 +226,29 @@ def _run_quiet_single_query(cli, effective_query, emitter=None):
         # The exit line below reports session_id to stderr for automation wrappers;
         # without this sync it would point at the ended parent after compression.
         _sync_cli_session_id_from_agent(cli)
+        # Kanban worker: a turn that died on a retryable provider failure is retried
+        # IN PLACE (same session, context preserved) instead of ending the run silently.
+        # Authority (typed retryable failure, no interrupt / terminal settlement) and the
+        # live run/claim proof live in agent/kanban_turn_recovery.py. Runs BEFORE the turn
+        # report so everything downstream — report, follow-ups, response, goal gate, exit —
+        # sees the post-recovery disposition.
+        from agent.kanban_turn_recovery import recover_failed_kanban_turns as _recover_turns
+
+        def _quiet_recover_turn(nudge):
+            nonlocal result
+            _history = result.get("messages") if isinstance(result, dict) else None
+            result = cli.agent.run_conversation(
+                user_message=nudge,
+                conversation_history=_history or cli.conversation_history,
+                **author_kwargs,
+            )
+            _sync_cli_session_id_from_agent(cli)
+
+        _recover_turns(
+            _quiet_recover_turn,
+            lambda: result,
+            emit=lambda msg: print(msg, file=sys.stderr, flush=True),
+        )
         # The turn is over and persisted: the one-shot exit linger that follows protects nested
         # notify_on_complete replies and is NOT part of the spawner's delivery (#113608). The
         # report carries what this run will print, so a spawner booking a child still lingering
@@ -279,9 +308,24 @@ def _run_quiet_single_query(cli, effective_query, emitter=None):
     elif response:
         print(response)
 
+    # ONE exit-code decision point for this driver, shared with the non-quiet one-shot
+    # path via _single_query_exit_code: a quota/billing wall keeps the EX_TEMPFAIL
+    # sentinel so the dispatcher releases the task without counting a failure, and any
+    # other unfinished worker turn exits non-zero instead of ending as a silent rc=0
+    # that reads as a protocol violation. Computed BEFORE goal continuation — the
+    # post-recovery disposition must be authoritative for the rest of the driver.
+    _exit_code = _single_query_exit_code(result)
+
     # Kanban goal_mode: keep working in THIS session until a judge agrees the card is
     # done, the worker terminates it, or the turn budget runs out (sticky block).
-    if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
+    # ONLY a settled, authorized turn may continue: a result that remains
+    # failed/unfinished — recovery exhausted or DENIED (lease expired, claim lost,
+    # quota wall, interrupt, terminal settlement) — must reach the shared exit path
+    # below without goal continuation. goal_run_status() checks only run identity,
+    # not the claim lock or either expiry, so an unreaped expired-lease row still
+    # reports "running" and a continue verdict would re-enter the model under an
+    # authority this process can no longer prove (round-3 finding).
+    if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1" and _exit_code == 0:
         try:
             _run_kanban_goal_loop_q(cli, response)
         except Exception as _goal_exc:
@@ -290,7 +334,6 @@ def _run_quiet_single_query(cli, effective_query, emitter=None):
     if emitter is None:
         print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
 
-    _exit_code = _single_query_exit_code(result)
     if emitter is not None:
         _exit_code = emitter.emit_result(result, session_id=cli.session_id or "", exit_code=_exit_code)
     exit_single_query(_exit_code)
@@ -516,10 +559,33 @@ def _run_single_query_mode(cli, query, image, quiet, oneshot, stream_json: bool
             cli.console.print(f"[bold blue]{t('cli.single_query.query_label')}[/] {_query_label}")
         cli._show_security_advisories()
         response = cli.chat(query, images=single_query_images or None)
+        # Kanban worker: a failed-silently turn used to end the run as rc=0 with no
+        # terminal kanban call, so the dispatcher booked a protocol violation and
+        # cold-restarted the task from scratch. Retry the authorised turn IN PLACE
+        # (see agent/kanban_turn_recovery.py) and CARRY THE RECOVERED RESPONSE: the
+        # goal judge below must evaluate the deliverable of the latest settled turn,
+        # never the stale provider error of the turn recovery replaced (re-review P2).
+        from agent.kanban_turn_recovery import recover_failed_kanban_turns
+
+        def _nonquiet_recover_turn(nudge):
+            nonlocal response
+            response = cli.chat(nudge)
+            return response
+
+        recover_failed_kanban_turns(
+            _nonquiet_recover_turn,
+            lambda: getattr(cli, "_last_turn_result", None),
+            emit=lambda msg: print(msg, file=sys.stderr, flush=True),
+        )
         # Kanban goal_mode on the `-q` path: same judge loop as `-Q`, but each follow-up turn
         # runs through cli.chat so the worker log keeps its live tool feed (the dispatcher
         # used to force -Q here, which left goal_mode cards with a blank Worker log).
-        if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
+        # Gated on the post-recovery disposition exactly like the `-Q` block: a result that
+        # remains failed/unfinished — recovery exhausted or DENIED (claim lost, quota wall,
+        # interrupt, terminal settlement) — must reach the shared exit path below with zero
+        # further model entry, because goal_run_status() checks run identity only and would
+        # otherwise continue under an authority this process can no longer prove.
+        if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1" and _single_query_exit_code(cli._last_turn_result) == 0:
             try:
                 _run_kanban_goal_loop_chat(cli, response or "")
             except Exception as _goal_exc: