fix(agent): compose task-bound cron protocol gate with fleet fixes

agent/agent_init.py

@@ -1051,10 +1051,11 @@ def _load_tools(agent, enabled_toolsets, disabled_toolsets):
     )
 
     agent.valid_tool_names = {tool["function"]["name"] for tool in agent.tools} if agent.tools else set()
-    # Kanban guidance is session-static (kanban_show iff HERMES_KANBAN_TASK); resolve once.
-    from agent.prompt_builder import KANBAN_GUIDANCE
-    agent._kanban_worker_guidance = (
-        KANBAN_GUIDANCE if "kanban_show" in agent.valid_tool_names else ""
+    # Resolve task-bound Kanban guidance once so the prompt remains byte-stable.
+    from agent.system_prompt import resolve_kanban_worker_guidance
+
+    agent._kanban_worker_guidance = resolve_kanban_worker_guidance(
+        agent.valid_tool_names
     )
     if agent.quiet_mode:
         return

agent/system_prompt.py

@@ -30,6 +30,26 @@
 from utils import is_truthy_value
 
 logger = logging.getLogger(__name__)
+
+
+def resolve_kanban_worker_guidance(valid_tool_names) -> str:
+    """Return lifecycle guidance only for the task-bound dispatcher worker."""
+    if "kanban_show" not in valid_tool_names:
+        return ""
+    if not os.environ.get("HERMES_KANBAN_TASK"):
+        return ""
+    try:
+        from agent.delegation_context import is_dispatcher_owned_worker_context
+
+        if not is_dispatcher_owned_worker_context():
+            return ""
+    except Exception:
+        # Lifecycle instructions can mutate durable board state; ambiguous
+        # ownership must fail closed rather than borrow an ambient task.
+        return ""
+    return KANBAN_GUIDANCE
+
+
 _PLUGIN_SECTION_FRAME_RE = re.compile(
     r"^## Plugin Context: (?P<id>[a-z0-9][a-z0-9._-]{0,127})\n<!-- hermes-plugin-section-chars:(?P<chars>[0-9]{1,4}) -->\n\n",
     re.MULTILINE,
@@ -281,11 +301,12 @@ def _tool_guidance_block(agent: Any) -> Optional[str]:
             memory_guidance = MEMORY_GUIDANCE
         elif getattr(agent, "_user_profile_enabled", True):
             memory_guidance = USER_PROFILE_GUIDANCE
-    # Kanban lifecycle: resolved once at __init__ (_kanban_worker_guidance);
-    # the kanban_show fallback covers code paths that bypass agent_init.
+    # Kanban lifecycle is task-bound, not merely tool-bound. Configured
+    # orchestrators expose kanban_show without owning a current task.
     _kanban_guidance = getattr(agent, "_kanban_worker_guidance", None)
-    if _kanban_guidance is None and "kanban_show" in names:
-        _kanban_guidance = KANBAN_GUIDANCE
+    if _kanban_guidance is None:
+        # Compatibility fallback for lightweight agents that bypass _load_tools.
+        _kanban_guidance = resolve_kanban_worker_guidance(names)
     tool_guidance = [
         memory_guidance,
         SESSION_SEARCH_GUIDANCE if "session_search" in names else None,

cli.py

@@ -968,8 +968,77 @@ 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
+    from hermes_cli import kanban_db_connect as kbc
+    from hermes_cli import kanban_db_dispatch as kbd
+
+    delays = retry_delays or (0.0,)
+    for attempt, delay in enumerate(delays):
+        if delay:
+            time.sleep(delay)
+        try:
+            with kbc.connect_closing() as conn:
+                disposition = kbd.finalize_clean_worker_exit(conn, task_id, run_id)
+            with kbc.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:
         # Order matters: linger for spawned background work BEFORE any teardown (the
         # parent owns those children's stdout pipes); then the durable flush, since
@@ -984,9 +1053,37 @@ def _finalize_single_query(cli) -> None:
             except Exception:
                 logger.debug("one-shot %s failed", what, 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:

cron/jobs.py

@@ -5,6 +5,7 @@
 import copy
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+import errno
 import json
 import logging
 import shutil
@@ -200,6 +201,11 @@ def _jobs_lock_file() -> Path:
     return _current_cron_store().cron_dir / ".jobs.lock"
 
 
+def _is_fire_fence_contention(exc: OSError) -> bool:
+    """Return whether a nonblocking platform lock reports an active holder."""
+    return exc.errno in {errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK}
+
+
 def _acquire_flock(lock_fd, timeout: float) -> Optional[bool]:
     """Bounded exclusive lock: True when acquired, False on timeout, None when no backend exists. A
     blocking flock(LOCK_EX) taken under the in-process lock would let a wedged sibling freeze
@@ -211,7 +217,9 @@ def _acquire_flock(lock_fd, timeout: float) -> Optional[bool]:
             try:
                 fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                 return True
-            except (OSError, IOError):
+            except (OSError, IOError) as exc:
+                if not _is_fire_fence_contention(exc):
+                    raise
                 if time.monotonic() >= deadline:
                     return False
                 time.sleep(0.1)
@@ -289,7 +297,7 @@ def _jobs_lock():
 
 
 @contextlib.contextmanager
-def _fire_job_lock(job_id: str):
+def _fire_job_lock(job_id: str, *, raise_unavailable: bool = False):
     """Serialize one job's owner mutations and external side effects. Unlike the global jobs lock
     this may be held across network delivery; scoped to one profile + job so unrelated jobs keep
     progressing. Fails closed when cross-process locking is unavailable."""
@@ -316,18 +324,28 @@ def _fire_job_lock(job_id: str):
         lock_path = cron_dir / f".fire-{uuid.uuid5(uuid.NAMESPACE_URL, lock_key).hex}.lock"
         lock_fd = None
         acquired = False
+        unavailable_error = None
         try:
             lock_fd = open(lock_path, "a+", encoding="utf-8")
             lock_fd.seek(0)
             result = _acquire_flock(lock_fd, _JOBS_LOCK_TIMEOUT_SECONDS)
             if result is None:  # pragma: no cover - supported platforms provide one backend
+                unavailable_error = RuntimeError(
+                    "No cross-process lock backend for cron fire fence")
                 logger.error("No cross-process lock backend for cron fire fence")
             elif not result:
                 logger.error("Timed out waiting for fire fence %s; failing closed", lock_path)
             acquired = bool(result)
         except (OSError, IOError) as exc:
+            unavailable_error = exc
             logger.error("Cron fire fence unavailable for %s: %s", job_id, exc)
 
+        if unavailable_error is not None and raise_unavailable:
+            if lock_fd is not None:
+                lock_fd.close()
+                lock_fd = None
+            raise unavailable_error
+
         held_locks[lock_key] = acquired
         try:
             yield acquired
@@ -2229,14 +2247,16 @@ def mark_job_run(
     status: Optional[str] = None,
     *,
     expected_fire_owner: Optional[str] = None,
-) -> bool:
+    provider_backoff: Optional[Dict[str, Any]] = None,
+) -> Optional[bool]:
     """Mark a job as run: update last_run_at/last_status, bump completed, recompute next_run_at,
     and retire the record as a terminal completion when the repeat limit is reached.
 
     ``delivery_error`` is separate from the agent error: agent succeeded but delivery failed records
     ``last_status = "delivery_failed"`` (never "ok") while ``failure_streak`` is left alone. An
-    explicit ``status`` (e.g. "blocked_config") overrides the derived value. False when the fence
-    can't be taken, the job is missing, or ``expected_fire_owner`` no longer holds the fire claim.
+    explicit ``status`` (e.g. "blocked_config") overrides the derived value. Returns None when
+    the fence cannot be taken, False when the job is missing or the owner changed, and True after
+    an authoritative write.
     """
     def apply(jobs, _i, job):
         if expected_fire_owner is not None:
@@ -2248,6 +2268,10 @@ def apply(jobs, _i, job):
                 return False
         now = _hermes_now().isoformat()
         _record_run_outcome(job, success, error, delivery_error, status, now)
+        if success:
+            job.pop("provider_backoff", None)
+        elif provider_backoff is not None:
+            job["provider_backoff"] = copy.deepcopy(provider_backoff)
         _advance_after_run(job, now)
         save_jobs(jobs)
         return True
@@ -2259,7 +2283,10 @@ def locked():
             return False
         return found
 
-    return _under_fire_fence(job_id, locked)
+    with _fire_job_lock(job_id) as acquired:
+        if not acquired:
+            return None
+        return locked()
 
 
 def _write_oneshot_diagnostic(job: Dict[str, Any], text: str, what: str) -> bool:
@@ -2476,7 +2503,8 @@ def _machine_id() -> str:
 
 
 def claim_job_for_fire(
-    job_id: str, *, claim_ttl_seconds: int = 300, force: bool = False, return_job: bool = False,
+    job_id: str, *, claim_ttl_seconds: int = 300, force: bool = False,
+    allow_provider_backoff: bool = False, return_job: bool = False,
 ) -> Union[bool, Dict[str, Any]]:
     """Atomically claim a job for one external 'fire' (multi-machine at-most-once); True iff THIS
     caller won (``CronScheduler.fire_due``: exactly one of N replicas runs a job). Under the
@@ -2494,6 +2522,14 @@ def apply(jobs, _i, job):
         if not force and not is_job_runnable(job):
             return False
         now = _hermes_now()
+        from cron.rate_limit_backoff import provider_backoff_active
+
+        if (
+            not force
+            and not allow_provider_backoff
+            and provider_backoff_active(job, now=now)
+        ):
+            return False
         if _claim_is_live(job.get("fire_claim"), now, claim_ttl_seconds):
             return False  # someone holds a fresh claim
         if force:
@@ -2511,13 +2547,15 @@ def apply(jobs, _i, job):
     return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False))
 
 
-def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> bool:
-    """Refresh an active ``fire_claim`` without extending another owner's lease: an execution may
-    outlive the TTL, and the owner check stops a stale runner from refreshing a recovered claim."""
+def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> Optional[bool]:
+    """Refresh an active fire claim; None means fence contention prevented inspection."""
     def apply(jobs, _i, job):
         return _refresh_claim(jobs, job.get("fire_claim"), expected_owner)
 
-    return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False))
+    with _fire_job_lock(job_id, raise_unavailable=True) as acquired:
+        if not acquired:
+            return None
+        return _with_job(job_id, apply, False)
 
 
 # Completed one-shots are retained in jobs.json (final status stays inspectable) and pruned by
@@ -2973,6 +3011,14 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
             if _has_pause_marker(job):
                 _self_disable_half_paused(job, scan)
                 continue
+            from cron.rate_limit_backoff import provider_backoff_active
+
+            manual_override = bool(
+                job.get("manual_run_at")
+                and job.get("manual_run_at") == job.get("next_run_at")
+            )
+            if not manual_override and provider_backoff_active(job, now=scan.now):
+                continue
             if _evaluate_due_job(job, scan, run_claim_ttl):
                 due.append(job)
         except Exception: