Skip to content

fix(kanban): fail closed on missing worker handoff #102620

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jayleaton wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(kanban): fail closed on missing worker handoff #102620

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
546b132
[verified] fix(kanban): fail closed on missing worker handoff
OutThisLife Sep 4, 2026
de20c16
fix(kanban): exempt durable PRODUCT_SIGNOFF tasks from violation breaker
OutThisLife Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 95 additions & 0 deletions cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1470,8 +1470,75 @@ def _wait_for_oneshot_background_completions(cli) -> None:
)


def _ensure_kanban_worker_lifecycle(
task_id: str,
run_id: int,
*,
clean_exit: bool,
retry_delays: tuple[float, ...] = (0.0, 0.05, 0.2),
) -> bool:
"""Verify a clean Kanban worker exit has a durable terminal run record.

Returns ``True`` when the run was already finalized (including review
handoffs) or this is not a clean exit. A missing handoff is durably
recorded as a protocol violation and returns ``False`` so the process
cannot report nominal success. Transient DB failures are retried; every
successful write is read back through a fresh connection before success is
reported.
"""
if not clean_exit:
return True

from hermes_cli import kanban_db as kb

delays = retry_delays or (0.0,)
for attempt, delay in enumerate(delays):
if delay:
time.sleep(delay)
try:
with kb.connect_closing() as conn:
disposition = kb.finalize_clean_worker_exit(conn, task_id, run_id)
with kb.connect_closing() as verify_conn:
run = kb.get_run(verify_conn, run_id)
durable = bool(
run is not None
and run.task_id == task_id
and run.outcome is not None
and run.ended_at is not None
)
if not durable:
raise RuntimeError(
f"Kanban run {run_id} has no durable terminal lifecycle transition"
)
return disposition != "protocol_violation"
except Exception:
if attempt + 1 == len(delays):
logger.error(
"Kanban worker lifecycle finalization failed for %s run %s",
task_id,
run_id,
exc_info=True,
)
else:
logger.debug(
"retrying Kanban worker lifecycle finalization for %s run %s",
task_id,
run_id,
exc_info=True,
)
return False


def _finalize_single_query(cli) -> None:
"""Close one-shot CLI resources before releasing the active session lease."""
active_exception = sys.exc_info()[1]
clean_exit = active_exception is None or (
isinstance(active_exception, SystemExit)
and active_exception.code in (None, 0)
)
task_id = os.environ.get("HERMES_KANBAN_TASK", "").strip()
raw_run_id = os.environ.get("HERMES_KANBAN_RUN_ID", "").strip()
lifecycle_ok = True
try:
# Linger (bounded) for background processes the turn spawned with
# notify_on_complete=true BEFORE any teardown. The one-shot parent
Expand All @@ -1492,9 +1559,37 @@ def _finalize_single_query(cli) -> None:
except Exception:
logger.debug("one-shot session store flush failed", exc_info=True)
_notify_single_query_session_finalize(cli)
if task_id:
if not raw_run_id:
lifecycle_ok = False
logger.error(
"missing HERMES_KANBAN_RUN_ID for worker %s",
task_id,
)
else:
try:
run_id = int(raw_run_id)
except ValueError:
lifecycle_ok = False
logger.error(
"invalid HERMES_KANBAN_RUN_ID=%r for worker %s",
raw_run_id,
task_id,
)
else:
lifecycle_ok = _ensure_kanban_worker_lifecycle(
task_id,
run_id,
clean_exit=clean_exit,
)
# Lifecycle verification precedes cleanup because cleanup's watchdog may
# terminate a wedged process with os._exit(0). A Kanban worker must have
# durably closed its exact run before that fail-safe can report success.
_run_cleanup(notify_session_finalize=False)
finally:
cli._release_active_session()
if not lifecycle_ok:
raise SystemExit(1)


def _reset_terminal_input_modes_on_exit() -> None:
Expand Down
225 changes: 183 additions & 42 deletions hermes_cli/kanban_db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8380,6 +8380,100 @@ def _defer_reclaim_for_live_worker(
_append_event(conn, task_id, "reclaim_deferred", payload, run_id=run_id)


def finalize_clean_worker_exit(
conn: sqlite3.Connection,
task_id: str,
expected_run_id: int,
) -> str:
"""Fail closed when a worker returns normally without a terminal handoff.

The worker process calls this at its own clean-exit boundary, before the
dispatcher can lose the exit status to PID reaping. A completed, blocked,
review-requested, or otherwise closed run is preserved verbatim. An open
run still owned by this task is atomically closed as a protocol violation
and returned to its source lane for a bounded retry; it is never inferred
as completed, and the breaker is applied only after the existing violation
limit is exhausted.

Returns ``"already_finalized"``, ``"protocol_violation"``, or
``"superseded"``. Replayed calls are idempotent because only the exact
active run can take the transition.
"""
error_text = (
"worker exited cleanly (rc=0) without a durable terminal lifecycle "
"transition — protocol violation. Verify prior work and call "
"kanban_complete, kanban_block, or kanban_request_review."
)
with write_txn(conn):
run = conn.execute(
"SELECT outcome, ended_at, metadata FROM task_runs "
"WHERE id = ? AND task_id = ?",
(int(expected_run_id), task_id),
).fetchone()
if run is None:
return "superseded"
if run["outcome"] is not None or run["ended_at"] is not None:
try:
metadata = json.loads(run["metadata"]) if run["metadata"] else {}
except (TypeError, ValueError):
metadata = {}
if (
run["outcome"] == "crashed"
and isinstance(metadata, dict)
and metadata.get("protocol_violation") is True
and metadata.get("source") == "worker_exit_finalizer"
):
return "protocol_violation"
return "already_finalized"

retry_status = _retry_status_for_run(conn, task_id, int(expected_run_id))
cur = conn.execute(
"UPDATE tasks SET status = ?, claim_lock = NULL, "
"claim_expires = NULL, worker_pid = NULL, last_failure_error = ? "
"WHERE id = ? AND status = 'running' AND current_run_id = ?",
(retry_status, error_text[:500], task_id, int(expected_run_id)),
)
if cur.rowcount != 1:
# Ownership moved between the run read and the CAS. Never mutate a
# successor run or infer completion from task-level status.
return "superseded"
payload = {
"exit_code": 0,
"protocol_violation": True,
"retry_status": retry_status,
"source": "worker_exit_finalizer",
}
run_id = _end_run(
conn,
task_id,
outcome="crashed",
status="crashed",
error=error_text,
metadata=payload,
)
if run_id != int(expected_run_id):
raise RuntimeError(
f"worker finalizer closed run {run_id}, expected {expected_run_id}"
)
_append_event(
conn,
task_id,
"protocol_violation",
payload,
run_id=run_id,
)
_account_protocol_violation(
conn,
task_id,
error_text,
event_payload_extra={
"protocol_violations": _protocol_violation_streak(conn, task_id),
"source": "worker_exit_finalizer",
},
)
return "protocol_violation"


def heartbeat_worker(
conn: sqlite3.Connection,
task_id: str,
Expand Down Expand Up @@ -8803,6 +8897,33 @@ def _error_fingerprint(error_text: str) -> str:
# precedence ``_record_task_failure`` documents for every other failure kind.
_PROTOCOL_VIOLATION_FAILURE_LIMIT = 3

# Durable marker identifying a task that requires explicit product/human
# sign-off (``PRODUCT_SIGNOFF`` in the title or body). Such tasks are exempt
# from the clean-exit violation breaker below: the task body defines the
# worker's job as reporting back for sign-off rather than driving to a
# terminal transition, so repeated clean exits without one are a property of
# the workflow, not a runaway worker — the breaker must never auto-block
# them. The classification is durable (read from the task row at accounting
# time, not from run metadata), so it survives retries, reclaims, and
# dispatcher restarts.
_PRODUCT_SIGNOFF_MARKER = "PRODUCT_SIGNOFF"


def _is_product_signoff_task(
conn: sqlite3.Connection, task_id: str
) -> bool:
"""True when the task durably requires product sign-off."""
row = conn.execute(
"SELECT title, body FROM tasks WHERE id = ?", (task_id,)
).fetchone()
if row is None:
return False
for column in ("title", "body"):
text = row[column]
if text and _PRODUCT_SIGNOFF_MARKER in text:
return True
return False

# How far back to walk a task's closed runs when counting the violation
# streak. The streak trips at a handful of violations, so anything beyond a
# few dozen rows (violations interleaved with neutral rate-limited requeues)
Expand Down Expand Up @@ -8860,6 +8981,55 @@ def _protocol_violation_streak(conn: sqlite3.Connection, task_id: str) -> int:
return streak


def _account_protocol_violation(
conn: sqlite3.Connection,
task_id: str,
error_text: str,
*,
event_payload_extra: Optional[dict] = None,
) -> bool:
"""Apply the bounded clean-exit violation breaker exactly once.

Tasks that durably require product sign-off (``PRODUCT_SIGNOFF`` marker in
title/body, see ``_is_product_signoff_task``) are exempt: the violation is
still recorded and the task retried, but the breaker never auto-blocks
them regardless of how many times the limit is reached.
"""
if _is_product_signoff_task(conn, task_id):
return False
streak = _protocol_violation_streak(conn, task_id)
row = conn.execute(
"SELECT max_retries FROM tasks WHERE id = ?", (task_id,),
).fetchone()
if row is None:
return False
task_override = row["max_retries"]
violation_limit = (
int(task_override)
if task_override is not None
else _PROTOCOL_VIOLATION_FAILURE_LIMIT
)
if streak < violation_limit:
return False
extra = dict(event_payload_extra or {})
extra.update({
"protocol_violations": streak,
"protocol_violation_limit": violation_limit,
})
return _record_task_failure(
conn,
task_id,
error=error_text,
outcome="crashed",
failure_limit=violation_limit,
force_trip=True,
release_claim=False,
end_run=False,
event_payload_extra=extra,
_allow_nested_txn=conn.in_transaction,
)


def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
"""Reclaim ``running`` tasks whose worker PID is no longer alive.

Expand Down Expand Up @@ -9077,47 +9247,11 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
_fp_counts[fp] = _fp_counts.get(fp, 0) + 1
for tid, pid, claimer, protocol_violation, error_text in crash_details:
if protocol_violation:
streak = _protocol_violation_streak(conn, tid)
trow = conn.execute(
"SELECT max_retries FROM tasks WHERE id = ?", (tid,),
).fetchone()
if trow is None:
continue # task deleted mid-loop
task_override = (
trow["max_retries"] if "max_retries" in trow.keys() else None
)
violation_limit = (
int(task_override)
if task_override is not None
else _PROTOCOL_VIOLATION_FAILURE_LIMIT
)
if streak < violation_limit:
# Below budget: the task is already back at ``ready``
# (respawn allowed) with ``last_failure_error`` stamped.
# Deliberately no ``_record_task_failure`` call — a
# below-budget violation must not consume the unified
# failure budget, just as other failure kinds don't
# consume this one.
continue
# Streak reached the bound: trip the breaker. ``force_trip``
# skips the threshold resolution inside
# ``_record_task_failure`` because the decision — including
# the per-task ``max_retries`` override — was already made
# against the violation streak above.
tripped = _record_task_failure(
conn, tid,
error=error_text,
outcome="crashed",
failure_limit=violation_limit,
force_trip=True,
release_claim=False,
end_run=False,
event_payload_extra={
"pid": pid,
"claimer": claimer,
"protocol_violations": streak,
"protocol_violation_limit": violation_limit,
},
tripped = _account_protocol_violation(
conn,
tid,
error_text,
event_payload_extra={"pid": pid, "claimer": claimer},
)
if tripped:
auto_blocked.append(tid)
Expand Down Expand Up @@ -9171,6 +9305,7 @@ def _record_task_failure(
release_claim: bool = False,
end_run: bool = False,
event_payload_extra: Optional[dict] = None,
_allow_nested_txn: bool = False,
) -> bool:
"""Record a non-success outcome (spawn_failed / crashed / timed_out)
and maybe trip the circuit breaker.
Expand Down Expand Up @@ -9218,7 +9353,7 @@ def _record_task_failure(
if failure_limit is None:
failure_limit = DEFAULT_FAILURE_LIMIT
blocked = False
with write_txn(conn):
with write_txn(conn, allow_nested=_allow_nested_txn):
row = conn.execute(
"SELECT consecutive_failures, status, max_retries, current_run_id "
"FROM tasks WHERE id = ?", (task_id,),
Expand All @@ -9244,6 +9379,12 @@ def _record_task_failure(
effective_limit = int(failure_limit)
limit_source = "dispatcher"

# A caller that already exhausted a separate bounded-retry policy is
# tripping this breaker at that policy's threshold. Persist at least
# that threshold so recompute_ready cannot immediately undo the block.
if force_trip:
failures = max(failures, effective_limit)

if force_trip or failures >= effective_limit:
# Trip the breaker.
if release_claim:
Expand Down
Loading