review follow-up: violation-only retry streak with defined max_retries precedence
Address the hermes-sweeper review of #61233: the bounded retry budget is now a clean-exit-specific streak, not a share of the unified consecutive_failures counter. - detect_crashed_workers stamps a protocol_violation marker into the violation run's metadata (via the event payload _end_run copies); _protocol_violation_streak derives the streak from run history: consecutive most-recent violation runs, rate_limited runs neutral (mirroring their unified-counter treatment), any other closed run resets it. Mixed failure kinds can neither consume nor extend the budget. - Below-budget violations no longer call _record_task_failure at all: the task returns to ready with last_failure_error stamped directly (including the corrective retry guidance wording adopted from #61817, which build_worker_context surfaces to the retry worker) and the unified counter is untouched, keeping the two budgets independent. - At the bound the trip funnels through _record_task_failure with a new keyword-only force_trip=True: the reaper has already resolved the per-task max_retries override against the violation streak itself, so the threshold comparison is skipped rather than double-applied. max_retries keeps its documented top precedence in both directions: max_retries=1 blocks on the first violation, max_retries=5 blocks on the fifth consecutive one, unset uses the default bound of 3. - Replace the first-violation-blocks regression test with five tests: first occurrence retries (ready + guidance stamped + no gave_up + unified counter untouched); streak trips exactly at the bound with protocol_violations/protocol_violation_limit in the gave_up payload and the auto-blocked side channel set; a prior nonzero crash does not consume the violation budget; a non-violation failure between violations resets the streak; max_retries precedence both directions. All five fail against the previously reviewed diff and pass with this follow-up. The test harness resolves hermes_cli.kanban_db fresh and uses that single module object for the exit registry, liveness patch, and reaper — earlier suite tests reload the module, and the old mixed-object harness made _classify_worker_exit return unknown (the reason the old test failed in full-suite runs on main). Kanban suite: zero introduced failures vs upstream/main tip (62 vs 63 pre-existing environmental failures — the one no longer failing is the old violation test this replaces; 662 passed vs 657). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>fix/verification-admin-route-recovery
slow4cyl
kshitij
parent
c3656e9f0c
commit
452861fdc1
|
|
@ -6571,8 +6571,72 @@ def _error_fingerprint(error_text: str) -> str:
|
|||
# on a later run (a goal-mode finalize nudge, or the model simply emitting the
|
||||
# tool call next time), so a protocol violation is NOT deterministic — give it a
|
||||
# bounded retry before the breaker trips instead of blocking on the first hit.
|
||||
#
|
||||
# The budget is a violation-only STREAK, not a share of the unified
|
||||
# ``consecutive_failures`` counter: it counts consecutive clean-exit protocol
|
||||
# violations (derived from run history by ``_protocol_violation_streak``), so
|
||||
# earlier timeouts / nonzero exits neither consume nor extend it, and a
|
||||
# below-budget violation does not tick the unified counter either. A per-task
|
||||
# ``max_retries`` overrides this bound — the same "task override wins"
|
||||
# precedence ``_record_task_failure`` documents for every other failure kind.
|
||||
_PROTOCOL_VIOLATION_FAILURE_LIMIT = 3
|
||||
|
||||
# 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)
|
||||
# can only mean "way past the bound" anyway.
|
||||
_PROTOCOL_VIOLATION_SCAN_LIMIT = 50
|
||||
|
||||
|
||||
def _protocol_violation_streak(conn: sqlite3.Connection, task_id: str) -> int:
|
||||
"""Count the task's trailing run of clean-exit protocol violations.
|
||||
|
||||
Walks the task's closed runs newest-first — including the violation run
|
||||
``detect_crashed_workers`` just closed — and counts how many in a row were
|
||||
clean-exit protocol violations:
|
||||
|
||||
* ``rate_limited`` runs are neutral and skipped: a quota wall says nothing
|
||||
about the task, exactly as it is neutral for the unified
|
||||
``consecutive_failures`` counter.
|
||||
* Any other closed run (completed, plain crash, timeout, spawn failure,
|
||||
reclaim, …) breaks the streak, so the bounded retry budget counts ONLY
|
||||
protocol violations — mixed failure kinds can neither consume nor
|
||||
extend it.
|
||||
|
||||
Violation runs are recognized by the ``protocol_violation`` marker that
|
||||
``detect_crashed_workers`` stamps into the run metadata; the violation
|
||||
error text is matched as a fallback for runs recorded before the marker
|
||||
existed.
|
||||
"""
|
||||
streak = 0
|
||||
rows = conn.execute(
|
||||
"SELECT outcome, error, metadata FROM task_runs "
|
||||
"WHERE task_id = ? AND ended_at IS NOT NULL "
|
||||
"ORDER BY id DESC LIMIT ?",
|
||||
(task_id, _PROTOCOL_VIOLATION_SCAN_LIMIT),
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
outcome = row["outcome"] or ""
|
||||
if outcome == "rate_limited":
|
||||
continue
|
||||
if outcome == "crashed":
|
||||
is_violation = False
|
||||
raw_meta = row["metadata"]
|
||||
if raw_meta:
|
||||
try:
|
||||
is_violation = bool(
|
||||
json.loads(raw_meta).get("protocol_violation")
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
is_violation = False
|
||||
if not is_violation:
|
||||
is_violation = "protocol violation" in (row["error"] or "")
|
||||
if is_violation:
|
||||
streak += 1
|
||||
continue
|
||||
break
|
||||
return streak
|
||||
|
||||
|
||||
def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
|
||||
"""Reclaim ``running`` tasks whose worker PID is no longer alive.
|
||||
|
|
@ -6607,8 +6671,9 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
|
|||
# Per-crash details collected inside the main txn, used after it
|
||||
# closes to run ``_record_task_failure`` (which needs its own
|
||||
# write_txn so can't nest). ``protocol_violation`` flags the
|
||||
# clean-exit-but-still-running case so we can trip the breaker
|
||||
# immediately instead of incrementing by 1.
|
||||
# clean-exit-but-still-running case, which is accounted against its
|
||||
# own bounded violation streak instead of the unified failure
|
||||
# counter (see the post-txn loop below).
|
||||
crash_details: list[tuple[str, int, str, bool, str]] = []
|
||||
# (task_id, pid, claimer, protocol_violation, error_text)
|
||||
with write_txn(conn):
|
||||
|
|
@ -6639,18 +6704,29 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
|
|||
if kind == "clean_exit":
|
||||
# Worker subprocess returned 0 but its task is still
|
||||
# ``running`` in the DB — it exited without calling
|
||||
# ``kanban_complete`` / ``kanban_block``. Retrying won't
|
||||
# help.
|
||||
# ``kanban_complete`` / ``kanban_block``. Overwhelmingly the
|
||||
# work itself succeeded and only the paperwork was skipped, so
|
||||
# a retry usually completes; the corrective sentence below is
|
||||
# surfaced to the retry worker via the prior-attempt error in
|
||||
# ``build_worker_context`` (guidance approach from #61817).
|
||||
protocol_violation = True
|
||||
error_text = (
|
||||
"worker exited cleanly (rc=0) without calling "
|
||||
"kanban_complete or kanban_block — protocol violation"
|
||||
"kanban_complete or kanban_block — protocol violation. "
|
||||
"If the prior run already did the work, verify it and "
|
||||
"report the result via kanban_complete; a run that ends "
|
||||
"without a terminal kanban call counts as failed no "
|
||||
"matter what it did."
|
||||
)
|
||||
event_kind = "protocol_violation"
|
||||
event_payload = {
|
||||
"pid": pid,
|
||||
"claimer": row["claim_lock"],
|
||||
"exit_code": code,
|
||||
# Durable marker for _protocol_violation_streak: _end_run
|
||||
# copies this payload into the run metadata, which is how
|
||||
# the violation-only retry budget is derived later.
|
||||
"protocol_violation": True,
|
||||
}
|
||||
elif kind == "rate_limited":
|
||||
# Worker bailed because the provider rate-limited / exhausted
|
||||
|
|
@ -6721,22 +6797,39 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
|
|||
)
|
||||
rate_limited.append(row["id"])
|
||||
else:
|
||||
if protocol_violation:
|
||||
# Stamp the failure error now: a below-budget
|
||||
# violation never reaches ``_record_task_failure``
|
||||
# (which stamps this column for every other failure
|
||||
# kind), yet the board UI and the retry worker's
|
||||
# context still need the violation message + the
|
||||
# corrective guidance it carries.
|
||||
conn.execute(
|
||||
"UPDATE tasks SET last_failure_error = ? "
|
||||
"WHERE id = ?",
|
||||
(error_text[:500], row["id"]),
|
||||
)
|
||||
crashed.append(row["id"])
|
||||
crash_details.append(
|
||||
(row["id"], pid, row["claim_lock"],
|
||||
protocol_violation, error_text)
|
||||
)
|
||||
# Outside the main txn: increment the unified failure counter for
|
||||
# each crashed task. If the breaker trips, the task transitions
|
||||
# ready → blocked with a ``gave_up`` event on top of the ``crashed``
|
||||
# event we already emitted.
|
||||
# Outside the main txn: account each crashed task and maybe trip the
|
||||
# breaker (the task transitions ready → blocked with a ``gave_up`` event
|
||||
# on top of the event we already emitted).
|
||||
#
|
||||
# Protocol-violation crashes (clean exit, no terminal tool call) get a
|
||||
# BOUNDED retry, not an immediate trip: empirically ~96% of these tasks
|
||||
# complete on a later run (a goal-mode finalize nudge, or the model simply
|
||||
# emitting kanban_complete/kanban_block next time), so blocking on the first
|
||||
# occurrence just churned them through the respawn cycle. Systemic
|
||||
# same-error crashes still trip immediately.
|
||||
# occurrence just churned them through the respawn cycle. The retry budget
|
||||
# is a violation-only streak (``_protocol_violation_streak``): earlier
|
||||
# timeouts / nonzero exits neither consume nor extend it, and a
|
||||
# below-budget violation does not tick the unified
|
||||
# ``consecutive_failures`` counter, so the two budgets stay independent.
|
||||
# A per-task ``max_retries`` overrides the violation bound with the same
|
||||
# top precedence it has for every other failure kind. Systemic same-error
|
||||
# crashes still trip immediately.
|
||||
auto_blocked: list[str] = []
|
||||
if crash_details:
|
||||
# Fingerprint errors to detect systemic failures.
|
||||
|
|
@ -6745,25 +6838,59 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
|
|||
fp = _error_fingerprint(err_text)
|
||||
_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,
|
||||
},
|
||||
)
|
||||
if tripped:
|
||||
auto_blocked.append(tid)
|
||||
continue
|
||||
fp = _error_fingerprint(error_text)
|
||||
is_systemic = (
|
||||
not protocol_violation
|
||||
and _fp_counts.get(fp, 0) >= 3
|
||||
)
|
||||
# Systemic same-error crashes trip on the first hit; a protocol
|
||||
# violation gets a bounded retry (it usually recovers); everything
|
||||
# else uses the task's normal failure budget.
|
||||
if is_systemic:
|
||||
_flim = 1
|
||||
elif protocol_violation:
|
||||
_flim = _PROTOCOL_VIOLATION_FAILURE_LIMIT
|
||||
else:
|
||||
_flim = None
|
||||
is_systemic = _fp_counts.get(fp, 0) >= 3
|
||||
tripped = _record_task_failure(
|
||||
conn, tid,
|
||||
error=error_text,
|
||||
outcome="crashed",
|
||||
failure_limit=_flim,
|
||||
failure_limit=1 if is_systemic else None,
|
||||
release_claim=False,
|
||||
end_run=False,
|
||||
event_payload_extra={"pid": pid, "claimer": claimer},
|
||||
|
|
@ -6788,6 +6915,7 @@ def _record_task_failure(
|
|||
*,
|
||||
outcome: str,
|
||||
failure_limit: int = None,
|
||||
force_trip: bool = False,
|
||||
release_claim: bool = False,
|
||||
end_run: bool = False,
|
||||
event_payload_extra: Optional[dict] = None,
|
||||
|
|
@ -6825,6 +6953,15 @@ def _record_task_failure(
|
|||
2. caller-supplied ``failure_limit`` (gateway passes the config
|
||||
value from ``kanban.failure_limit``; tests pass fixed values)
|
||||
3. ``DEFAULT_FAILURE_LIMIT``
|
||||
|
||||
``force_trip=True`` trips the breaker unconditionally, skipping the
|
||||
counter-vs-threshold comparison (the resolution order above is then
|
||||
only reported in the ``gave_up`` payload, not re-evaluated). Callers
|
||||
use it when they have already applied their own bounded-retry policy
|
||||
— e.g. the clean-exit protocol-violation streak in
|
||||
``detect_crashed_workers``, which resolves the per-task
|
||||
``max_retries`` override against the violation streak itself. The
|
||||
failure is still counted into ``consecutive_failures``.
|
||||
"""
|
||||
if failure_limit is None:
|
||||
failure_limit = DEFAULT_FAILURE_LIMIT
|
||||
|
|
@ -6851,7 +6988,7 @@ def _record_task_failure(
|
|||
effective_limit = int(failure_limit)
|
||||
limit_source = "dispatcher"
|
||||
|
||||
if failures >= effective_limit:
|
||||
if force_trip or failures >= effective_limit:
|
||||
# Trip the breaker.
|
||||
if release_claim:
|
||||
# Spawn path: still running, also clear claim state.
|
||||
|
|
|
|||
|
|
@ -4398,43 +4398,71 @@ def test_detect_crashed_workers_increments_counter(kanban_home):
|
|||
conn.close()
|
||||
|
||||
|
||||
def test_detect_crashed_workers_protocol_violation_auto_blocks(kanban_home):
|
||||
"""A worker that exited rc=0 while its task was still ``running``
|
||||
is a protocol violation (agent answered conversationally without
|
||||
calling kanban_complete / kanban_block). Retrying will just loop,
|
||||
so auto-block immediately instead of waiting for the breaker to
|
||||
trip at ``DEFAULT_FAILURE_LIMIT``.
|
||||
def _drive_worker_exit(conn, tid, fake_pid, raw_status):
|
||||
"""Claim ``tid``, record ``raw_status`` for its dead worker pid, and run
|
||||
one reaper pass.
|
||||
|
||||
Regression test for the respawn-loop-after-completion bug reported
|
||||
against small local models (gemma4-e2b q4) where the model writes
|
||||
the answer as plain text and the CLI exits rc=0 cleanly.
|
||||
Deliberately resolves ``hermes_cli.kanban_db`` fresh and uses that single
|
||||
module object for the exit registry, the liveness patch, AND the reaper:
|
||||
earlier tests in a full-suite run can reload the module, and recording
|
||||
the exit into one module object while reaping through another (stale)
|
||||
one makes ``_classify_worker_exit`` return ``unknown`` — silently turning
|
||||
a clean-exit protocol violation into a plain crash.
|
||||
"""
|
||||
import hermes_cli.kanban_db as _kb
|
||||
host_prefix = _kb._claimer_id().split(":", 1)[0]
|
||||
claimed = _kb.claim_task(conn, tid, claimer=f"{host_prefix}:mock")
|
||||
assert claimed is not None, "task was not claimable for the next attempt"
|
||||
_kb._set_worker_pid(conn, tid, fake_pid)
|
||||
_kb._record_worker_exit(fake_pid, raw_status)
|
||||
original_alive = _kb._pid_alive
|
||||
_kb._pid_alive = lambda p: False
|
||||
try:
|
||||
return _kb.detect_crashed_workers(conn)
|
||||
finally:
|
||||
_kb._pid_alive = original_alive
|
||||
|
||||
|
||||
def _drive_protocol_violation(conn, tid, fake_pid):
|
||||
"""One clean-exit protocol violation reaper pass for ``tid``.
|
||||
|
||||
os.W_EXITCODE(status=0, signal=0) == 0 on POSIX.
|
||||
"""
|
||||
return _drive_worker_exit(conn, tid, fake_pid, 0)
|
||||
|
||||
|
||||
def _drive_nonzero_crash(conn, tid, fake_pid):
|
||||
"""One plain non-zero-exit crash reaper pass for ``tid``.
|
||||
|
||||
W_EXITCODE(1, 0) == 256 — WIFEXITED True, WEXITSTATUS == 1.
|
||||
"""
|
||||
return _drive_worker_exit(conn, tid, fake_pid, 256)
|
||||
|
||||
|
||||
def test_detect_crashed_workers_protocol_violation_first_occurrence_retries(kanban_home):
|
||||
"""A first clean-exit protocol violation gets a retry, not a block.
|
||||
|
||||
A worker that exited rc=0 while its task was still ``running`` skipped
|
||||
the terminal kanban call (model answered conversationally, transient tool
|
||||
wedge). Empirically these overwhelmingly complete on respawn, so the
|
||||
first violation must leave the task ``ready`` with corrective guidance
|
||||
stamped in ``last_failure_error`` — not trip the breaker like the pre-fix
|
||||
behavior did. The violation is accounted against its own violation-only
|
||||
streak, so it must NOT tick the unified ``consecutive_failures`` counter.
|
||||
"""
|
||||
conn = kb.connect()
|
||||
try:
|
||||
tid = kb.create_task(conn, title="quiet", assignee="worker")
|
||||
host_prefix = _kb._claimer_id().split(":", 1)[0]
|
||||
lock = f"{host_prefix}:mock"
|
||||
kb.claim_task(conn, tid, claimer=lock)
|
||||
fake_pid = 999998
|
||||
kb._set_worker_pid(conn, tid, fake_pid)
|
||||
|
||||
# Simulate the reap loop having recorded a clean exit for this pid.
|
||||
# os.W_EXITCODE(status=0, signal=0) == 0 on POSIX.
|
||||
_kb._record_worker_exit(fake_pid, 0)
|
||||
# Force liveness check to say "dead" for the fake pid.
|
||||
original_alive = _kb._pid_alive
|
||||
_kb._pid_alive = lambda p: False
|
||||
try:
|
||||
result_crashed = kb.detect_crashed_workers(conn)
|
||||
finally:
|
||||
_kb._pid_alive = original_alive
|
||||
|
||||
result_crashed = _drive_protocol_violation(conn, tid, 999998)
|
||||
assert tid in result_crashed, "should be detected as crashed"
|
||||
|
||||
task = kb.get_task(conn, tid)
|
||||
assert task.status == "blocked", (
|
||||
f"protocol violation should auto-block on first occurrence, "
|
||||
f"got status={task.status}"
|
||||
assert task.status == "ready", (
|
||||
f"first protocol violation should retry, got status={task.status}"
|
||||
)
|
||||
assert task.consecutive_failures == 0, (
|
||||
"a below-budget violation must not consume the unified failure "
|
||||
f"budget, got consecutive_failures={task.consecutive_failures}"
|
||||
)
|
||||
assert "kanban_complete" in (task.last_failure_error or ""), (
|
||||
f"expected protocol-violation message, got {task.last_failure_error!r}"
|
||||
|
|
@ -4450,13 +4478,175 @@ def test_detect_crashed_workers_protocol_violation_auto_blocks(kanban_home):
|
|||
assert "crashed" not in kinds, (
|
||||
f"should NOT emit 'crashed' event on clean exit, got {kinds}"
|
||||
)
|
||||
assert "gave_up" in kinds, (
|
||||
f"breaker should trip, expected 'gave_up' event, got {kinds}"
|
||||
assert "gave_up" not in kinds, (
|
||||
f"breaker must not trip on the first violation, got {kinds}"
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_detect_crashed_workers_protocol_violation_streak_trips_at_limit(kanban_home):
|
||||
"""The violation streak trips the terminal path exactly at the bound.
|
||||
|
||||
Genuine repeat offenders (a worker whose CLI keeps returning 0 without a
|
||||
terminal transition) must still surface to a human: the
|
||||
``_PROTOCOL_VIOLATION_FAILURE_LIMIT``-th consecutive violation blocks the
|
||||
task with a ``gave_up`` event carrying the streak accounting.
|
||||
"""
|
||||
import hermes_cli.kanban_db as _kb
|
||||
conn = kb.connect()
|
||||
try:
|
||||
tid = kb.create_task(conn, title="quiet", assignee="worker")
|
||||
limit = _kb._PROTOCOL_VIOLATION_FAILURE_LIMIT
|
||||
for i in range(limit - 1):
|
||||
_drive_protocol_violation(conn, tid, 990000 + i)
|
||||
assert kb.get_task(conn, tid).status == "ready", (
|
||||
f"violation {i + 1}/{limit} should still retry"
|
||||
)
|
||||
|
||||
_drive_protocol_violation(conn, tid, 990900)
|
||||
|
||||
task = kb.get_task(conn, tid)
|
||||
assert task.status == "blocked", (
|
||||
f"violation streak at the bound must block, got {task.status}"
|
||||
)
|
||||
events = kb.list_events(conn, tid)
|
||||
kinds = [e.kind for e in events]
|
||||
assert kinds.count("protocol_violation") == limit
|
||||
assert "crashed" not in kinds
|
||||
gave_up = [e for e in events if e.kind == "gave_up"]
|
||||
assert len(gave_up) == 1, f"expected exactly one gave_up, got {kinds}"
|
||||
payload = gave_up[0].payload or {}
|
||||
assert payload.get("protocol_violations") == limit
|
||||
assert payload.get("protocol_violation_limit") == limit
|
||||
# Side channel consumed by dispatch_once — read through the same
|
||||
# (current) module object the reaper ran in, see _drive_worker_exit.
|
||||
assert tid in _kb.detect_crashed_workers._last_auto_blocked
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_protocol_violation_budget_not_consumed_by_other_failures(kanban_home):
|
||||
"""Mixed failure kinds must not consume the violation retry budget.
|
||||
|
||||
Regression for the #61233 review finding: expressed as a plain
|
||||
``failure_limit`` over the unified ``consecutive_failures`` counter, the
|
||||
violation budget was consumed by earlier timeouts / nonzero exits. As a
|
||||
violation-only streak, a prior real crash must not eat violation
|
||||
retries, and below-budget violations must leave the unified counter
|
||||
untouched (so the two budgets stay independent).
|
||||
"""
|
||||
import hermes_cli.kanban_db as _kb
|
||||
conn = kb.connect()
|
||||
try:
|
||||
tid = kb.create_task(conn, title="mixed", assignee="worker")
|
||||
|
||||
# One real crash: unified counter ticks to 1 (below
|
||||
# DEFAULT_FAILURE_LIMIT=2 — task stays ready).
|
||||
_drive_nonzero_crash(conn, tid, 991000)
|
||||
task = kb.get_task(conn, tid)
|
||||
assert task.status == "ready"
|
||||
assert task.consecutive_failures == 1
|
||||
|
||||
# Two violations after it: streak 1 and 2 — both retry, unified
|
||||
# counter untouched. (Pre-fix: the crash consumed the budget and the
|
||||
# violations blocked well before three of them happened.)
|
||||
for i, pid in enumerate((991001, 991002)):
|
||||
_drive_protocol_violation(conn, tid, pid)
|
||||
task = kb.get_task(conn, tid)
|
||||
assert task.status == "ready", (
|
||||
f"violation {i + 1} after a crash must still retry, "
|
||||
f"got {task.status}"
|
||||
)
|
||||
assert task.consecutive_failures == 1, (
|
||||
"below-budget violations must not tick the unified counter"
|
||||
)
|
||||
|
||||
# Third consecutive violation: streak hits the bound — blocked.
|
||||
_drive_protocol_violation(conn, tid, 991003)
|
||||
task = kb.get_task(conn, tid)
|
||||
assert task.status == "blocked"
|
||||
gave_up = [e for e in kb.list_events(conn, tid) if e.kind == "gave_up"]
|
||||
assert len(gave_up) == 1
|
||||
assert (gave_up[0].payload or {}).get("protocol_violations") == \
|
||||
_kb._PROTOCOL_VIOLATION_FAILURE_LIMIT
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_protocol_violation_streak_resets_on_other_failure_kind(kanban_home):
|
||||
"""A non-violation failure between violations resets the streak.
|
||||
|
||||
The budget counts CONSECUTIVE clean-exit violations: two violations, a
|
||||
real crash, then two more violations is a streak of 2 — not 4 — so the
|
||||
fourth violation must still retry; only a third consecutive one blocks.
|
||||
"""
|
||||
conn = kb.connect()
|
||||
try:
|
||||
tid = kb.create_task(conn, title="reset", assignee="worker")
|
||||
|
||||
_drive_protocol_violation(conn, tid, 993000)
|
||||
_drive_protocol_violation(conn, tid, 993001)
|
||||
assert kb.get_task(conn, tid).status == "ready"
|
||||
|
||||
# Real crash breaks the streak (and ticks the unified counter to 1).
|
||||
_drive_nonzero_crash(conn, tid, 993002)
|
||||
assert kb.get_task(conn, tid).status == "ready"
|
||||
|
||||
# Streak restarts at 1, 2 — the pre-crash violations no longer count.
|
||||
_drive_protocol_violation(conn, tid, 993003)
|
||||
assert kb.get_task(conn, tid).status == "ready", (
|
||||
"violation streak must reset after a non-violation failure"
|
||||
)
|
||||
_drive_protocol_violation(conn, tid, 993004)
|
||||
assert kb.get_task(conn, tid).status == "ready"
|
||||
|
||||
# Third consecutive violation since the crash: blocked.
|
||||
_drive_protocol_violation(conn, tid, 993005)
|
||||
assert kb.get_task(conn, tid).status == "blocked"
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_protocol_violation_respects_max_retries_precedence(kanban_home):
|
||||
"""Per-task ``max_retries`` overrides the violation bound, both ways.
|
||||
|
||||
Same top precedence it has for every other failure kind in
|
||||
``_record_task_failure``: ``max_retries=1`` blocks on the FIRST violation
|
||||
(zero retries — the pre-fix behavior, now opt-in per task);
|
||||
``max_retries=5`` keeps retrying past the default bound of 3 and blocks
|
||||
on the 5th consecutive violation.
|
||||
"""
|
||||
conn = kb.connect()
|
||||
try:
|
||||
strict = kb.create_task(
|
||||
conn, title="strict", assignee="worker", max_retries=1,
|
||||
)
|
||||
_drive_protocol_violation(conn, strict, 992000)
|
||||
task = kb.get_task(conn, strict)
|
||||
assert task.status == "blocked", (
|
||||
f"max_retries=1 must block on the first violation, got {task.status}"
|
||||
)
|
||||
gave_up = [e for e in kb.list_events(conn, strict) if e.kind == "gave_up"]
|
||||
assert len(gave_up) == 1
|
||||
payload = gave_up[0].payload or {}
|
||||
assert payload.get("protocol_violations") == 1
|
||||
assert payload.get("protocol_violation_limit") == 1
|
||||
|
||||
lenient = kb.create_task(
|
||||
conn, title="lenient", assignee="worker", max_retries=5,
|
||||
)
|
||||
for i in range(4):
|
||||
_drive_protocol_violation(conn, lenient, 992100 + i)
|
||||
assert kb.get_task(conn, lenient).status == "ready", (
|
||||
f"violation {i + 1}/5 should retry under max_retries=5"
|
||||
)
|
||||
_drive_protocol_violation(conn, lenient, 992104)
|
||||
assert kb.get_task(conn, lenient).status == "blocked"
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_detect_crashed_workers_nonzero_exit_uses_default_limit(kanban_home):
|
||||
"""A worker that exited non-zero (real error / crash) uses the
|
||||
normal counter path — one failure doesn't trip the breaker.
|
||||
|
|
|
|||
Loading…
Reference in New Issue