From df1e021b3b38f93278d243d4aadd3872c4df0600 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Thu, 3 Sep 2026 07:59:37 +0700 Subject: [PATCH 1/5] [verified] fix(cron): preserve active runs across gateway restart --- cron/delivery_queue.py | 244 +++++++++++++++ cron/executions.py | 27 ++ cron/scheduler.py | 277 +++++++++++++++++- gateway/run.py | 39 ++- hermes_cli/kanban_db.py | 40 ++- tests/cron/test_delivery_queue.py | 94 ++++++ tests/cron/test_execution_ledger.py | 2 +- tests/cron/test_parallel_pool.py | 2 +- tests/cron/test_restart_safe_worker.py | 245 ++++++++++++++++ tests/cron/test_run_one_job.py | 10 +- tests/cron/test_script_claim_heartbeat.py | 2 +- .../test_cron_delivery_housekeeping.py | 82 ++++++ .../test_kanban_gateway_restart_handoff.py | 178 +++++++++++ tools/process_registry.py | 29 ++ 14 files changed, 1260 insertions(+), 11 deletions(-) create mode 100644 cron/delivery_queue.py create mode 100644 tests/cron/test_delivery_queue.py create mode 100644 tests/cron/test_restart_safe_worker.py create mode 100644 tests/gateway/test_cron_delivery_housekeeping.py create mode 100644 tests/hermes_cli/test_kanban_gateway_restart_handoff.py diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py new file mode 100644 index 000000000000..a7dbc8eea283 --- /dev/null +++ b/cron/delivery_queue.py @@ -0,0 +1,244 @@ +"""Profile-local durable handoff for cron delivery through live gateway adapters. + +A restart-safe cron worker executes outside the gateway cgroup. It cannot own +relay/E2EE adapter objects, so it queues the final send here. A gateway claims +each row at most once. If that gateway dies after claiming, the outcome is +marked unknown and never retried: losing a delivery is safer than duplicating a +possibly-completed send. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import threading +import time +import uuid +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Callable, Iterator, Optional + +from hermes_constants import get_hermes_home +from hermes_time import now as _hermes_now + +DELIVERY_DB: Optional[Path] = None +_PROCESS_ID = uuid.uuid4().hex +_lock = threading.RLock() +_ACTIVE_DELIVERIES: set[str] = set() +_TERMINAL = ("delivered", "failed", "unknown") + + +def _path() -> Path: + return DELIVERY_DB or (get_hermes_home().resolve() / "cron" / "deliveries.db") + + +@contextmanager +def _transaction() -> Iterator[sqlite3.Connection]: + with _lock: + path = _path() + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path, timeout=5) + try: + path.chmod(0o600) + except OSError: + pass + conn.row_factory = sqlite3.Row + try: + conn.execute("PRAGMA busy_timeout=5000") + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=FULL") + conn.execute( + """CREATE TABLE IF NOT EXISTS deliveries ( + execution_id TEXT PRIMARY KEY, + job_json TEXT NOT NULL, + content TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN + ('pending','delivering','delivered','failed','unknown')), + owner_process_id TEXT, + owner_pid INTEGER, + owner_started_at INTEGER, + created_at TEXT NOT NULL, + finished_at TEXT, + error TEXT + )""" + ) + with conn: + yield conn + finally: + conn.close() + + +def _process_start_time(pid: int) -> Optional[int]: + try: + from gateway.status import get_process_start_time + + return get_process_start_time(pid) + except Exception: + return None + + +def _owner_is_live(pid: int, started_at: Optional[int]) -> bool: + try: + from gateway.status import _pid_exists + + if not _pid_exists(pid): + return False + except Exception: + return True + if started_at is None: + return pid == os.getpid() + return _process_start_time(pid) == started_at + + +def enqueue(execution_id: str, job: dict, content: str) -> dict: + """Persist one idempotent delivery request before the worker waits.""" + with _transaction() as conn: + conn.execute( + """INSERT OR IGNORE INTO deliveries + (execution_id, job_json, content, status, created_at) + VALUES (?, ?, ?, 'pending', ?)""", + ( + str(execution_id), + json.dumps(job, ensure_ascii=False, sort_keys=True), + str(content), + _hermes_now().isoformat(), + ), + ) + row = conn.execute( + "SELECT * FROM deliveries WHERE execution_id=?", (str(execution_id),) + ).fetchone() + return dict(row) + + +def get_status(execution_id: str) -> Optional[dict]: + with _transaction() as conn: + row = conn.execute( + "SELECT * FROM deliveries WHERE execution_id=?", (str(execution_id),) + ).fetchone() + return dict(row) if row is not None else None + + +def claim_next() -> Optional[dict]: + """Atomically claim one pending send before touching the transport.""" + pid = os.getpid() + started = _process_start_time(pid) + with _transaction() as conn: + row = conn.execute( + "SELECT execution_id FROM deliveries WHERE status='pending' " + "ORDER BY created_at, execution_id LIMIT 1" + ).fetchone() + if row is None: + return None + cur = conn.execute( + """UPDATE deliveries SET status='delivering', owner_process_id=?, + owner_pid=?, owner_started_at=? + WHERE execution_id=? AND status='pending'""", + (_PROCESS_ID, pid, started, row["execution_id"]), + ) + if cur.rowcount != 1: + return None + claimed = conn.execute( + "SELECT * FROM deliveries WHERE execution_id=?", (row["execution_id"],) + ).fetchone() + _ACTIVE_DELIVERIES.add(row["execution_id"]) + result = dict(claimed) + result["job"] = json.loads(result.pop("job_json")) + return result + + +def _finish(execution_id: str, *, error: Optional[str]) -> bool: + status = "failed" if error else "delivered" + with _transaction() as conn: + cur = conn.execute( + """UPDATE deliveries SET status=?, finished_at=?, error=? + WHERE execution_id=? AND status='delivering' + AND owner_process_id=? AND owner_pid=?""", + ( + status, + _hermes_now().isoformat(), + error, + execution_id, + _PROCESS_ID, + os.getpid(), + ), + ) + return cur.rowcount == 1 + + +def recover_abandoned() -> int: + """Fence dead delivery owners as unknown; never replay uncertain sends.""" + changed = 0 + with _transaction() as conn: + rows = conn.execute( + "SELECT execution_id, owner_process_id, owner_pid, owner_started_at " + "FROM deliveries WHERE status='delivering'" + ).fetchall() + for row in rows: + same_process = row["owner_process_id"] == _PROCESS_ID + if same_process: + with _lock: + if row["execution_id"] in _ACTIVE_DELIVERIES: + continue + elif _owner_is_live(int(row["owner_pid"]), row["owner_started_at"]): + continue + error = ( + "Gateway finished delivery but could not persist its outcome; " + "send was not retried." + if same_process + else "Gateway exited during delivery; send outcome is unknown and was not retried." + ) + cur = conn.execute( + """UPDATE deliveries SET status='unknown', finished_at=?, error=? + WHERE execution_id=? AND status='delivering'""", + ( + _hermes_now().isoformat(), + error, + row["execution_id"], + ), + ) + changed += cur.rowcount + return changed + + +def drain(send: Callable[[dict, str], Optional[str]], *, limit: int = 20) -> int: + """Deliver pending rows through *send*, terminalizing every claimed row.""" + recover_abandoned() + processed = 0 + for _ in range(max(0, limit)): + row = claim_next() + if row is None: + break + with _lock: + _ACTIVE_DELIVERIES.add(row["execution_id"]) + try: + try: + error = send(row["job"], row["content"]) + except BaseException as exc: + error = f"{type(exc).__name__}: {exc}" + _finish(row["execution_id"], error=error) + finally: + with _lock: + _ACTIVE_DELIVERIES.discard(row["execution_id"]) + processed += 1 + return processed + + +def enqueue_and_wait( + execution_id: str, + job: dict, + content: str, + *, + timeout: Optional[float] = None, +) -> Optional[str]: + """Queue delivery and wait for a gateway's terminal at-most-once outcome.""" + enqueue(execution_id, job, content) + deadline = None if timeout is None else time.monotonic() + timeout + while deadline is None or time.monotonic() < deadline: + row = get_status(execution_id) + if row and row["status"] in _TERMINAL: + return None if row["status"] == "delivered" else str( + row.get("error") or f"delivery {row['status']}" + ) + time.sleep(0.25) + return "timed out waiting for live gateway delivery; request remains pending" diff --git a/cron/executions.py b/cron/executions.py index efadab464a2f..d05afc2decc6 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -160,6 +160,33 @@ def create_execution(job_id: str, *, source: str) -> Dict[str, Any]: return record # type: ignore[return-value] +def adopt_claimed_execution(execution_id: str) -> Optional[Dict[str, Any]]: + """Atomically transfer and start an attempt in its worker process. + + The dispatching gateway creates the row before spawning a restart-safe + worker. Adoption is the single ``claimed`` → ``running`` gate: only the + winner may acknowledge ownership or run side effects. + """ + pid = os.getpid() + process_started_at = _process_start_time(pid) + now = _hermes_now().isoformat() + with _transaction() as conn: + cur = conn.execute( + """UPDATE executions + SET process_id=?, pid=?, process_started_at=?, + status='running', started_at=? + WHERE id=? AND status='claimed'""", + (_PROCESS_ID, pid, process_started_at, now, execution_id), + ) + if cur.rowcount != 1: + return None + record = _record(conn.execute( + "SELECT * FROM executions WHERE id=?", (execution_id,) + ).fetchone()) + _emit_execution_state(record) + return record + + def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: """Transition one claimed attempt to running exactly once.""" now = _hermes_now().isoformat() diff --git a/cron/scheduler.py b/cron/scheduler.py index ba2d0bde4bde..aeda1b4196a5 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3203,6 +3203,16 @@ def _deliver_result( logger.warning("Job '%s': %s", job["id"], msg) return msg + # Restart-safe workers intentionally have no live gateway adapter objects. + # Hand the send back through a durable queue so the current or replacement + # gateway performs it with relay/E2EE parity. The execution id is the + # idempotency key; the queue never retries an uncertain claimed send. + external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER", "") + if external_execution and adapters is None: + from cron.delivery_queue import enqueue_and_wait + + return enqueue_and_wait(external_execution, job, content) + from tools.send_message_tool import _send_to_platform from gateway.config import load_gateway_config, Platform @@ -4118,6 +4128,20 @@ def _deliver_result( return None +def drain_delivery_queue(adapters, loop) -> int: + """Send queued worker results through this gateway's live adapters.""" + from cron.delivery_queue import drain + + return drain( + lambda queued_job, queued_content: _deliver_result( + queued_job, + queued_content, + adapters=adapters, + loop=loop, + ) + ) + + _DEFAULT_SCRIPT_TIMEOUT = 3600 # seconds (1 hour) # Backward-compatible module override used by tests and emergency monkeypatches. _SCRIPT_TIMEOUT = _DEFAULT_SCRIPT_TIMEOUT @@ -7368,6 +7392,34 @@ def run_one_job( run cooperatively — agent interruption AND script process-tree kill — through the single fenced completion path. """ + # Every gateway path (built-in scheduler, external providers, and direct + # API fires) crosses this seam. Ensure the detached worker has a durable + # attempt to adopt before any launch can occur. + if adapters is not None and not job.get("execution_id"): + execution = create_execution(job["id"], source="direct") + job["execution_id"] = execution["id"] + + if adapters is not None: + try: + if _launch_external_cron_worker(job): + return True + except Exception as handoff_error: + error = f"Restart-safe cron worker dispatch failed: {handoff_error}" + logger.error("Job '%s': %s", job["id"], error) + claim = job.get("fire_claim") + owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" + try: + mark_job_run( + job["id"], + False, + error, + **({"expected_fire_owner": owner} if owner else {}), + ) + finally: + execution_id = job.get("execution_id") + if execution_id: + finish_execution(execution_id, success=False, error=error) + return True if extra_prompt is None: # A gateway-forwarded manual run (`hermes cron run --prompt` / # cronjob(action='run', prompt=...) on a relay-fronted target) stamps @@ -7491,7 +7543,16 @@ def _fire_claim_ownership_lost() -> bool: # The attempt is claimed durably before executor/provider dispatch and # becomes running only immediately before the actual run. - mark_execution_running(execution_id) + # Detached workers atomically transition the attempt to running while + # adopting it. In-process paths must win the claimed->running CAS + # here before any user script or agent side effect may begin. + external_owner = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") == execution_id + if not external_owner and mark_execution_running(execution_id) is None: + logger.warning( + "Cron job %s lost execution ownership before start; skipping", + job["id"], + ) + return True # Run and deliver under the profile's secret scope. get_secret() fails # closed outside a scope once profile isolation is active, and cron @@ -7967,6 +8028,210 @@ def _fire_claim_ownership_lost() -> bool: reset_terminal_scope(_terminal_scope_token) +def _launch_external_cron_worker(job: dict) -> bool: + """Launch *job* outside a managed gateway cgroup when required. + + Returns ``False`` when the caller is not a managed systemd gateway and the + existing in-process path should be used. In managed topology, failure to + establish the transient scope raises: falling back would recreate the + restart interruption this handoff exists to prevent. + """ + execution_id = str(job["execution_id"]) + job_id = str(job["id"]) + handoff_dir = _get_hermes_home() / "cron" / "external-workers" + payload_path = handoff_dir / f"{execution_id}.json" + ack_path = handoff_dir / f"{execution_id}.ready" + command = [ + sys.executable, + "-m", + "cron.scheduler", + "--external-worker-file", + str(payload_path), + "--ack-file", + str(ack_path), + ] + + from agent.secret_scope import is_multiplex_active + from tools.environments.local import build_subprocess_env + from tools.process_registry import restart_safe_gateway_child_argv + + multiplex_active = is_multiplex_active() + scoped_command = restart_safe_gateway_child_argv( + command, + unit_suffix=f"cron-{job_id}-exec-{execution_id}", + ) + if scoped_command == command: + return False + + _ensure_cron_dir(handoff_dir) + try: + handoff_dir.chmod(0o700) + except OSError: + pass + fd = os.open(payload_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(fd, "w", encoding="utf-8") as payload_file: + json.dump( + { + "job": job, + "profile_home": str(_get_hermes_home().resolve()), + "multiplex_active": multiplex_active, + }, + payload_file, + ) + payload_file.flush() + os.fsync(payload_file.fileno()) + except BaseException: + payload_path.unlink(missing_ok=True) + raise + + worker_env = build_subprocess_env( + scrub_secrets=multiplex_active, + inherit_profile_home=True, + extra={"HERMES_HOME": str(_get_hermes_home().resolve())}, + ) + try: + process = subprocess.Popen( + scoped_command, + cwd=str(Path(__file__).resolve().parent.parent), + env=worker_env, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + creationflags=windows_hide_flags(), + ) + except BaseException: + payload_path.unlink(missing_ok=True) + raise + + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + if ack_path.exists(): + try: + acknowledgement = json.loads(ack_path.read_text(encoding="utf-8")) + except Exception: + logger.exception( + "Cron external worker %s published an unreadable acknowledgement; " + "treating handoff as ownership-uncertain", + execution_id, + ) + return True + finally: + ack_path.unlink(missing_ok=True) + if acknowledgement.get("execution_id") != execution_id: + logger.error( + "Cron external worker acknowledgement mismatch for %s; " + "treating handoff as ownership-uncertain", + execution_id, + ) + return True + logger.info( + "Cron job '%s' handed to restart-safe worker pid=%s execution=%s", + job_id, + acknowledgement.get("pid"), + execution_id, + ) + return True + returncode = process.poll() + if returncode is not None: + payload_path.unlink(missing_ok=True) + raise RuntimeError( + f"cron external worker exited before ownership acknowledgement " + f"(exit {returncode})" + ) + time.sleep(0.05) + + # The child may have adopted the durable row just before publishing its + # acknowledgement. Never fall back to in-process execution on an uncertain + # handoff: that could duplicate side effects. The execution owner/dead-owner + # recovery ledger remains the authority. + logger.warning( + "Cron external worker for job '%s' did not acknowledge within 5s; " + "leaving the durable execution claim untouched", + job_id, + ) + return True + + +def _run_external_worker_payload(payload_path: Path, ack_path: Path) -> bool: + """Adopt and execute one gateway-dispatched cron payload. + + The execution row is created by the gateway before spawn, then transferred + here before the ready acknowledgement is published. No side effect runs + unless that durable ownership transfer succeeds. + """ + try: + payload = json.loads(payload_path.read_text(encoding="utf-8")) + job = payload["job"] + profile_home = Path(payload["profile_home"]).resolve() + execution_id = str(job["execution_id"]) + except Exception: + logger.exception("Cron external worker could not load payload %s", payload_path) + return False + finally: + try: + payload_path.unlink(missing_ok=True) + except OSError: + pass + + from agent.secret_scope import ( + build_profile_secret_scope, + is_multiplex_active, + reset_secret_scope, + set_multiplex_active, + set_secret_scope, + ) + from cron.executions import adopt_claimed_execution + from hermes_cli.env_loader import hydrate_profile_secret_sources + from hermes_constants import ( + reset_hermes_home_override, + set_hermes_home_override, + ) + + home_token = set_hermes_home_override(profile_home) + previous_multiplex = is_multiplex_active() + multiplex_active = bool(payload.get("multiplex_active", False)) + set_multiplex_active(multiplex_active) + hydrate_profile_secret_sources(profile_home) + secret_token = set_secret_scope(build_profile_secret_scope(profile_home)) + try: + with use_cron_store(profile_home): + if adopt_claimed_execution(execution_id) is None: + logger.error( + "Cron external worker refused execution %s: durable ownership " + "could not be established", + execution_id, + ) + return False + try: + ack_path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(ack_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as ack_file: + json.dump({"pid": os.getpid(), "execution_id": execution_id}, ack_file) + ack_file.flush() + os.fsync(ack_file.fileno()) + except Exception: + logger.exception( + "Cron external worker could not publish ready acknowledgement for %s", + execution_id, + ) + return False + old_external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") + os.environ["_HERMES_CRON_EXTERNAL_WORKER"] = execution_id + try: + return run_one_job(job, adapters=None, loop=None, verbose=False) + finally: + if old_external_execution is None: + os.environ.pop("_HERMES_CRON_EXTERNAL_WORKER", None) + else: + os.environ["_HERMES_CRON_EXTERNAL_WORKER"] = old_external_execution + finally: + reset_secret_scope(secret_token) + set_multiplex_active(previous_multiplex) + reset_hermes_home_override(home_token) + + def _notify_provider_jobs_changed() -> None: """Best-effort: tell the active scheduler provider the job set changed. @@ -8582,4 +8847,14 @@ def _on_done(_f: concurrent.futures.Future) -> None: if __name__ == "__main__": + if "--external-worker-file" in sys.argv: + import argparse + + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument("--external-worker-file", type=Path, required=True) + parser.add_argument("--ack-file", type=Path, required=True) + args = parser.parse_args() + raise SystemExit( + 0 if _run_external_worker_payload(args.external_worker_file, args.ack_file) else 1 + ) tick(verbose=True) diff --git a/gateway/run.py b/gateway/run.py index 26024795258e..27f88ea75f17 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -33276,7 +33276,30 @@ def _run_planned_stop_watcher( stop_event.wait(poll_interval) -def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60, cron_provider=None): +def _drain_restart_safe_cron_deliveries(adapters, loop, runner=None) -> None: + """Drain each profile's worker queue through its matching live adapters.""" + from cron import scheduler as cron_scheduler + + if adapters: + cron_scheduler.drain_delivery_queue(adapters, loop) + if runner is None: + return + for profile_name, profile_home in _handoff_watch_scopes(runner)[1:]: + profile_adapters = getattr(runner, "_profile_adapters", {}).get(profile_name) + if not profile_adapters: + continue + with _profile_runtime_scope(profile_home): + cron_scheduler.drain_delivery_queue(profile_adapters, loop) + + +def _start_gateway_housekeeping( + stop_event: threading.Event, + adapters=None, + loop=None, + interval: int = 60, + cron_provider=None, + runner=None, +): """Background thread for gateway-only periodic chores (NOT cron). Split out of the historical ``_start_cron_ticker`` so the cron *trigger* @@ -33331,6 +33354,19 @@ def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop while not stop_event.is_set(): tick_count += 1 + # Restart-safe cron workers run outside the gateway cgroup and queue + # their final send for whichever gateway instance is live. Drain on + # the gateway-wide housekeeper rather than the built-in scheduler tick: + # external providers do not run that ticker. + profile_adapters = ( + getattr(runner, "_profile_adapters", {}) if runner is not None else {} + ) + if adapters or any(profile_adapters.values()): + try: + _drain_restart_safe_cron_deliveries(adapters, loop, runner) + except Exception as exc: + logger.debug("Cron durable delivery queue drain error: %s", exc) + if tick_count % CHANNEL_DIR_EVERY == 0 and adapters: try: from gateway.channel_directory import build_channel_directory @@ -34512,6 +34548,7 @@ def _request() -> None: "adapters": runner.adapters, "loop": asyncio.get_running_loop(), "cron_provider": cron_provider, + "runner": runner, }, daemon=True, name="gateway-housekeeping", diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 198669792e9e..d685853066c9 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -10717,6 +10717,32 @@ def _retag_legacy_worker_sessions(workspaces_root_path: str) -> None: _log.debug("kanban worker: legacy session retag skipped (%s)", exc) +def _restart_safe_worker_argv(task: Task, command: list[str]) -> list[str]: + """Wrap a managed-gateway worker in the shared restart-safe scope.""" + if task.current_run_id is None: + # Outside managed systemd this is harmless, but a managed dispatch must + # never mint an untraceable scope. Check topology through the shared + # helper first, using a placeholder suffix that cannot be launched. + from tools.process_registry import restart_safe_gateway_child_argv + + scoped = restart_safe_gateway_child_argv( + command, unit_suffix=f"kanban-{task.id}-run-missing" + ) + if scoped is not command: + raise RuntimeError( + "cannot create restart-safe systemd scope for Kanban worker: " + "the claimed task has no current run id" + ) + return command + + from tools.process_registry import restart_safe_gateway_child_argv + + return restart_safe_gateway_child_argv( + command, + unit_suffix=f"kanban-{task.id}-run-{task.current_run_id}", + ) + + def _default_spawn( task: Task, workspace: str, @@ -10744,7 +10770,13 @@ def _default_spawn( profile_arg = normalize_profile_name(task.assignee) prompt = f"work kanban task {task.id}" - env = dict(os.environ) + from agent.secret_scope import is_multiplex_active + from tools.environments.local import build_subprocess_env + + env = build_subprocess_env( + scrub_secrets=is_multiplex_active(), + inherit_profile_home=True, + ) # The dispatcher is detached from every conversation. Its worker must never # inherit routing mirrored by a previous gateway turn, even before the first # session binds ContextVars in this process. @@ -10895,6 +10927,12 @@ def _default_spawn( # turn, prints text, exits rc=0, and the dispatcher records a # protocol violation (incident 2026-06-09 t_d9cbe312). cmd.append("-Q") + + # A worker spawned by a managed systemd gateway must leave the gateway's + # cgroup before startup; otherwise restarting the service kills the worker + # that is performing the handoff. + cmd = _restart_safe_worker_argv(task, cmd) + # Redirect output to a per-task log under /logs/. # Anchored at the board root (not the shared kanban root), so # `hermes kanban log` on a specific board reads its own file and diff --git a/tests/cron/test_delivery_queue.py b/tests/cron/test_delivery_queue.py new file mode 100644 index 000000000000..7fc65171585a --- /dev/null +++ b/tests/cron/test_delivery_queue.py @@ -0,0 +1,94 @@ +"""Durable at-most-once delivery handoff for restart-safe cron workers.""" + +from __future__ import annotations + +from unittest.mock import Mock + +import pytest + + +def test_pending_delivery_is_claimed_and_sent_once(tmp_path, monkeypatch): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue("exec-1", {"id": "job-1"}, "brief") + send = Mock(return_value=None) + + assert queue.drain(send) == 1 + assert queue.drain(send) == 0 + send.assert_called_once_with({"id": "job-1"}, "brief") + assert queue.get_status("exec-1")["status"] == "delivered" + + +def test_dead_delivery_owner_becomes_unknown_and_is_not_retried( + tmp_path, monkeypatch +): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue("exec-1", {"id": "job-1"}, "brief") + assert queue.claim_next() is not None + monkeypatch.setattr(queue, "_PROCESS_ID", "replacement-gateway") + monkeypatch.setattr(queue, "_owner_is_live", lambda _pid, _started: False) + + assert queue.recover_abandoned() == 1 + send = Mock() + assert queue.drain(send) == 0 + send.assert_not_called() + assert queue.get_status("exec-1")["status"] == "unknown" + + +def test_delivery_failure_is_terminal_and_not_retried(tmp_path, monkeypatch): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue("exec-1", {"id": "job-1"}, "brief") + send = Mock(return_value="transport failed") + + assert queue.drain(send) == 1 + assert queue.drain(send) == 0 + assert send.call_count == 1 + status = queue.get_status("exec-1") + assert status["status"] == "failed" + assert status["error"] == "transport failed" + + +def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + job = {"id": "job-3", "deliver": "origin"} + + error = queue.enqueue_and_wait("exec-3", job, "result", timeout=0) + + assert "timed out" in error + assert queue.get_status("exec-3")["status"] == "pending" + send = Mock(return_value=None) + assert queue.drain(send) == 1 + send.assert_called_once() + + +def test_same_gateway_recovers_terminalization_failure_without_resending( + tmp_path, monkeypatch +): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue("exec-4", {"id": "job-4"}, "result") + send = Mock(return_value=None) + original_finish = queue._finish + monkeypatch.setattr( + queue, + "_finish", + Mock(side_effect=OSError("database temporarily unavailable")), + ) + + with pytest.raises(OSError, match="temporarily unavailable"): + queue.drain(send) + + monkeypatch.setattr(queue, "_finish", original_finish) + assert queue.drain(send) == 0 + send.assert_called_once() + status = queue.get_status("exec-4") + assert status["status"] == "unknown" + assert "not retried" in status["error"] diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index ffe39a516451..f2b8a9cacf78 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -221,7 +221,7 @@ def test_run_one_job_records_running_then_terminal(monkeypatch): monkeypatch.setattr( scheduler, "mark_execution_running", - lambda execution_id: events.append(("running", execution_id)), + lambda execution_id: events.append(("running", execution_id)) or {}, raising=False, ) monkeypatch.setattr( diff --git a/tests/cron/test_parallel_pool.py b/tests/cron/test_parallel_pool.py index b159f265d761..9853dbc2291e 100644 --- a/tests/cron/test_parallel_pool.py +++ b/tests/cron/test_parallel_pool.py @@ -187,7 +187,7 @@ def create_execution_side_effect(job_id, source): if job_id == "healthy-job" else None, ) - monkeypatch.setattr(sched, "mark_execution_running", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "mark_execution_running", lambda *_a, **_kw: {}) monkeypatch.setattr(sched, "heartbeat_fire_claim", lambda *_a, **_kw: True) n = sched.tick(verbose=False) diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py new file mode 100644 index 000000000000..e0c0207f41ab --- /dev/null +++ b/tests/cron/test_restart_safe_worker.py @@ -0,0 +1,245 @@ +"""Restart-safe cron worker handoff and ownership contracts.""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import Mock + +import pytest + + +@pytest.fixture +def execution_ledger(tmp_path, monkeypatch): + import cron.executions as executions + + monkeypatch.setattr(executions, "EXECUTIONS_FILE", tmp_path / "executions.db") + return executions + + +def test_execution_owner_moves_to_external_worker_before_running( + execution_ledger, monkeypatch +): + record = execution_ledger.create_execution("job-1", source="builtin") + monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242) + monkeypatch.setattr(execution_ledger, "_process_start_time", lambda pid: 9876) + + adopted = execution_ledger.adopt_claimed_execution(record["id"]) + + assert adopted is not None + assert adopted["pid"] == 4242 + assert adopted["process_started_at"] == 9876 + assert adopted["status"] == "running" + assert execution_ledger.adopt_claimed_execution(record["id"]) is None + assert execution_ledger.mark_execution_running(record["id"]) is None + + +def test_restart_safe_gateway_child_fails_closed_without_scope(monkeypatch): + import tools.process_registry as process_registry + + monkeypatch.setattr(process_registry, "_IS_WINDOWS", False) + monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) + monkeypatch.setenv("INVOCATION_ID", "managed-service") + monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", lambda: False) + + with pytest.raises(RuntimeError, match="systemd-run --user --scope is unavailable"): + process_registry.restart_safe_gateway_child_argv( + ["python", "worker.py"], unit_suffix="cron-job-1" + ) + + +def test_restart_safe_gateway_child_is_unchanged_outside_managed_gateway(monkeypatch): + import tools.process_registry as process_registry + + command = ["python", "worker.py"] + monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: False) + + assert process_registry.restart_safe_gateway_child_argv( + command, unit_suffix="cron-job-1" + ) is command + + +def test_external_worker_adopts_execution_and_runs_payload_once( + tmp_path, monkeypatch +): + import cron.scheduler as scheduler + + payload = tmp_path / "payload.json" + ack = tmp_path / "ready.json" + payload.write_text( + json.dumps({ + "job": {"id": "job-1", "execution_id": "exec-1"}, + "profile_home": str(tmp_path / "profile"), + }), + encoding="utf-8", + ) + from hermes_constants import get_hermes_home + + observed_homes = [] + adopted = Mock( + side_effect=lambda execution_id: ( + observed_homes.append(get_hermes_home().resolve()) + or {"id": execution_id, "status": "running"} + ) + ) + run = Mock( + side_effect=lambda *_args, **_kwargs: ( + observed_homes.append(get_hermes_home().resolve()) or True + ) + ) + monkeypatch.setattr("cron.executions.adopt_claimed_execution", adopted) + monkeypatch.setattr(scheduler, "run_one_job", run) + + assert scheduler._run_external_worker_payload(payload, ack) is True + + adopted.assert_called_once_with("exec-1") + run.assert_called_once() + assert run.call_args.args[0]["id"] == "job-1" + expected_home = (tmp_path / "profile").resolve() + assert observed_homes == [expected_home, expected_home] + assert ack.exists() + assert not payload.exists() + + +def test_external_worker_refuses_to_run_without_durable_ownership( + tmp_path, monkeypatch +): + import cron.scheduler as scheduler + + payload = tmp_path / "payload.json" + ack = tmp_path / "ready.json" + payload.write_text( + json.dumps({ + "job": {"id": "job-1", "execution_id": "exec-1"}, + "profile_home": str(tmp_path / "profile"), + }), + encoding="utf-8", + ) + monkeypatch.setattr("cron.executions.adopt_claimed_execution", lambda _id: None) + run = Mock() + monkeypatch.setattr(scheduler, "run_one_job", run) + + assert scheduler._run_external_worker_payload(payload, ack) is False + + run.assert_not_called() + assert not ack.exists() + + +def test_launch_external_worker_uses_restart_safe_scope_and_acknowledges( + tmp_path, monkeypatch +): + import cron.scheduler as scheduler + + job = {"id": "job-1", "execution_id": "exec-1", "prompt": "work"} + monkeypatch.setattr(scheduler, "_get_hermes_home", lambda: tmp_path) + wrapped_commands = [] + + def wrap(command, *, unit_suffix): + wrapped_commands.append((command, unit_suffix)) + return ["scope", "--", *command] + + monkeypatch.setattr( + "tools.process_registry.restart_safe_gateway_child_argv", wrap + ) + + class FakeProcess: + returncode = None + + def poll(self): + return self.returncode + + spawned = [] + + def popen(command, **kwargs): + spawned.append((command, kwargs)) + ack_index = command.index("--ack-file") + 1 + Path(command[ack_index]).write_text( + json.dumps({"pid": 4321, "execution_id": "exec-1"}), + encoding="utf-8", + ) + return FakeProcess() + + monkeypatch.setattr(scheduler.subprocess, "Popen", popen) + monkeypatch.setenv("ANTHROPIC_API_KEY", "should-not-cross-profile") + from agent.secret_scope import set_multiplex_active + + set_multiplex_active(True) + try: + assert scheduler._launch_external_cron_worker(job) is True + finally: + set_multiplex_active(False) + assert wrapped_commands[0][1] == "cron-job-1-exec-exec-1" + assert spawned[0][0][0:2] == ["scope", "--"] + assert spawned[0][1]["start_new_session"] is True + assert "ANTHROPIC_API_KEY" not in spawned[0][1]["env"] + payload = json.loads((tmp_path / "cron/external-workers/exec-1.json").read_text()) + assert payload["multiplex_active"] is True + + +def test_launch_external_worker_stays_in_process_outside_managed_gateway( + monkeypatch, +): + import cron.scheduler as scheduler + + command_calls = [] + + def unchanged(command, *, unit_suffix): + command_calls.append((command, unit_suffix)) + return command + + monkeypatch.setattr( + "tools.process_registry.restart_safe_gateway_child_argv", unchanged + ) + popen = Mock() + monkeypatch.setattr(scheduler.subprocess, "Popen", popen) + + assert scheduler._launch_external_cron_worker( + {"id": "job-1", "execution_id": "exec-1"} + ) is False + assert command_calls + popen.assert_not_called() + + +def test_shared_run_path_hands_gateway_fire_to_external_worker(monkeypatch): + import cron.scheduler as scheduler + + launch = Mock(return_value=True) + run = Mock(side_effect=AssertionError("agent ran inside gateway")) + monkeypatch.setattr(scheduler, "_launch_external_cron_worker", launch) + monkeypatch.setattr(scheduler, "run_job", run) + job = {"id": "job-1", "execution_id": "exec-1"} + + assert scheduler.run_one_job(job, adapters={"discord": object()}) is True + + launch.assert_called_once_with(job) + run.assert_not_called() + + +def test_shared_run_path_creates_execution_before_managed_handoff(monkeypatch): + import cron.scheduler as scheduler + + created = Mock(return_value={"id": "exec-new"}) + launch = Mock(return_value=True) + monkeypatch.setattr(scheduler, "create_execution", created) + monkeypatch.setattr(scheduler, "_launch_external_cron_worker", launch) + job = {"id": "manual-job"} + + assert scheduler.run_one_job(job, adapters={"discord": object()}) is True + + created.assert_called_once_with("manual-job", source="direct") + assert job["execution_id"] == "exec-new" + launch.assert_called_once_with(job) + + +def test_lost_execution_start_cas_prevents_side_effects(monkeypatch): + import cron.scheduler as scheduler + + run = Mock(side_effect=AssertionError("side effect ran without ownership")) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(scheduler, "run_job", run) + + assert scheduler.run_one_job( + {"id": "job-1", "execution_id": "exec-1"}, adapters=None + ) is True + run.assert_not_called() diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index 93bcef00a519..9b241de1b63e 100644 --- a/tests/cron/test_run_one_job.py +++ b/tests/cron/test_run_one_job.py @@ -88,7 +88,7 @@ def test_run_one_job_exception_delivers_failure_alert(monkeypatch): s, "create_execution", lambda *_a, **_kw: {"id": "exec-j3"} ) monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True) - monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: {}) monkeypatch.setattr( s, "run_job", @@ -141,7 +141,7 @@ def test_run_one_job_exception_records_failure_alert_delivery_error(monkeypatch) s, "create_execution", lambda *_a, **_kw: {"id": "exec-j4"} ) monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True) - monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: {}) monkeypatch.setattr( s, "run_job", @@ -165,7 +165,7 @@ def _patch_escaped_failure(monkeypatch, delivered, *, exec_id, err): """Make run_job raise, and capture what the escape handler delivers.""" monkeypatch.setattr(s, "create_execution", lambda *_a, **_kw: {"id": exec_id}) monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True) - monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: {}) monkeypatch.setattr( s, "run_job", @@ -246,7 +246,7 @@ def test_run_one_job_exception_after_delivery_does_not_redeliver(monkeypatch): s, "create_execution", lambda *_a, **_kw: {"id": "exec-j5"} ) monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True) - monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: {}) monkeypatch.setattr( s, "run_job", @@ -288,7 +288,7 @@ def test_run_one_job_keyboard_interrupt_skips_delivery_and_reraises(monkeypatch) s, "create_execution", lambda *_a, **_kw: {"id": "exec-j6"} ) monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True) - monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: {}) monkeypatch.setattr( s, "run_job", diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index effb3f8dd40b..a989b01457eb 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -564,7 +564,7 @@ def owned_fence(*_args, **_kwargs): finish = MagicMock() monkeypatch.setattr(scheduler, "heartbeat_fire_claim", lambda *args, **kwargs: True) monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_args, **_kwargs: True) - monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_args: None) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_args: {}) monkeypatch.setattr( scheduler, "run_job", diff --git a/tests/gateway/test_cron_delivery_housekeeping.py b/tests/gateway/test_cron_delivery_housekeeping.py new file mode 100644 index 000000000000..8e38071679c9 --- /dev/null +++ b/tests/gateway/test_cron_delivery_housekeeping.py @@ -0,0 +1,82 @@ +"""Gateway-independent draining of restart-safe cron deliveries.""" + +from contextlib import contextmanager +from types import SimpleNamespace + +import cron.scheduler as scheduler +import gateway.run as gateway_run + + +class _OneTickStopEvent: + def __init__(self): + self.waited = False + + def is_set(self): + return self.waited + + def wait(self, timeout=None): + self.waited = True + return True + + +def test_gateway_housekeeping_drains_cron_delivery_with_live_adapters(monkeypatch): + adapters = {"discord": object()} + loop = object() + calls = [] + monkeypatch.setattr( + scheduler, + "drain_delivery_queue", + lambda live_adapters, live_loop: calls.append((live_adapters, live_loop)), + raising=False, + ) + + gateway_run._start_gateway_housekeeping( + _OneTickStopEvent(), adapters=adapters, loop=loop, interval=0 + ) + + assert calls == [(adapters, loop)] + + +def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( + tmp_path, monkeypatch +): + root_adapters = {} + secondary_adapters = {"telegram": "secondary"} + runner = SimpleNamespace( + config=SimpleNamespace(multiplex_profiles=True), + adapters=root_adapters, + _profile_adapters={"secondary": secondary_adapters}, + ) + secondary_home = tmp_path / "secondary" + calls = [] + + monkeypatch.setattr( + gateway_run, + "_handoff_watch_scopes", + lambda _runner: [(None, None), ("secondary", secondary_home)], + ) + + @contextmanager + def fake_scope(home): + calls.append(("scope", home)) + yield + + monkeypatch.setattr(gateway_run, "_profile_runtime_scope", fake_scope) + monkeypatch.setattr( + scheduler, + "drain_delivery_queue", + lambda adapters, loop: calls.append(("drain", adapters)), + ) + + gateway_run._start_gateway_housekeeping( + _OneTickStopEvent(), + adapters=root_adapters, + loop=object(), + interval=0, + runner=runner, + ) + + assert calls == [ + ("scope", secondary_home), + ("drain", secondary_adapters), + ] diff --git a/tests/hermes_cli/test_kanban_gateway_restart_handoff.py b/tests/hermes_cli/test_kanban_gateway_restart_handoff.py new file mode 100644 index 000000000000..d0efb6ed893d --- /dev/null +++ b/tests/hermes_cli/test_kanban_gateway_restart_handoff.py @@ -0,0 +1,178 @@ +"""Managed-gateway isolation for dispatcher-owned Kanban workers.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def worker_setup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[Path, kb.Task]: + root = tmp_path / ".hermes" + profile = root / "profiles" / "coder" + profile.mkdir(parents=True) + root.joinpath("config.yaml").write_text("{}\n", encoding="utf-8") + profile.joinpath("config.yaml").write_text("{}\n", encoding="utf-8") + monkeypatch.setenv("HERMES_HOME", str(root)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: ["hermes"]) + + workspace = tmp_path / "candidate-worktree" + workspace.mkdir() + task = kb.Task( + id="t_candidate_restart", + title="activate candidate", + body=None, + assignee="coder", + status="running", + priority=0, + created_by="test", + created_at=1, + started_at=1, + completed_at=None, + workspace_kind="worktree", + workspace_path=str(workspace), + claim_lock="host:dispatcher", + claim_expires=999, + tenant=None, + branch_name="wt/t_candidate_restart", + current_run_id=23, + ) + return workspace, task + + +@pytest.mark.linux_only +def test_managed_gateway_worker_is_spawned_in_restart_safe_scope( + worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch +) -> None: + workspace, task = worker_setup + captured_cmd: list[str] = [] + captured_env: dict[str, str] = {} + captured_cwd: str | None = None + + class FakeProc: + pid = 4242 + + def fake_popen(cmd, **kwargs): + nonlocal captured_cwd + captured_cmd.extend(cmd) + captured_env.update(kwargs.get("env") or {}) + captured_cwd = kwargs.get("cwd") + return FakeProc() + + monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") + monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-cross-profile") + monkeypatch.setattr("agent.secret_scope.is_multiplex_active", lambda: True) + monkeypatch.setattr(subprocess, "Popen", fake_popen) + monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: True) + monkeypatch.setattr("tools.process_registry._systemd_run_user_scope_available", lambda: True) + monkeypatch.setattr("tools.process_registry._worker_memory_max_bytes", lambda: 536_870_912) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run") + + assert kb._default_spawn(task, str(workspace)) == 4242 + assert captured_cmd[:4] == ["/usr/bin/systemd-run", "--user", "--scope", "--quiet"] + unit_index = captured_cmd.index("--unit") + assert captured_cmd[unit_index + 1] == "hermes-worker-kanban-t_candidate_restart-run-23" + assert "MemoryMax=536870912" in captured_cmd + separator = captured_cmd.index("--") + assert captured_cmd[separator + 1 : separator + 4] == ["hermes", "-p", "coder"] + assert captured_cwd == str(workspace) + assert captured_env["HERMES_KANBAN_TASK"] == task.id + assert captured_env["HERMES_KANBAN_RUN_ID"] == "23" + assert "ANTHROPIC_API_KEY" not in captured_env + + +@pytest.mark.linux_only +def test_managed_gateway_worker_spawn_fails_closed_without_scope( + worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch +) -> None: + workspace, task = worker_setup + popen_calls: list[list[str]] = [] + monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") + monkeypatch.setattr(subprocess, "Popen", lambda cmd, **kwargs: popen_calls.append(list(cmd))) + monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: True) + monkeypatch.setattr("tools.process_registry._systemd_run_user_scope_available", lambda: False) + + with pytest.raises(RuntimeError, match="restart-safe systemd scope"): + kb._default_spawn(task, str(workspace)) + assert popen_calls == [] + + +@pytest.mark.linux_only +def test_managed_gateway_scope_builder_fails_closed_if_binary_disappears( + worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch +) -> None: + workspace, task = worker_setup + monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") + monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: True) + monkeypatch.setattr("tools.process_registry._systemd_run_user_scope_available", lambda: True) + monkeypatch.setattr("shutil.which", lambda _name: None) + monkeypatch.setattr(subprocess, "Popen", lambda *_args, **_kwargs: pytest.fail("unsafe direct spawn")) + + with pytest.raises(RuntimeError, match="restart-safe systemd scope"): + kb._default_spawn(task, str(workspace)) + + +def test_standalone_dispatcher_keeps_direct_worker_spawn( + worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch +) -> None: + workspace, task = worker_setup + captured_cmd: list[str] = [] + + class FakeProc: + pid = 4243 + + monkeypatch.setattr(subprocess, "Popen", lambda cmd, **kwargs: captured_cmd.extend(cmd) or FakeProc()) + monkeypatch.setattr("tools.process_registry._is_supervised_gateway_process", lambda: False) + monkeypatch.setattr( + "tools.process_registry._systemd_run_user_scope_available", + lambda: pytest.fail("scope probe must not run outside managed gateway"), + ) + + assert kb._default_spawn(task, str(workspace)) == 4243 + assert captured_cmd[:3] == ["hermes", "-p", "coder"] + + +@pytest.mark.linux_only +def test_real_user_systemd_scope_preserves_worker_context( + worker_setup: tuple[Path, kb.Task], monkeypatch: pytest.MonkeyPatch +) -> None: + from tools import process_registry + + if not process_registry._systemd_run_user_scope_available(): + pytest.skip("systemd-run --user --scope is unavailable on this host") + + workspace, task = worker_setup + receipt = workspace / "worker-receipt.json" + script = ( + "import json, os, pathlib, sys, time; " + "pathlib.Path(sys.argv[1]).write_text(json.dumps({" + "'pid': os.getpid(), 'cwd': os.getcwd(), " + "'task': os.environ.get('HERMES_KANBAN_TASK'), " + "'run': os.environ.get('HERMES_KANBAN_RUN_ID'), " + "'cgroup': pathlib.Path('/proc/self/cgroup').read_text()})); time.sleep(0.5)" + ) + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: [sys.executable, "-c", script, str(receipt)]) + monkeypatch.setenv("INVOCATION_ID", "managed-gateway-test") + monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) + + pid = kb._default_spawn(task, str(workspace)) + deadline = time.monotonic() + 5 + while not receipt.exists() and time.monotonic() < deadline: + time.sleep(0.05) + + assert receipt.exists() + payload = json.loads(receipt.read_text(encoding="utf-8")) + assert payload["pid"] == pid + assert payload["cwd"] == str(workspace) + assert payload["task"] == task.id + assert payload["run"] == "23" + assert ".scope" in payload["cgroup"] + assert "hermes-gateway.service" not in payload["cgroup"] diff --git a/tools/process_registry.py b/tools/process_registry.py index 8937c4cce8fc..4ec461df14b2 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -323,6 +323,35 @@ def _build_systemd_scope_argv( ] +def restart_safe_gateway_child_argv( + command: List[str], *, unit_suffix: str +) -> List[str]: + """Place a managed-systemd gateway child outside the gateway cgroup. + + Children that must survive an intentional gateway restart cannot rely on + ``start_new_session`` alone: systemd still kills every process in the + service cgroup. In that topology, require a transient user scope and fail + closed if it cannot be established. Standalone processes, non-systemd + supervisors, and non-Linux hosts retain the direct command. + """ + if _IS_WINDOWS: + return command + if not _is_supervised_gateway_process() or not os.environ.get("INVOCATION_ID"): + return command + if not _systemd_run_user_scope_available(): + raise RuntimeError( + "cannot create restart-safe systemd scope for gateway child: " + "systemd-run --user --scope is unavailable" + ) + scoped = _build_systemd_scope_argv(command, unit_suffix=unit_suffix) + if scoped == command: + raise RuntimeError( + "cannot create restart-safe systemd scope for gateway child: " + "systemd-run disappeared after the availability probe" + ) + return scoped + + def _stop_systemd_unit(unit_name: str) -> bool: """Stop a transient systemd user scope by unit name. From b7a6d6f575b302152fdd01733b0bb23650c00d60 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Thu, 3 Sep 2026 10:01:35 +0700 Subject: [PATCH 2/5] [verified] fix(cron): harden gateway restart handoff --- cron/delivery_queue.py | 64 ++++- cron/executions.py | 74 ++++- cron/scheduler.py | 130 ++++++++- gateway/run.py | 9 +- tests/cron/test_cron_failure_deliver.py | 2 +- tests/cron/test_delivery_queue.py | 76 ++++- tests/cron/test_execution_ledger.py | 48 ++++ tests/cron/test_restart_safe_worker.py | 269 ++++++++++++++++++ .../test_cron_delivery_housekeeping.py | 21 +- tests/tools/test_cronjob_run_background.py | 22 ++ tools/cronjob_tools.py | 12 + 11 files changed, 691 insertions(+), 36 deletions(-) diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index a7dbc8eea283..99757f7cdff5 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -27,6 +27,34 @@ _lock = threading.RLock() _ACTIVE_DELIVERIES: set[str] = set() _TERMINAL = ("delivered", "failed", "unknown") +MAX_TERMINAL_DELIVERIES = 1000 + + +def _prune_terminal_unlocked(conn: sqlite3.Connection) -> None: + """Redact terminal payloads and retain only bounded outcome metadata.""" + conn.execute( + """UPDATE deliveries SET job_json='{}', content='' + WHERE status IN ('delivered','failed','unknown') + AND (job_json != '{}' OR content != '')""" + ) + keep = max(0, int(MAX_TERMINAL_DELIVERIES)) + terminal_count = int( + conn.execute( + "SELECT COUNT(*) FROM deliveries " + "WHERE status IN ('delivered','failed','unknown')" + ).fetchone()[0] + ) + excess = terminal_count - keep + if excess > 0: + conn.execute( + """DELETE FROM deliveries WHERE execution_id IN ( + SELECT execution_id FROM deliveries + WHERE status IN ('delivered','failed','unknown') + ORDER BY finished_at, created_at, execution_id + LIMIT ? + )""", + (excess,), + ) def _path() -> Path: @@ -53,6 +81,7 @@ def _transaction() -> Iterator[sqlite3.Connection]: execution_id TEXT PRIMARY KEY, job_json TEXT NOT NULL, content TEXT NOT NULL, + for_failure INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL CHECK(status IN ('pending','delivering','delivered','failed','unknown')), owner_process_id TEXT, @@ -63,7 +92,16 @@ def _transaction() -> Iterator[sqlite3.Connection]: error TEXT )""" ) + columns = { + str(row[1]) for row in conn.execute("PRAGMA table_info(deliveries)") + } + if "for_failure" not in columns: + conn.execute( + "ALTER TABLE deliveries " + "ADD COLUMN for_failure INTEGER NOT NULL DEFAULT 0" + ) with conn: + _prune_terminal_unlocked(conn) yield conn finally: conn.close() @@ -91,17 +129,24 @@ def _owner_is_live(pid: int, started_at: Optional[int]) -> bool: return _process_start_time(pid) == started_at -def enqueue(execution_id: str, job: dict, content: str) -> dict: +def enqueue( + execution_id: str, + job: dict, + content: str, + *, + for_failure: bool = False, +) -> dict: """Persist one idempotent delivery request before the worker waits.""" with _transaction() as conn: conn.execute( """INSERT OR IGNORE INTO deliveries - (execution_id, job_json, content, status, created_at) - VALUES (?, ?, ?, 'pending', ?)""", + (execution_id, job_json, content, for_failure, status, created_at) + VALUES (?, ?, ?, ?, 'pending', ?)""", ( str(execution_id), json.dumps(job, ensure_ascii=False, sort_keys=True), str(content), + int(bool(for_failure)), _hermes_now().isoformat(), ), ) @@ -163,6 +208,7 @@ def _finish(execution_id: str, *, error: Optional[str]) -> bool: os.getpid(), ), ) + _prune_terminal_unlocked(conn) return cur.rowcount == 1 @@ -198,10 +244,13 @@ def recover_abandoned() -> int: ), ) changed += cur.rowcount + _prune_terminal_unlocked(conn) return changed -def drain(send: Callable[[dict, str], Optional[str]], *, limit: int = 20) -> int: +def drain( + send: Callable[[dict, str, bool], Optional[str]], *, limit: int = 20 +) -> int: """Deliver pending rows through *send*, terminalizing every claimed row.""" recover_abandoned() processed = 0 @@ -213,7 +262,9 @@ def drain(send: Callable[[dict, str], Optional[str]], *, limit: int = 20) -> int _ACTIVE_DELIVERIES.add(row["execution_id"]) try: try: - error = send(row["job"], row["content"]) + error = send( + row["job"], row["content"], bool(row["for_failure"]) + ) except BaseException as exc: error = f"{type(exc).__name__}: {exc}" _finish(row["execution_id"], error=error) @@ -229,10 +280,11 @@ def enqueue_and_wait( job: dict, content: str, *, + for_failure: bool = False, timeout: Optional[float] = None, ) -> Optional[str]: """Queue delivery and wait for a gateway's terminal at-most-once outcome.""" - enqueue(execution_id, job, content) + enqueue(execution_id, job, content, for_failure=for_failure) deadline = None if timeout is None else time.monotonic() + timeout while deadline is None or time.monotonic() < deadline: row = get_status(execution_id) diff --git a/cron/executions.py b/cron/executions.py index d05afc2decc6..9cf204e559d9 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -10,6 +10,7 @@ import os import sqlite3 import threading +import time import uuid from contextlib import contextmanager from typing import Any, Dict, Iterator, List, Optional @@ -22,6 +23,7 @@ # profile's execution records into the import-time home. EXECUTIONS_FILE: Optional[Path] = None MAX_TERMINAL_EXECUTIONS = 1000 +HANDOFF_ADOPTION_GRACE_SECONDS = 30.0 _TERMINAL_STATES = ("completed", "failed", "unknown") _lock = threading.RLock() _PROCESS_ID = uuid.uuid4().hex @@ -52,12 +54,26 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')), + handoff_pending INTEGER NOT NULL DEFAULT 0, + handoff_started_at REAL, claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT )""" ) + columns = { + str(row[1]) for row in conn.execute("PRAGMA table_info(executions)") + } + if "handoff_pending" not in columns: + conn.execute( + "ALTER TABLE executions " + "ADD COLUMN handoff_pending INTEGER NOT NULL DEFAULT 0" + ) + if "handoff_started_at" not in columns: + conn.execute( + "ALTER TABLE executions ADD COLUMN handoff_started_at REAL" + ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_executions_job_claimed " "ON executions(job_id, claimed_at DESC, id DESC)" @@ -160,6 +176,25 @@ def create_execution(job_id: str, *, source: str) -> Dict[str, Any]: return record # type: ignore[return-value] +def mark_execution_handoff_pending(execution_id: str) -> Optional[Dict[str, Any]]: + """Fence restart recovery while an external worker is adopting a claim.""" + with _transaction() as conn: + cur = conn.execute( + """UPDATE executions + SET handoff_pending=1, handoff_started_at=? + WHERE id=? AND status='claimed' + AND process_id=? AND pid=?""", + (time.time(), execution_id, _PROCESS_ID, os.getpid()), + ) + if cur.rowcount != 1: + return None + record = _record(conn.execute( + "SELECT * FROM executions WHERE id=?", (execution_id,) + ).fetchone()) + _emit_execution_state(record) + return record + + def adopt_claimed_execution(execution_id: str) -> Optional[Dict[str, Any]]: """Atomically transfer and start an attempt in its worker process. @@ -174,7 +209,8 @@ def adopt_claimed_execution(execution_id: str) -> Optional[Dict[str, Any]]: cur = conn.execute( """UPDATE executions SET process_id=?, pid=?, process_started_at=?, - status='running', started_at=? + status='running', started_at=?, handoff_pending=0, + handoff_started_at=NULL WHERE id=? AND status='claimed'""", (_PROCESS_ID, pid, process_started_at, now, execution_id), ) @@ -192,8 +228,10 @@ def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: now = _hermes_now().isoformat() with _transaction() as conn: cur = conn.execute( - """UPDATE executions SET status='running', started_at=? - WHERE id=? AND status='claimed'""", + """UPDATE executions + SET status='running', started_at=?, handoff_pending=0, + handoff_started_at=NULL + WHERE id=? AND status='claimed' AND handoff_pending=0""", (now, execution_id), ) if cur.rowcount != 1: @@ -215,7 +253,9 @@ def finish_execution( detail = None if success else (str(error) if error else "unknown failure") with _transaction() as conn: cur = conn.execute( - """UPDATE executions SET status=?, finished_at=?, error=? + """UPDATE executions + SET status=?, finished_at=?, error=?, handoff_pending=0, + handoff_started_at=NULL WHERE id=? AND status IN ('claimed','running')""", (status, now, detail, execution_id), ) @@ -236,7 +276,9 @@ def recover_interrupted_executions() -> int: recovered: List[Dict[str, Any]] = [] with _transaction() as conn: rows = conn.execute( - """SELECT id, process_id, pid, process_started_at FROM executions + """SELECT id, process_id, pid, process_started_at, + handoff_pending, handoff_started_at + FROM executions WHERE status IN ('claimed','running')""" ).fetchall() for row in rows: @@ -244,8 +286,18 @@ def recover_interrupted_executions() -> int: continue if _owner_is_live(int(row["pid"]), row["process_started_at"]): continue + handoff_started_at = row["handoff_started_at"] + if ( + row["handoff_pending"] + and handoff_started_at is not None + and time.time() - float(handoff_started_at) + < HANDOFF_ADOPTION_GRACE_SECONDS + ): + continue cur = conn.execute( - """UPDATE executions SET status='unknown', finished_at=?, error=? + """UPDATE executions + SET status='unknown', finished_at=?, error=?, + handoff_pending=0, handoff_started_at=NULL WHERE id=? AND status IN ('claimed','running')""", (now, "Scheduler restarted after this execution's owner exited before a durable " @@ -290,6 +342,16 @@ def list_executions( return [dict(row) for row in rows] +def get_execution(execution_id: str) -> Optional[Dict[str, Any]]: + """Return one exact execution attempt, or ``None`` when it is absent.""" + with _transaction() as conn: + row = conn.execute( + "SELECT * FROM executions WHERE id=?", + (str(execution_id),), + ).fetchone() + return dict(row) if row is not None else None + + def latest_execution(job_id: str) -> Optional[Dict[str, Any]]: rows = list_executions(job_id=job_id, limit=1) return rows[0] if rows else None diff --git a/cron/scheduler.py b/cron/scheduler.py index aeda1b4196a5..8191dd710333 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -725,7 +725,14 @@ def _resolve_job_reasoning_config(job: dict, cfg: dict, model: str) -> dict | No save_job_output, use_cron_store, ) -from cron.executions import create_execution, finish_execution, mark_execution_running +from cron.executions import ( + create_execution, + finish_execution, + get_execution, + mark_execution_handoff_pending, + mark_execution_running, + recover_interrupted_executions, +) # Sentinel: when a cron agent has nothing new to report, it can start its # response with this marker to suppress delivery. Output is still saved @@ -769,6 +776,10 @@ def _is_cron_silence_response(text: str) -> bool: _parallel_pool_max_workers: Optional[int] = None _running_job_ids: set = set() _running_fire_owners: dict[str, dict[object, tuple[Optional[str], Path]]] = {} +# Parent gateway threads synchronously waiting on restart-safe scope workers. +# Shutdown must not misclassify these as ownerless in-process runs: the tool +# process sweep cannot reach the worker's transient scope. +_restart_safe_waiter_job_ids: set[str] = set() _running_lock = threading.Lock() # Wall-clock (time.time()) instant each in-flight job id was claimed by @@ -1287,9 +1298,11 @@ def mark_running_jobs_interrupted( Returns the list of job IDs marked, for the caller to log. """ with _running_lock: + restart_safe_waiters = set(_restart_safe_waiter_job_ids) active_fires = [ (token, job_id, owner, profile_home) for job_id, executions in _running_fire_owners.items() + if job_id not in restart_safe_waiters for token, (owner, profile_home) in executions.items() ] if only_owners is not None: @@ -1301,7 +1314,9 @@ def mark_running_jobs_interrupted( if only_owners is None: active_fires.extend( (None, job_id, None, _get_hermes_home()) - for job_id in _running_job_ids - registered_ids + for job_id in ( + _running_job_ids - registered_ids - restart_safe_waiters + ) ) _interrupted_job_ids.update( token if token is not None else job_id @@ -3211,7 +3226,12 @@ def _deliver_result( if external_execution and adapters is None: from cron.delivery_queue import enqueue_and_wait - return enqueue_and_wait(external_execution, job, content) + return enqueue_and_wait( + external_execution, + job, + content, + for_failure=for_failure, + ) from tools.send_message_tool import _send_to_platform from gateway.config import load_gateway_config, Platform @@ -4133,11 +4153,12 @@ def drain_delivery_queue(adapters, loop) -> int: from cron.delivery_queue import drain return drain( - lambda queued_job, queued_content: _deliver_result( + lambda queued_job, queued_content, queued_for_failure: _deliver_result( queued_job, queued_content, adapters=adapters, loop=loop, + for_failure=queued_for_failure, ) ) @@ -7395,11 +7416,13 @@ def run_one_job( # Every gateway path (built-in scheduler, external providers, and direct # API fires) crosses this seam. Ensure the detached worker has a durable # attempt to adopt before any launch can occur. - if adapters is not None and not job.get("execution_id"): + if not job.get("execution_id"): execution = create_execution(job["id"], source="direct") job["execution_id"] = execution["id"] - if adapters is not None: + execution_id = str(job["execution_id"]) + external_owner = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") == execution_id + if not external_owner: try: if _launch_external_cron_worker(job): return True @@ -7416,9 +7439,7 @@ def run_one_job( **({"expected_fire_owner": owner} if owner else {}), ) finally: - execution_id = job.get("execution_id") - if execution_id: - finish_execution(execution_id, success=False, error=error) + finish_execution(execution_id, success=False, error=error) return True if extra_prompt is None: # A gateway-forwarded manual run (`hermes cron run --prompt` / @@ -7521,6 +7542,7 @@ def _fire_claim_ownership_lost() -> bool: ) _scope_token = None + _terminal_scope_token = None try: # Pre-run dispatch claim (issue #38758): atomically commit a finite # one-shot's dispatch BEFORE its side effect runs, so a tick that dies @@ -8028,6 +8050,65 @@ def _fire_claim_ownership_lost() -> bool: reset_terminal_scope(_terminal_scope_token) +def _wait_for_external_cron_worker_body( + process: subprocess.Popen, + *, + execution_id: str, +) -> bool: + """Preserve ``run_one_job``'s synchronous contract after handoff. + + The worker owns the durable execution and survives this gateway process. + The caller nevertheless waits while it remains alive so manual/background + callers do not release their in-process guard or report stale job state. + A gateway replacement may kill this waiter; it does not kill the scoped + worker or change its ledger ownership. + """ + terminal_states = {"completed", "failed", "unknown"} + + def _is_terminal() -> bool: + current = get_execution(execution_id) + return bool(current and current.get("status") in terminal_states) + + while True: + if _is_terminal(): + return True + returncode = process.poll() + if returncode is not None: + # The worker can commit its terminal row and exit between the first + # read and poll(). Re-read the exact attempt before declaring that + # it died without terminalizing. + if _is_terminal(): + return True + # If the adopted worker died without terminalizing, its owner is + # now provably gone. Recover to ``unknown`` rather than routing the + # exception through the pre-handoff dispatch-failure path, which + # would falsely assert that no side effect could have happened. + recover_interrupted_executions() + if _is_terminal(): + return True + raise RuntimeError( + "cron external worker exited before durable recovery could " + f"terminalize its execution state (exit {returncode})" + ) + time.sleep(0.05) + + +def _wait_for_external_cron_worker( + process: subprocess.Popen, + *, + execution_id: str, + job_id: Optional[str] = None, +) -> bool: + try: + return _wait_for_external_cron_worker_body( + process, execution_id=execution_id + ) + finally: + if job_id is not None: + with _running_lock: + _restart_safe_waiter_job_ids.discard(job_id) + + def _launch_external_cron_worker(job: dict) -> bool: """Launch *job* outside a managed gateway cgroup when required. @@ -8063,6 +8144,11 @@ def _launch_external_cron_worker(job: dict) -> bool: if scoped_command == command: return False + if mark_execution_handoff_pending(execution_id) is None: + raise RuntimeError( + "cron execution claim changed before external worker handoff" + ) + _ensure_cron_dir(handoff_dir) try: handoff_dir.chmod(0o700) @@ -8105,6 +8191,9 @@ def _launch_external_cron_worker(job: dict) -> bool: payload_path.unlink(missing_ok=True) raise + with _running_lock: + _restart_safe_waiter_job_ids.add(job_id) + deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: if ack_path.exists(): @@ -8116,25 +8205,36 @@ def _launch_external_cron_worker(job: dict) -> bool: "treating handoff as ownership-uncertain", execution_id, ) - return True + return _wait_for_external_cron_worker( + process, execution_id=execution_id, job_id=job_id + ) finally: ack_path.unlink(missing_ok=True) - if acknowledgement.get("execution_id") != execution_id: + if ( + not isinstance(acknowledgement, dict) + or acknowledgement.get("execution_id") != execution_id + ): logger.error( "Cron external worker acknowledgement mismatch for %s; " "treating handoff as ownership-uncertain", execution_id, ) - return True + return _wait_for_external_cron_worker( + process, execution_id=execution_id, job_id=job_id + ) logger.info( "Cron job '%s' handed to restart-safe worker pid=%s execution=%s", job_id, acknowledgement.get("pid"), execution_id, ) - return True + return _wait_for_external_cron_worker( + process, execution_id=execution_id, job_id=job_id + ) returncode = process.poll() if returncode is not None: + with _running_lock: + _restart_safe_waiter_job_ids.discard(job_id) payload_path.unlink(missing_ok=True) raise RuntimeError( f"cron external worker exited before ownership acknowledgement " @@ -8151,7 +8251,9 @@ def _launch_external_cron_worker(job: dict) -> bool: "leaving the durable execution claim untouched", job_id, ) - return True + return _wait_for_external_cron_worker( + process, execution_id=execution_id, job_id=job_id + ) def _run_external_worker_payload(payload_path: Path, ack_path: Path) -> bool: diff --git a/gateway/run.py b/gateway/run.py index 27f88ea75f17..e17ab9eddf83 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -33280,13 +33280,13 @@ def _drain_restart_safe_cron_deliveries(adapters, loop, runner=None) -> None: """Drain each profile's worker queue through its matching live adapters.""" from cron import scheduler as cron_scheduler - if adapters: + if adapters is not None: cron_scheduler.drain_delivery_queue(adapters, loop) if runner is None: return for profile_name, profile_home in _handoff_watch_scopes(runner)[1:]: profile_adapters = getattr(runner, "_profile_adapters", {}).get(profile_name) - if not profile_adapters: + if profile_adapters is None: continue with _profile_runtime_scope(profile_home): cron_scheduler.drain_delivery_queue(profile_adapters, loop) @@ -33358,10 +33358,7 @@ def _start_gateway_housekeeping( # their final send for whichever gateway instance is live. Drain on # the gateway-wide housekeeper rather than the built-in scheduler tick: # external providers do not run that ticker. - profile_adapters = ( - getattr(runner, "_profile_adapters", {}) if runner is not None else {} - ) - if adapters or any(profile_adapters.values()): + if adapters is not None or runner is not None: try: _drain_restart_safe_cron_deliveries(adapters, loop, runner) except Exception as exc: diff --git a/tests/cron/test_cron_failure_deliver.py b/tests/cron/test_cron_failure_deliver.py index 1695447ad0d4..88e6fee2441f 100644 --- a/tests/cron/test_cron_failure_deliver.py +++ b/tests/cron/test_cron_failure_deliver.py @@ -76,7 +76,7 @@ async def fake_sender(pconfig, chat_id, message, *, thread_id=None, monkeypatch.setattr(s, "create_execution", lambda *_a, **_kw: {"id": "exec-t"}) monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True) - monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr(s, "mark_execution_running", lambda _execution_id: {}) monkeypatch.setattr( s, "save_job_output", lambda jid, out: state["saved"].append(jid) or f"/tmp/{jid}.txt", diff --git a/tests/cron/test_delivery_queue.py b/tests/cron/test_delivery_queue.py index 7fc65171585a..67539a6ba3c3 100644 --- a/tests/cron/test_delivery_queue.py +++ b/tests/cron/test_delivery_queue.py @@ -2,6 +2,7 @@ from __future__ import annotations +import sqlite3 from unittest.mock import Mock import pytest @@ -16,8 +17,79 @@ def test_pending_delivery_is_claimed_and_sent_once(tmp_path, monkeypatch): assert queue.drain(send) == 1 assert queue.drain(send) == 0 - send.assert_called_once_with({"id": "job-1"}, "brief") - assert queue.get_status("exec-1")["status"] == "delivered" + send.assert_called_once_with({"id": "job-1"}, "brief", False) + status = queue.get_status("exec-1") + assert status["status"] == "delivered" + assert status["job_json"] == "{}" + assert status["content"] == "" + + +def test_terminal_delivery_retention_is_bounded(tmp_path, monkeypatch): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + monkeypatch.setattr(queue, "MAX_TERMINAL_DELIVERIES", 2, raising=False) + for index in range(4): + execution_id = f"exec-{index}" + queue.enqueue(execution_id, {"id": f"job-{index}"}, f"brief-{index}") + assert queue.claim_next()["execution_id"] == execution_id + assert queue._finish(execution_id, error=None) + + assert queue.get_status("exec-0") is None + assert queue.get_status("exec-1") is None + assert queue.get_status("exec-2")["status"] == "delivered" + assert queue.get_status("exec-3")["status"] == "delivered" + + +def test_failure_delivery_lane_survives_durable_handoff(tmp_path, monkeypatch): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue( + "exec-failure", + {"id": "job-failure", "failure_deliver": "local"}, + "failed", + for_failure=True, + ) + send = Mock(return_value=None) + + assert queue.drain(send) == 1 + send.assert_called_once_with( + {"id": "job-failure", "failure_deliver": "local"}, + "failed", + True, + ) + + +def test_legacy_queue_schema_adds_failure_lane_before_enqueue(tmp_path, monkeypatch): + import cron.delivery_queue as queue + + db = tmp_path / "deliveries.db" + with sqlite3.connect(db) as conn: + conn.execute( + """CREATE TABLE deliveries ( + execution_id TEXT PRIMARY KEY, + job_json TEXT NOT NULL, + content TEXT NOT NULL, + status TEXT NOT NULL, + owner_process_id TEXT, + owner_pid INTEGER, + owner_started_at INTEGER, + created_at TEXT NOT NULL, + finished_at TEXT, + error TEXT + )""" + ) + monkeypatch.setattr(queue, "DELIVERY_DB", db) + + queue.enqueue( + "exec-migrated", + {"id": "job-migrated"}, + "failed", + for_failure=True, + ) + + assert queue.get_status("exec-migrated")["for_failure"] == 1 def test_dead_delivery_owner_becomes_unknown_and_is_not_retried( diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index f2b8a9cacf78..641be1fdf9d5 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -39,6 +39,54 @@ def test_execution_transitions_are_durable(monkeypatch, tmp_path): assert persisted == [completed] +def test_execution_can_be_loaded_by_exact_attempt_id(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + first = executions.create_execution("same-job", source="builtin") + second = executions.create_execution("same-job", source="builtin") + + assert executions.get_execution(first["id"]) == first + assert executions.get_execution(second["id"]) == second + assert executions.get_execution("missing") is None + + +def test_fresh_external_handoff_is_not_recovered_before_worker_adopts( + monkeypatch, tmp_path +): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("handoff-job", source="builtin") + assert executions.mark_execution_handoff_pending(record["id"]) is not None + + monkeypatch.setattr(executions, "_PROCESS_ID", "replacement-gateway") + monkeypatch.setattr(executions, "_owner_is_live", lambda _pid, _started: False) + + assert executions.recover_interrupted_executions() == 0 + assert executions.get_execution(record["id"])["status"] == "claimed" + adopted = executions.adopt_claimed_execution(record["id"]) + assert adopted["status"] == "running" + assert adopted["handoff_pending"] == 0 + + +def test_stale_external_handoff_is_recovered_unknown(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("handoff-job", source="builtin") + pending = executions.mark_execution_handoff_pending(record["id"]) + + monkeypatch.setattr(executions, "_PROCESS_ID", "replacement-gateway") + monkeypatch.setattr(executions, "_owner_is_live", lambda _pid, _started: False) + monkeypatch.setattr( + executions.time, + "time", + lambda: pending["handoff_started_at"] + + executions.HANDOFF_ADOPTION_GRACE_SECONDS + + 1, + ) + + assert executions.recover_interrupted_executions() == 1 + recovered = executions.get_execution(record["id"]) + assert recovered["status"] == "unknown" + assert recovered["handoff_pending"] == 0 + + def test_execution_ledger_follows_the_current_profile_home(monkeypatch, tmp_path): import cron.executions as executions diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index e0c0207f41ab..8e1a2135d232 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -2,7 +2,14 @@ from __future__ import annotations +import asyncio import json +import os +import signal +import subprocess +import sys +import threading +import time from pathlib import Path from unittest.mock import Mock @@ -34,6 +41,29 @@ def test_execution_owner_moves_to_external_worker_before_running( assert execution_ledger.mark_execution_running(record["id"]) is None +def test_genuine_external_worker_crash_is_recovered_unknown( + execution_ledger, monkeypatch +): + record = execution_ledger.create_execution("job-crash", source="builtin") + script = ( + "import os\n" + "from pathlib import Path\n" + "import cron.executions as executions\n" + f"executions.EXECUTIONS_FILE = Path({str(execution_ledger.EXECUTIONS_FILE)!r})\n" + f"assert executions.adopt_claimed_execution({record['id']!r}) is not None\n" + "os._exit(9)\n" + ) + + crashed = subprocess.run([sys.executable, "-c", script], check=False) + assert crashed.returncode == 9 + + monkeypatch.setattr(execution_ledger, "_PROCESS_ID", "replacement-scheduler") + assert execution_ledger.recover_interrupted_executions() == 1 + recovered = execution_ledger.latest_execution("job-crash") + assert recovered["status"] == "unknown" + assert "whether side effects ran is unknown" in recovered["error"] + + def test_restart_safe_gateway_child_fails_closed_without_scope(monkeypatch): import tools.process_registry as process_registry @@ -159,7 +189,17 @@ def popen(command, **kwargs): ) return FakeProcess() + handoff = Mock(return_value={"id": "exec-1", "handoff_pending": 1}) + monkeypatch.setattr(scheduler, "mark_execution_handoff_pending", handoff) monkeypatch.setattr(scheduler.subprocess, "Popen", popen) + observed_statuses = iter( + [ + {"id": "exec-1", "status": "running"}, + {"id": "exec-1", "status": "completed"}, + ] + ) + get = Mock(side_effect=lambda _execution_id: next(observed_statuses)) + monkeypatch.setattr(scheduler, "get_execution", get) monkeypatch.setenv("ANTHROPIC_API_KEY", "should-not-cross-profile") from agent.secret_scope import set_multiplex_active @@ -172,10 +212,58 @@ def popen(command, **kwargs): assert spawned[0][0][0:2] == ["scope", "--"] assert spawned[0][1]["start_new_session"] is True assert "ANTHROPIC_API_KEY" not in spawned[0][1]["env"] + handoff.assert_called_once_with("exec-1") + assert get.call_count == 2 payload = json.loads((tmp_path / "cron/external-workers/exec-1.json").read_text()) assert payload["multiplex_active"] is True +def test_external_worker_exit_rechecks_exact_execution_before_failure(monkeypatch): + import cron.scheduler as scheduler + + statuses = iter( + [ + {"id": "exec-1", "status": "running"}, + {"id": "exec-1", "status": "completed"}, + ] + ) + get = Mock(side_effect=lambda _execution_id: next(statuses)) + monkeypatch.setattr(scheduler, "get_execution", get, raising=False) + process = Mock() + process.poll.return_value = 0 + + assert scheduler._wait_for_external_cron_worker( + process, execution_id="exec-1" + ) is True + assert get.call_count == 2 + + +def test_external_worker_crash_recovers_uncertain_attempt(monkeypatch): + import cron.scheduler as scheduler + + statuses = iter( + [ + {"id": "exec-1", "status": "running"}, + {"id": "exec-1", "status": "running"}, + {"id": "exec-1", "status": "unknown"}, + ] + ) + get = Mock(side_effect=lambda _execution_id: next(statuses)) + recover = Mock(return_value=1) + monkeypatch.setattr(scheduler, "get_execution", get) + monkeypatch.setattr( + scheduler, "recover_interrupted_executions", recover, raising=False + ) + process = Mock() + process.poll.return_value = 9 + + assert scheduler._wait_for_external_cron_worker( + process, execution_id="exec-1" + ) is True + recover.assert_called_once_with() + assert get.call_count == 3 + + def test_launch_external_worker_stays_in_process_outside_managed_gateway( monkeypatch, ): @@ -215,6 +303,40 @@ def test_shared_run_path_hands_gateway_fire_to_external_worker(monkeypatch): run.assert_not_called() +def test_shutdown_does_not_interrupt_restart_safe_waiter(): + import cron.scheduler as scheduler + + job_id = "external-waiter" + scheduler._running_job_ids.add(job_id) + scheduler._restart_safe_waiter_job_ids.add(job_id) + try: + assert scheduler.mark_running_jobs_interrupted("gateway restart") == [] + assert job_id not in scheduler._interrupted_job_ids + finally: + scheduler._restart_safe_waiter_job_ids.discard(job_id) + scheduler._running_job_ids.discard(job_id) + scheduler._interrupted_job_ids.discard(job_id) + + +def test_gateway_tool_run_without_adapter_objects_hands_off(monkeypatch): + import cron.scheduler as scheduler + + created = Mock(return_value={"id": "exec-tool"}) + launch = Mock(return_value=True) + run = Mock(side_effect=AssertionError("agent ran inside gateway")) + monkeypatch.setattr(scheduler, "create_execution", created) + monkeypatch.setattr(scheduler, "_launch_external_cron_worker", launch) + monkeypatch.setattr(scheduler, "run_job", run) + job = {"id": "tool-job"} + + assert scheduler.run_one_job(job, adapters=None) is True + + created.assert_called_once_with("tool-job", source="direct") + assert job["execution_id"] == "exec-tool" + launch.assert_called_once_with(job) + run.assert_not_called() + + def test_shared_run_path_creates_execution_before_managed_handoff(monkeypatch): import cron.scheduler as scheduler @@ -243,3 +365,150 @@ def test_lost_execution_start_cas_prevents_side_effects(monkeypatch): {"id": "job-1", "execution_id": "exec-1"}, adapters=None ) is True run.assert_not_called() + + +@pytest.mark.linux_only +@pytest.mark.live_system_guard_bypass +def test_managed_gateway_restart_preserves_active_worker_and_single_side_effect( + tmp_path, monkeypatch +): + import cron.delivery_queue as delivery_queue + import cron.executions as executions + import cron.scheduler as scheduler + from cron.jobs import create_job, use_cron_store + from gateway.config import Platform, PlatformConfig + from gateway.status import _pid_exists + from tools import process_registry + + if not process_registry._systemd_run_user_scope_available(): + pytest.skip("systemd-run --user --scope is unavailable on this host") + + home = tmp_path / "profile" + scripts_dir = home / "scripts" + scripts_dir.mkdir(parents=True) + started = tmp_path / "started" + release = tmp_path / "release" + side_effect = tmp_path / "side-effect" + probe = scripts_dir / "restart_probe.py" + probe.write_text( + "import pathlib, time\n" + f"started = pathlib.Path({str(started)!r})\n" + f"release = pathlib.Path({str(release)!r})\n" + f"side_effect = pathlib.Path({str(side_effect)!r})\n" + "started.write_text('started')\n" + "deadline = time.monotonic() + 15\n" + "while not release.exists() and time.monotonic() < deadline:\n" + " time.sleep(0.05)\n" + "if not release.exists():\n" + " raise SystemExit('release timeout')\n" + "with side_effect.open('a') as handle:\n" + " handle.write('once\\n')\n" + "print('completed')\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + with use_cron_store(home): + job = create_job( + prompt=None, + schedule="every 1h", + name="restart probe", + script=probe.name, + no_agent=True, + deliver="telegram:123", + ) + payload = tmp_path / "job.json" + launched = tmp_path / "launched.json" + payload.write_text(json.dumps(job), encoding="utf-8") + + sent = [] + adapter = Mock() + + async def send(_chat_id, content, metadata=None): + sent.append((content, metadata)) + return {"success": True, "message_id": "restart-delivery-1"} + + adapter.send = send + gateway_config = Mock() + gateway_config.platforms = { + Platform.TELEGRAM: PlatformConfig(enabled=True), + } + gateway_config.get_home_channel = lambda _platform: None + monkeypatch.setattr( + "gateway.config.load_gateway_config", lambda: gateway_config + ) + monkeypatch.setattr( + scheduler, "load_config", lambda: {"cron": {"wrap_response": False}} + ) + replacement_loop = asyncio.new_event_loop() + replacement_thread = threading.Thread( + target=replacement_loop.run_forever, + daemon=True, + ) + replacement_thread.start() + deadline = time.monotonic() + 2 + while not replacement_loop.is_running() and time.monotonic() < deadline: + time.sleep(0.01) + assert replacement_loop.is_running() + + harness = ( + "import json, os, pathlib, time\n" + f"os.environ['HERMES_HOME'] = {str(home)!r}\n" + "os.environ['INVOCATION_ID'] = 'restart-fixture'\n" + "from cron import scheduler\n" + "from tools import process_registry\n" + "process_registry._is_supervised_gateway_process = lambda: True\n" + f"job = json.loads(pathlib.Path({str(payload)!r}).read_text())\n" + "if not scheduler.run_one_job(job, adapters=None, loop=None):\n" + " raise SystemExit('worker was not isolated')\n" + f"pathlib.Path({str(launched)!r}).write_text('returned')\n" + ) + parent = subprocess.Popen([sys.executable, "-c", harness]) + worker_pid = None + try: + deadline = time.monotonic() + 10 + current = None + while time.monotonic() < deadline: + if parent.poll() is not None: + pytest.fail(f"gateway fixture exited early with {parent.returncode}") + current = executions.latest_execution(job["id"]) + if started.exists() and current and current.get("pid") != os.getpid(): + break + time.sleep(0.05) + assert started.exists() + assert current is not None + execution = current + worker_pid = int(current["pid"]) + assert not launched.exists(), "handoff returned before execution completed" + + # Replacing a managed gateway kills its old process tree. The active + # cron owner must remain in its transient scope and keep the same PID. + parent.terminate() + parent.wait(timeout=5) + assert _pid_exists(worker_pid) + + release.write_text("go", encoding="utf-8") + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + row = delivery_queue.get_status(execution["id"]) + if row and row["status"] == "pending": + scheduler.drain_delivery_queue( + {Platform.TELEGRAM: adapter}, replacement_loop + ) + current = executions.latest_execution(job["id"]) + if current and current["status"] == "completed": + break + time.sleep(0.05) + assert executions.latest_execution(job["id"])["status"] == "completed" + assert side_effect.read_text(encoding="utf-8").splitlines() == ["once"] + assert delivery_queue.get_status(execution["id"])["status"] == "delivered" + assert len(sent) == 1 + assert "completed" in sent[0][0] + finally: + replacement_loop.call_soon_threadsafe(replacement_loop.stop) + replacement_thread.join(timeout=2) + replacement_loop.close() + if parent.poll() is None: + parent.terminate() + parent.wait(timeout=5) + if worker_pid is not None and _pid_exists(worker_pid): + os.kill(worker_pid, signal.SIGKILL) diff --git a/tests/gateway/test_cron_delivery_housekeeping.py b/tests/gateway/test_cron_delivery_housekeeping.py index 8e38071679c9..9260c031fbe1 100644 --- a/tests/gateway/test_cron_delivery_housekeeping.py +++ b/tests/gateway/test_cron_delivery_housekeeping.py @@ -37,11 +37,29 @@ def test_gateway_housekeeping_drains_cron_delivery_with_live_adapters(monkeypatc assert calls == [(adapters, loop)] +def test_gateway_housekeeping_drains_cron_delivery_without_connected_adapters(monkeypatch): + adapters = {} + loop = object() + calls = [] + monkeypatch.setattr( + scheduler, + "drain_delivery_queue", + lambda live_adapters, live_loop: calls.append((live_adapters, live_loop)), + raising=False, + ) + + gateway_run._start_gateway_housekeeping( + _OneTickStopEvent(), adapters=adapters, loop=loop, interval=0 + ) + + assert calls == [(adapters, loop)] + + def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( tmp_path, monkeypatch ): root_adapters = {} - secondary_adapters = {"telegram": "secondary"} + secondary_adapters = {} runner = SimpleNamespace( config=SimpleNamespace(multiplex_profiles=True), adapters=root_adapters, @@ -77,6 +95,7 @@ def fake_scope(home): ) assert calls == [ + ("drain", root_adapters), ("scope", secondary_home), ("drain", secondary_adapters), ] diff --git a/tests/tools/test_cronjob_run_background.py b/tests/tools/test_cronjob_run_background.py index a35d4c4a8fbc..c120273c03e9 100644 --- a/tests/tools/test_cronjob_run_background.py +++ b/tests/tools/test_cronjob_run_background.py @@ -237,6 +237,28 @@ def probe_run(job, **kw): assert seen_during_run["registered"] is True assert "job-bg-09" not in sched.get_running_job_ids() # released after + def test_run_claimed_job_reports_exact_unknown_execution_not_stale_success(self): + from tools.cronjob_tools import _run_claimed_job + + def probe_run(job, **_kwargs): + job["execution_id"] = "exec-unknown" + return True + + with patch("cron.scheduler.run_one_job", side_effect=probe_run), \ + patch("cron.executions.get_execution", return_value={ + "id": "exec-unknown", + "status": "unknown", + "error": "worker owner exited", + }), \ + patch("tools.cronjob_tools.get_job", return_value={ + "last_status": "ok", + "last_error": None, + }): + res = _run_claimed_job(_job("job-bg-unknown")) + + assert res["success"] is False + assert res["error"] == "worker owner exited" + def test_background_dispatch_reports_running_job_immediately(self): """The dispatch path pre-checks the running set so a mid-run job reports in the tool response, not as a delayed completion event.""" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index cdcdc6cea5c4..e59f364979f2 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -1125,6 +1125,12 @@ def _heartbeat_loop() -> None: _registered = False release_running_job(job_id) refreshed = get_job(job_id) or {} + execution = None + execution_id = job.get("execution_id") + if execution_id: + from cron.executions import get_execution + + execution = get_execution(str(execution_id)) last_status = refreshed.get("last_status") # "delivery_failed" (#83993): the agent run itself succeeded but the # output never reached the user. That is NOT a success for the caller @@ -1136,6 +1142,12 @@ def _heartbeat_loop() -> None: run_error = refreshed.get("last_error") if last_status == "delivery_failed" and not run_error: run_error = refreshed.get("last_delivery_error") + if execution is not None and execution.get("status") != "completed": + ok = False + run_error = ( + execution.get("error") + or f"execution ended in {execution.get('status') or 'unknown'} state" + ) return { "claimed": True, "success": bool(processed and ok), From bc1ce7dc792f011880641636206bdf835dad6358 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Thu, 3 Sep 2026 12:03:58 +0700 Subject: [PATCH 3/5] [verified] fix(cron): close restart handoff races --- cron/delivery_queue.py | 102 ++++++++++++++++-- cron/executions.py | 23 ++-- gateway/run.py | 22 +++- tests/cron/test_delivery_queue.py | 56 ++++++++-- tests/cron/test_execution_ledger.py | 66 ++++++++++++ tests/cron/test_restart_safe_worker.py | 31 +++++- .../test_cron_delivery_housekeeping.py | 58 +++++++++- tools/process_registry.py | 2 +- 8 files changed, 329 insertions(+), 31 deletions(-) diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index 99757f7cdff5..013bef5048cd 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -19,6 +19,7 @@ from pathlib import Path from typing import Any, Callable, Iterator, Optional +from agent.redact import redact_sensitive_text from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -28,6 +29,7 @@ _ACTIVE_DELIVERIES: set[str] = set() _TERMINAL = ("delivered", "failed", "unknown") MAX_TERMINAL_DELIVERIES = 1000 +DEFAULT_DELIVERY_WAIT_TIMEOUT_SECONDS = 300.0 def _prune_terminal_unlocked(conn: sqlite3.Connection) -> None: @@ -46,6 +48,15 @@ def _prune_terminal_unlocked(conn: sqlite3.Connection) -> None: ) excess = terminal_count - keep if excess > 0: + conn.execute( + """INSERT OR IGNORE INTO delivery_tombstones + (execution_id, terminal_status, finished_at) + SELECT execution_id, status, finished_at FROM deliveries + WHERE status IN ('delivered','failed','unknown') + ORDER BY finished_at, created_at, execution_id + LIMIT ?""", + (excess,), + ) conn.execute( """DELETE FROM deliveries WHERE execution_id IN ( SELECT execution_id FROM deliveries @@ -92,6 +103,14 @@ def _transaction() -> Iterator[sqlite3.Connection]: error TEXT )""" ) + conn.execute( + """CREATE TABLE IF NOT EXISTS delivery_tombstones ( + execution_id TEXT PRIMARY KEY, + terminal_status TEXT NOT NULL CHECK(terminal_status IN + ('delivered','failed','unknown')), + finished_at TEXT + )""" + ) columns = { str(row[1]) for row in conn.execute("PRAGMA table_info(deliveries)") } @@ -138,6 +157,17 @@ def enqueue( ) -> dict: """Persist one idempotent delivery request before the worker waits.""" with _transaction() as conn: + tombstone = conn.execute( + "SELECT terminal_status, finished_at FROM delivery_tombstones " + "WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + if tombstone is not None: + return { + "execution_id": str(execution_id), + "status": tombstone["terminal_status"], + "finished_at": tombstone["finished_at"], + } conn.execute( """INSERT OR IGNORE INTO deliveries (execution_id, job_json, content, for_failure, status, created_at) @@ -161,7 +191,21 @@ def get_status(execution_id: str) -> Optional[dict]: row = conn.execute( "SELECT * FROM deliveries WHERE execution_id=?", (str(execution_id),) ).fetchone() - return dict(row) if row is not None else None + if row is not None: + return dict(row) + tombstone = conn.execute( + "SELECT execution_id, terminal_status, finished_at " + "FROM delivery_tombstones WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + if tombstone is None: + return None + return { + "execution_id": tombstone["execution_id"], + "status": tombstone["terminal_status"], + "finished_at": tombstone["finished_at"], + "error": None, + } def claim_next() -> Optional[dict]: @@ -194,6 +238,11 @@ def claim_next() -> Optional[dict]: def _finish(execution_id: str, *, error: Optional[str]) -> bool: status = "failed" if error else "delivered" + safe_error = ( + redact_sensitive_text(str(error), force=True, redact_url_credentials=True) + if error + else None + ) with _transaction() as conn: cur = conn.execute( """UPDATE deliveries SET status=?, finished_at=?, error=? @@ -202,7 +251,7 @@ def _finish(execution_id: str, *, error: Optional[str]) -> bool: ( status, _hermes_now().isoformat(), - error, + safe_error, execution_id, _PROCESS_ID, os.getpid(), @@ -275,6 +324,40 @@ def drain( return processed +def _terminalize_wait_timeout(execution_id: str) -> str: + """Fence a delivery whose worker can no longer wait for confirmation.""" + now = _hermes_now().isoformat() + pending_error = "timed out waiting for a live gateway; delivery was not attempted" + uncertain_error = ( + "timed out while gateway delivery was in progress; outcome is unknown and " + "was not retried" + ) + with _transaction() as conn: + cur = conn.execute( + """UPDATE deliveries SET status='failed', finished_at=?, error=? + WHERE execution_id=? AND status='pending'""", + (now, pending_error, str(execution_id)), + ) + if cur.rowcount: + _prune_terminal_unlocked(conn) + return pending_error + conn.execute( + """UPDATE deliveries SET status='unknown', finished_at=?, error=? + WHERE execution_id=? AND status='delivering'""", + (now, uncertain_error, str(execution_id)), + ) + row = conn.execute( + "SELECT status, error FROM deliveries WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + _prune_terminal_unlocked(conn) + if row is None: + return "timed out waiting for live gateway delivery" + if row["status"] == "delivered": + return "" + return str(row["error"] or f"delivery {row['status']}") + + def enqueue_and_wait( execution_id: str, job: dict, @@ -284,13 +367,20 @@ def enqueue_and_wait( timeout: Optional[float] = None, ) -> Optional[str]: """Queue delivery and wait for a gateway's terminal at-most-once outcome.""" - enqueue(execution_id, job, content, for_failure=for_failure) - deadline = None if timeout is None else time.monotonic() + timeout - while deadline is None or time.monotonic() < deadline: + queued = enqueue(execution_id, job, content, for_failure=for_failure) + if queued["status"] in _TERMINAL: + return None if queued["status"] == "delivered" else str( + queued.get("error") or f"delivery {queued['status']}" + ) + wait_timeout = ( + DEFAULT_DELIVERY_WAIT_TIMEOUT_SECONDS if timeout is None else max(0.0, timeout) + ) + deadline = time.monotonic() + wait_timeout + while time.monotonic() < deadline: row = get_status(execution_id) if row and row["status"] in _TERMINAL: return None if row["status"] == "delivered" else str( row.get("error") or f"delivery {row['status']}" ) time.sleep(0.25) - return "timed out waiting for live gateway delivery; request remains pending" + return _terminalize_wait_timeout(execution_id) or None diff --git a/cron/executions.py b/cron/executions.py index 9cf204e559d9..3abd5f54df1a 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -148,7 +148,7 @@ def _prune_unlocked(conn: sqlite3.Connection) -> None: """DELETE FROM executions WHERE id IN ( SELECT id FROM executions WHERE status IN ('completed','failed','unknown') - ORDER BY claimed_at DESC, id DESC LIMIT -1 OFFSET ? + ORDER BY finished_at DESC, claimed_at DESC, id DESC LIMIT -1 OFFSET ? )""", (limit,), ) @@ -211,7 +211,7 @@ def adopt_claimed_execution(execution_id: str) -> Optional[Dict[str, Any]]: SET process_id=?, pid=?, process_started_at=?, status='running', started_at=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status='claimed'""", + WHERE id=? AND status='claimed' AND handoff_pending=1""", (_PROCESS_ID, pid, process_started_at, now, execution_id), ) if cur.rowcount != 1: @@ -231,8 +231,9 @@ def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: """UPDATE executions SET status='running', started_at=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status='claimed' AND handoff_pending=0""", - (now, execution_id), + WHERE id=? AND status='claimed' AND handoff_pending=0 + AND process_id=? AND pid=?""", + (now, execution_id, _PROCESS_ID, os.getpid()), ) if cur.rowcount != 1: return None @@ -256,8 +257,9 @@ def finish_execution( """UPDATE executions SET status=?, finished_at=?, error=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status IN ('claimed','running')""", - (status, now, detail, execution_id), + WHERE id=? AND status IN ('claimed','running') + AND process_id=? AND pid=?""", + (status, now, detail, execution_id, _PROCESS_ID, os.getpid()), ) if cur.rowcount != 1: return None @@ -276,7 +278,7 @@ def recover_interrupted_executions() -> int: recovered: List[Dict[str, Any]] = [] with _transaction() as conn: rows = conn.execute( - """SELECT id, process_id, pid, process_started_at, + """SELECT id, status, process_id, pid, process_started_at, handoff_pending, handoff_started_at FROM executions WHERE status IN ('claimed','running')""" @@ -298,11 +300,14 @@ def recover_interrupted_executions() -> int: """UPDATE executions SET status='unknown', finished_at=?, error=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status IN ('claimed','running')""", + WHERE id=? AND status=? AND process_id=? AND pid=? + AND handoff_pending=? + AND handoff_started_at IS ?""", (now, "Scheduler restarted after this execution's owner exited before a durable " "terminal state; whether side effects ran is unknown.", - row["id"]), + row["id"], row["status"], row["process_id"], row["pid"], + row["handoff_pending"], row["handoff_started_at"]), ) changed += cur.rowcount if cur.rowcount: diff --git a/gateway/run.py b/gateway/run.py index e17ab9eddf83..22b9fa32653e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -33280,15 +33280,27 @@ def _drain_restart_safe_cron_deliveries(adapters, loop, runner=None) -> None: """Drain each profile's worker queue through its matching live adapters.""" from cron import scheduler as cron_scheduler - if adapters is not None: - cron_scheduler.drain_delivery_queue(adapters, loop) if runner is None: + if adapters is not None: + cron_scheduler.drain_delivery_queue(adapters, loop) return - for profile_name, profile_home in _handoff_watch_scopes(runner)[1:]: - profile_adapters = getattr(runner, "_profile_adapters", {}).get(profile_name) + for profile_name, profile_home in _handoff_watch_scopes(runner): + scoped_home = profile_home or get_hermes_home() + if profile_name is None: + profile_adapters = adapters + else: + profile_adapters = getattr(runner, "_profile_adapters", {}).get( + profile_name + ) if profile_adapters is None: continue - with _profile_runtime_scope(profile_home): + with _profile_runtime_scope(scoped_home): + if profile_name is not None and not profile_adapters and adapters: + routes = cron_scheduler._primary_profile_routes_for_current_home() + if routes: + profile_adapters = cron_scheduler.SharedRouteAdapters( + adapters, routes + ) cron_scheduler.drain_delivery_queue(profile_adapters, loop) diff --git a/tests/cron/test_delivery_queue.py b/tests/cron/test_delivery_queue.py index 67539a6ba3c3..b944fda22461 100644 --- a/tests/cron/test_delivery_queue.py +++ b/tests/cron/test_delivery_queue.py @@ -35,11 +35,22 @@ def test_terminal_delivery_retention_is_bounded(tmp_path, monkeypatch): assert queue.claim_next()["execution_id"] == execution_id assert queue._finish(execution_id, error=None) - assert queue.get_status("exec-0") is None - assert queue.get_status("exec-1") is None + # Pruning may discard verbose outcome rows, but never the durable + # idempotency tombstone for an execution that could be replayed later. + pruned = queue.get_status("exec-0") + assert pruned is not None + assert pruned["status"] == "delivered" + assert queue.get_status("exec-1")["status"] == "delivered" assert queue.get_status("exec-2")["status"] == "delivered" assert queue.get_status("exec-3")["status"] == "delivered" + # Pruning may discard verbose outcome rows, but never the durable + # idempotency tombstone for an execution that could be replayed later. + queue.enqueue("exec-0", {"id": "job-replayed"}, "duplicate brief") + send = Mock(return_value=None) + assert queue.drain(send) == 0 + send.assert_not_called() + def test_failure_delivery_lane_survives_durable_handoff(tmp_path, monkeypatch): import cron.delivery_queue as queue @@ -92,6 +103,29 @@ def test_legacy_queue_schema_adds_failure_lane_before_enqueue(tmp_path, monkeypa assert queue.get_status("exec-migrated")["for_failure"] == 1 +def test_wait_timeout_marks_inflight_delivery_unknown_without_retry( + tmp_path, monkeypatch +): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue("exec-inflight", {"id": "job-inflight"}, "result") + assert queue.claim_next() is not None + + error = queue.enqueue_and_wait( + "exec-inflight", {"id": "job-inflight"}, "result", timeout=0 + ) + + assert error is not None + assert "outcome is unknown" in error + status = queue.get_status("exec-inflight") + assert status is not None + assert status["status"] == "unknown" + send = Mock(return_value=None) + assert queue.drain(send) == 0 + send.assert_not_called() + + def test_dead_delivery_owner_becomes_unknown_and_is_not_retried( tmp_path, monkeypatch ): @@ -110,19 +144,23 @@ def test_dead_delivery_owner_becomes_unknown_and_is_not_retried( assert queue.get_status("exec-1")["status"] == "unknown" -def test_delivery_failure_is_terminal_and_not_retried(tmp_path, monkeypatch): +def test_delivery_failure_is_terminal_not_retried_and_redacted( + tmp_path, monkeypatch +): import cron.delivery_queue as queue monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") queue.enqueue("exec-1", {"id": "job-1"}, "brief") - send = Mock(return_value="transport failed") + send = Mock(return_value="request failed: https://example.test/?token=TOKEN123") assert queue.drain(send) == 1 assert queue.drain(send) == 0 assert send.call_count == 1 status = queue.get_status("exec-1") + assert status is not None assert status["status"] == "failed" - assert status["error"] == "transport failed" + assert "TOKEN123" not in status["error"] + assert "token=***" in status["error"] def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): @@ -134,10 +172,12 @@ def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): error = queue.enqueue_and_wait("exec-3", job, "result", timeout=0) assert "timed out" in error - assert queue.get_status("exec-3")["status"] == "pending" + status = queue.get_status("exec-3") + assert status is not None + assert status["status"] == "failed" send = Mock(return_value=None) - assert queue.drain(send) == 1 - send.assert_called_once() + assert queue.drain(send) == 0 + send.assert_not_called() def test_same_gateway_recovers_terminalization_failure_without_resending( diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 641be1fdf9d5..d68259ae4b4f 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -87,6 +87,54 @@ def test_stale_external_handoff_is_recovered_unknown(monkeypatch, tmp_path): assert recovered["handoff_pending"] == 0 +def test_recovery_does_not_overwrite_concurrent_worker_adoption(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("adoption-race", source="builtin") + pending = executions.mark_execution_handoff_pending(record["id"]) + assert pending is not None + monkeypatch.setattr(executions, "_PROCESS_ID", "replacement-scheduler") + monkeypatch.setattr( + executions.time, + "time", + lambda: pending["handoff_started_at"] + + executions.HANDOFF_ADOPTION_GRACE_SECONDS + + 1, + ) + + def adopt_while_liveness_is_checked(_pid, _started_at): + monkeypatch.setattr(executions, "_PROCESS_ID", "external-worker") + monkeypatch.setattr(executions.os, "getpid", lambda: 4242) + monkeypatch.setattr(executions, "_process_start_time", lambda _pid: 9876) + assert executions.adopt_claimed_execution(record["id"]) is not None + return False + + monkeypatch.setattr(executions, "_owner_is_live", adopt_while_liveness_is_checked) + + assert executions.recover_interrupted_executions() == 0 + current = executions.get_execution(record["id"]) + assert current is not None + assert current["status"] == "running" + assert current["process_id"] == "external-worker" + assert current["pid"] == 4242 + + +def test_foreign_process_cannot_start_or_finish_execution(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("owner-fence", source="builtin") + original_process_id = executions._PROCESS_ID + original_pid = record["pid"] + + monkeypatch.setattr(executions, "_PROCESS_ID", "foreign-process") + monkeypatch.setattr(executions.os, "getpid", lambda: original_pid + 1) + assert executions.mark_execution_running(record["id"]) is None + assert executions.finish_execution(record["id"], success=True) is None + + monkeypatch.setattr(executions, "_PROCESS_ID", original_process_id) + monkeypatch.setattr(executions.os, "getpid", lambda: original_pid) + assert executions.mark_execution_running(record["id"]) is not None + assert executions.finish_execution(record["id"], success=True) is not None + + def test_execution_ledger_follows_the_current_profile_home(monkeypatch, tmp_path): import cron.executions as executions @@ -131,6 +179,24 @@ def test_retention_bounds_terminal_history_but_preserves_inflight(monkeypatch, t assert executions.latest_execution("live")["status"] == "running" +def test_recently_finished_long_running_execution_survives_retention( + monkeypatch, tmp_path +): + executions = _point_ledger(monkeypatch, tmp_path) + monkeypatch.setattr(executions, "MAX_TERMINAL_EXECUTIONS", 1) + long_running = executions.create_execution("long-running", source="builtin") + assert executions.mark_execution_running(long_running["id"]) is not None + newer = executions.create_execution("newer", source="builtin") + assert executions.finish_execution(newer["id"], success=True) is not None + + finished = executions.finish_execution(long_running["id"], success=True) + + assert finished is not None + assert finished["status"] == "completed" + assert executions.get_execution(long_running["id"])["status"] == "completed" + assert executions.get_execution(newer["id"]) is None + + def test_corrupt_store_fails_closed_without_overwrite(monkeypatch, tmp_path): executions = _point_ledger(monkeypatch, tmp_path) executions.EXECUTIONS_FILE.parent.mkdir(parents=True) diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index 8e1a2135d232..ad996fd56de6 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -28,6 +28,7 @@ def test_execution_owner_moves_to_external_worker_before_running( execution_ledger, monkeypatch ): record = execution_ledger.create_execution("job-1", source="builtin") + assert execution_ledger.mark_execution_handoff_pending(record["id"]) is not None monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242) monkeypatch.setattr(execution_ledger, "_process_start_time", lambda pid: 9876) @@ -41,10 +42,22 @@ def test_execution_owner_moves_to_external_worker_before_running( assert execution_ledger.mark_execution_running(record["id"]) is None +def test_external_worker_cannot_adopt_execution_without_handoff_fence( + execution_ledger, monkeypatch +): + record = execution_ledger.create_execution("job-unfenced", source="builtin") + monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242) + monkeypatch.setattr(execution_ledger, "_process_start_time", lambda _pid: 9876) + + assert execution_ledger.adopt_claimed_execution(record["id"]) is None + assert execution_ledger.get_execution(record["id"])["status"] == "claimed" + + def test_genuine_external_worker_crash_is_recovered_unknown( execution_ledger, monkeypatch ): record = execution_ledger.create_execution("job-crash", source="builtin") + assert execution_ledger.mark_execution_handoff_pending(record["id"]) is not None script = ( "import os\n" "from pathlib import Path\n" @@ -64,10 +77,10 @@ def test_genuine_external_worker_crash_is_recovered_unknown( assert "whether side effects ran is unknown" in recovered["error"] +@pytest.mark.linux_only def test_restart_safe_gateway_child_fails_closed_without_scope(monkeypatch): import tools.process_registry as process_registry - monkeypatch.setattr(process_registry, "_IS_WINDOWS", False) monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) monkeypatch.setenv("INVOCATION_ID", "managed-service") monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", lambda: False) @@ -89,6 +102,22 @@ def test_restart_safe_gateway_child_is_unchanged_outside_managed_gateway(monkeyp ) is command +def test_restart_safe_gateway_child_never_probes_systemd_off_linux(monkeypatch): + import tools.process_registry as process_registry + + command = ["python", "worker.py"] + probe = Mock(side_effect=AssertionError("systemd probe ran off Linux")) + monkeypatch.setattr(process_registry, "_IS_LINUX", False) + monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) + monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", probe) + monkeypatch.setenv("INVOCATION_ID", "managed-service") + + assert process_registry.restart_safe_gateway_child_argv( + command, unit_suffix="cron-job-1" + ) is command + probe.assert_not_called() + + def test_external_worker_adopts_execution_and_runs_payload_once( tmp_path, monkeypatch ): diff --git a/tests/gateway/test_cron_delivery_housekeeping.py b/tests/gateway/test_cron_delivery_housekeeping.py index 9260c031fbe1..1228b04772b3 100644 --- a/tests/gateway/test_cron_delivery_housekeeping.py +++ b/tests/gateway/test_cron_delivery_housekeeping.py @@ -55,7 +55,7 @@ def test_gateway_housekeeping_drains_cron_delivery_without_connected_adapters(mo assert calls == [(adapters, loop)] -def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( +def test_multiplex_housekeeping_scopes_primary_and_drains_each_profile( tmp_path, monkeypatch ): root_adapters = {} @@ -65,9 +65,12 @@ def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( adapters=root_adapters, _profile_adapters={"secondary": secondary_adapters}, ) + root_home = tmp_path / "root" secondary_home = tmp_path / "secondary" calls = [] + monkeypatch.setattr(gateway_run, "get_hermes_home", lambda: root_home) + monkeypatch.setattr( gateway_run, "_handoff_watch_scopes", @@ -95,7 +98,60 @@ def fake_scope(home): ) assert calls == [ + ("scope", root_home), ("drain", root_adapters), ("scope", secondary_home), ("drain", secondary_adapters), ] + + +def test_multiplex_housekeeping_uses_primary_routes_for_credentialless_satellite( + tmp_path, monkeypatch +): + root_adapters = {"slack": object()} + secondary_home = tmp_path / "secondary" + runner = SimpleNamespace( + config=SimpleNamespace(multiplex_profiles=True), + adapters=root_adapters, + _profile_adapters={"secondary": {}}, + ) + calls = [] + routed = object() + + monkeypatch.setattr( + gateway_run, + "_handoff_watch_scopes", + lambda _runner: [(None, None), ("secondary", secondary_home)], + ) + + @contextmanager + def fake_scope(_home): + yield + + class FakeSharedRouteAdapters: + def __new__(cls, adapters, routes): + calls.append(("routed", adapters, routes)) + return routed + + monkeypatch.setattr(gateway_run, "_profile_runtime_scope", fake_scope) + monkeypatch.setattr(scheduler, "SharedRouteAdapters", FakeSharedRouteAdapters) + monkeypatch.setattr( + scheduler, + "_primary_profile_routes_for_current_home", + lambda: ["route-to-secondary"], + ) + monkeypatch.setattr( + scheduler, + "drain_delivery_queue", + lambda adapters, _loop: calls.append(("drain", adapters)), + ) + + gateway_run._drain_restart_safe_cron_deliveries( + root_adapters, object(), runner + ) + + assert calls == [ + ("drain", root_adapters), + ("routed", root_adapters, ["route-to-secondary"]), + ("drain", routed), + ] diff --git a/tools/process_registry.py b/tools/process_registry.py index 4ec461df14b2..8c1d5b66c7e0 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -334,7 +334,7 @@ def restart_safe_gateway_child_argv( closed if it cannot be established. Standalone processes, non-systemd supervisors, and non-Linux hosts retain the direct command. """ - if _IS_WINDOWS: + if not _IS_LINUX: return command if not _is_supervised_gateway_process() or not os.environ.get("INVOCATION_ID"): return command From c0d347757aaa75dcbc4d9b5883eb1d56c4b177ff Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 3 Sep 2026 12:14:59 +0530 Subject: [PATCH 4/5] fix(cron): keep unsent worker deliveries queued, cheapen handoff polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the salvaged restart-safe worker (#101877): - delivery_queue: a row still `pending` at the worker's wait timeout was marked `failed` and never drained, so any gateway outage longer than the 300s budget (e.g. a restart that runs `hermes update`) silently lost the delivery. Unclaimed rows are certainly unsent, not uncertain — leave them queued for the next gateway; only mid-send rows are fenced `unknown`. - delivery_queue: stop running the full-table prune UPDATE+COUNT inside every transaction (each `get_status` poll paid for it; terminalizing paths already prune explicitly); poll at 1s instead of 250ms. - delivery_queue/executions: use `hermes_state.apply_wal_with_fallback` (bare `journal_mode=WAL` raises on NFS/SMB homes) and the race-safe `hermes_cli.sqlite_util.add_column_if_missing`; drop the copied owner-liveness helpers in favour of the ones in cron.executions. - scheduler: the parent waited on the worker by re-opening the executions ledger every 50ms for the whole run (~20 opens/s, hours). Wait on the process with a 1s timeout instead — the worker commits its terminal row before exiting — and reap stranded payload/ack files once terminal. - scheduler: skip the housekeeping drain until a worker has actually created deliveries.db, so non-systemd gateways never open it. - scheduler: set up hermes logging in the detached worker entrypoint; it runs with stdout/stderr on DEVNULL and previously logged nowhere. - tests: test_lost_fire_claim_stops_stale_delivery still mocked `mark_execution_running -> None`, which now means "ownership lost, return before run_job" — the test passed without ever reaching the path it names. Mocking `{}` restores it (mutation-checked). --- cron/delivery_queue.py | 71 ++++++++----------- cron/executions.py | 21 +++--- cron/scheduler.py | 86 ++++++++++++++++------- tests/cron/test_delivery_queue.py | 15 ++-- tests/cron/test_restart_safe_worker.py | 19 +++-- tests/cron/test_script_claim_heartbeat.py | 2 +- 6 files changed, 126 insertions(+), 88 deletions(-) diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index 013bef5048cd..f03413b72714 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -20,6 +20,8 @@ from typing import Any, Callable, Iterator, Optional from agent.redact import redact_sensitive_text +from cron.executions import _owner_is_live, _process_start_time +from hermes_cli.sqlite_util import add_column_if_missing from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -84,8 +86,10 @@ def _transaction() -> Iterator[sqlite3.Connection]: pass conn.row_factory = sqlite3.Row try: + from hermes_state import apply_wal_with_fallback + conn.execute("PRAGMA busy_timeout=5000") - conn.execute("PRAGMA journal_mode=WAL") + apply_wal_with_fallback(conn, db_label="cron/deliveries.db") conn.execute("PRAGMA synchronous=FULL") conn.execute( """CREATE TABLE IF NOT EXISTS deliveries ( @@ -111,43 +115,19 @@ def _transaction() -> Iterator[sqlite3.Connection]: finished_at TEXT )""" ) - columns = { - str(row[1]) for row in conn.execute("PRAGMA table_info(deliveries)") - } - if "for_failure" not in columns: - conn.execute( - "ALTER TABLE deliveries " - "ADD COLUMN for_failure INTEGER NOT NULL DEFAULT 0" - ) + add_column_if_missing( + conn, "deliveries", "for_failure", + "for_failure INTEGER NOT NULL DEFAULT 0", + ) + # Pruning is done explicitly by the paths that create terminal + # rows (_finish / recover_abandoned / _terminalize_wait_timeout); + # read-only polls must not pay for a full-table UPDATE + COUNT. with conn: - _prune_terminal_unlocked(conn) yield conn finally: conn.close() -def _process_start_time(pid: int) -> Optional[int]: - try: - from gateway.status import get_process_start_time - - return get_process_start_time(pid) - except Exception: - return None - - -def _owner_is_live(pid: int, started_at: Optional[int]) -> bool: - try: - from gateway.status import _pid_exists - - if not _pid_exists(pid): - return False - except Exception: - return True - if started_at is None: - return pid == os.getpid() - return _process_start_time(pid) == started_at - - def enqueue( execution_id: str, job: dict, @@ -325,21 +305,28 @@ def drain( def _terminalize_wait_timeout(execution_id: str) -> str: - """Fence a delivery whose worker can no longer wait for confirmation.""" + """Fence a delivery whose worker can no longer wait for confirmation. + + A row still ``pending`` was provably never attempted, so it is left queued + for whichever gateway comes up next (a restart that includes an update can + easily exceed the worker's wait budget). Only a row caught mid-send is + uncertain and gets fenced ``unknown``. + """ now = _hermes_now().isoformat() - pending_error = "timed out waiting for a live gateway; delivery was not attempted" + pending_error = ( + "timed out waiting for a live gateway; delivery is still queued and " + "will be sent by the next gateway" + ) uncertain_error = ( "timed out while gateway delivery was in progress; outcome is unknown and " "was not retried" ) with _transaction() as conn: - cur = conn.execute( - """UPDATE deliveries SET status='failed', finished_at=?, error=? - WHERE execution_id=? AND status='pending'""", - (now, pending_error, str(execution_id)), - ) - if cur.rowcount: - _prune_terminal_unlocked(conn) + row = conn.execute( + "SELECT status FROM deliveries WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + if row is not None and row["status"] == "pending": return pending_error conn.execute( """UPDATE deliveries SET status='unknown', finished_at=?, error=? @@ -382,5 +369,5 @@ def enqueue_and_wait( return None if row["status"] == "delivered" else str( row.get("error") or f"delivery {row['status']}" ) - time.sleep(0.25) + time.sleep(1.0) return _terminalize_wait_timeout(execution_id) or None diff --git a/cron/executions.py b/cron/executions.py index 3abd5f54df1a..023ed75a4d42 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -62,18 +62,15 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: error TEXT )""" ) - columns = { - str(row[1]) for row in conn.execute("PRAGMA table_info(executions)") - } - if "handoff_pending" not in columns: - conn.execute( - "ALTER TABLE executions " - "ADD COLUMN handoff_pending INTEGER NOT NULL DEFAULT 0" - ) - if "handoff_started_at" not in columns: - conn.execute( - "ALTER TABLE executions ADD COLUMN handoff_started_at REAL" - ) + from hermes_cli.sqlite_util import add_column_if_missing + + add_column_if_missing( + conn, "executions", "handoff_pending", + "handoff_pending INTEGER NOT NULL DEFAULT 0", + ) + add_column_if_missing( + conn, "executions", "handoff_started_at", "handoff_started_at REAL" + ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_executions_job_claimed " "ON executions(job_id, claimed_at DESC, id DESC)" diff --git a/cron/scheduler.py b/cron/scheduler.py index 8191dd710333..65313c4a987f 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -4150,8 +4150,13 @@ def _deliver_result( def drain_delivery_queue(adapters, loop) -> int: """Send queued worker results through this gateway's live adapters.""" - from cron.delivery_queue import drain + from cron.delivery_queue import _path, drain + # Only restart-safe workers create the queue file. Every gateway (macOS, + # Windows, launchd, Docker) runs this housekeeping tick, so skip the sqlite + # open/create entirely until a worker has actually queued something. + if not _path().exists(): + return 0 return drain( lambda queued_job, queued_content, queued_for_failure: _deliver_result( queued_job, @@ -8069,28 +8074,33 @@ def _is_terminal() -> bool: current = get_execution(execution_id) return bool(current and current.get("status") in terminal_states) + # The worker commits its terminal row before its process exits, so exit is + # the correct wakeup. Each ledger read opens a connection and re-runs + # schema init; polling it at 50ms for an hours-long agent run is ~72k + # opens/hour of pure contention with the worker's own writes. while True: - if _is_terminal(): - return True - returncode = process.poll() - if returncode is not None: - # The worker can commit its terminal row and exit between the first - # read and poll(). Re-read the exact attempt before declaring that - # it died without terminalizing. - if _is_terminal(): - return True - # If the adopted worker died without terminalizing, its owner is - # now provably gone. Recover to ``unknown`` rather than routing the - # exception through the pre-handoff dispatch-failure path, which - # would falsely assert that no side effect could have happened. - recover_interrupted_executions() + try: + returncode = process.wait(timeout=1.0) + except subprocess.TimeoutExpired: if _is_terminal(): return True - raise RuntimeError( - "cron external worker exited before durable recovery could " - f"terminalize its execution state (exit {returncode})" - ) - time.sleep(0.05) + continue + # The worker can commit its terminal row and exit between the first + # read and wait(). Re-read the exact attempt before declaring that + # it died without terminalizing. + if _is_terminal(): + return True + # If the adopted worker died without terminalizing, its owner is + # now provably gone. Recover to ``unknown`` rather than routing the + # exception through the pre-handoff dispatch-failure path, which + # would falsely assert that no side effect could have happened. + recover_interrupted_executions() + if _is_terminal(): + return True + raise RuntimeError( + "cron external worker exited before durable recovery could " + f"terminalize its execution state (exit {returncode})" + ) def _wait_for_external_cron_worker( @@ -8098,6 +8108,7 @@ def _wait_for_external_cron_worker( *, execution_id: str, job_id: Optional[str] = None, + handoff_files: tuple[Path, ...] = (), ) -> bool: try: return _wait_for_external_cron_worker_body( @@ -8107,6 +8118,13 @@ def _wait_for_external_cron_worker( if job_id is not None: with _running_lock: _restart_safe_waiter_job_ids.discard(job_id) + # The execution is terminal or its worker is dead: nobody will read a + # payload or acknowledgement left behind by a late/unread handoff. + for stale in handoff_files: + try: + stale.unlink(missing_ok=True) + except OSError: + pass def _launch_external_cron_worker(job: dict) -> bool: @@ -8206,7 +8224,10 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id, ) return _wait_for_external_cron_worker( - process, execution_id=execution_id, job_id=job_id + process, + execution_id=execution_id, + job_id=job_id, + handoff_files=(payload_path,), ) finally: ack_path.unlink(missing_ok=True) @@ -8220,7 +8241,10 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id, ) return _wait_for_external_cron_worker( - process, execution_id=execution_id, job_id=job_id + process, + execution_id=execution_id, + job_id=job_id, + handoff_files=(payload_path,), ) logger.info( "Cron job '%s' handed to restart-safe worker pid=%s execution=%s", @@ -8229,7 +8253,10 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id, ) return _wait_for_external_cron_worker( - process, execution_id=execution_id, job_id=job_id + process, + execution_id=execution_id, + job_id=job_id, + handoff_files=(payload_path,), ) returncode = process.poll() if returncode is not None: @@ -8252,7 +8279,10 @@ def _launch_external_cron_worker(job: dict) -> bool: job_id, ) return _wait_for_external_cron_worker( - process, execution_id=execution_id, job_id=job_id + process, + execution_id=execution_id, + job_id=job_id, + handoff_files=(payload_path, ack_path), ) @@ -8956,6 +8986,14 @@ def _on_done(_f: concurrent.futures.Future) -> None: parser.add_argument("--external-worker-file", type=Path, required=True) parser.add_argument("--ack-file", type=Path, required=True) args = parser.parse_args() + # The gateway spawns this worker with stdout/stderr on DEVNULL; without + # a handler every adoption/ack failure below would be invisible. + try: + from hermes_logging import setup_logging + + setup_logging(hermes_home=_get_hermes_home(), mode="cron") + except Exception: + pass raise SystemExit( 0 if _run_external_worker_payload(args.external_worker_file, args.ack_file) else 1 ) diff --git a/tests/cron/test_delivery_queue.py b/tests/cron/test_delivery_queue.py index b944fda22461..66ad30bb3f24 100644 --- a/tests/cron/test_delivery_queue.py +++ b/tests/cron/test_delivery_queue.py @@ -163,7 +163,11 @@ def test_delivery_failure_is_terminal_not_retried_and_redacted( assert "token=***" in status["error"] -def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): +def test_wait_timeout_leaves_unclaimed_delivery_queued_for_next_gateway( + tmp_path, monkeypatch +): + """A row nobody claimed was never attempted: it is not uncertain, so a + gateway outage longer than the worker's wait budget must not lose it.""" import cron.delivery_queue as queue monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") @@ -171,13 +175,14 @@ def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): error = queue.enqueue_and_wait("exec-3", job, "result", timeout=0) - assert "timed out" in error + assert "timed out" in error and "still queued" in error status = queue.get_status("exec-3") assert status is not None - assert status["status"] == "failed" + assert status["status"] == "pending" send = Mock(return_value=None) - assert queue.drain(send) == 0 - send.assert_not_called() + assert queue.drain(send) == 1 + send.assert_called_once_with(job, "result", False) + assert queue.get_status("exec-3")["status"] == "delivered" def test_same_gateway_recovers_terminalization_failure_without_resending( diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index ad996fd56de6..4820c3e89990 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -207,10 +207,19 @@ class FakeProcess: def poll(self): return self.returncode + def wait(self, timeout=None): + if self.returncode is None: + raise subprocess.TimeoutExpired(cmd="worker", timeout=timeout) + return self.returncode + spawned = [] + payloads = [] + def popen(command, **kwargs): spawned.append((command, kwargs)) + payload_index = command.index("--external-worker-file") + 1 + payloads.append(json.loads(Path(command[payload_index]).read_text())) ack_index = command.index("--ack-file") + 1 Path(command[ack_index]).write_text( json.dumps({"pid": 4321, "execution_id": "exec-1"}), @@ -243,8 +252,9 @@ def popen(command, **kwargs): assert "ANTHROPIC_API_KEY" not in spawned[0][1]["env"] handoff.assert_called_once_with("exec-1") assert get.call_count == 2 - payload = json.loads((tmp_path / "cron/external-workers/exec-1.json").read_text()) - assert payload["multiplex_active"] is True + assert payloads[0]["multiplex_active"] is True + # Once the attempt is terminal the parent reaps its own handoff artifacts. + assert not (tmp_path / "cron/external-workers/exec-1.json").exists() def test_external_worker_exit_rechecks_exact_execution_before_failure(monkeypatch): @@ -260,6 +270,7 @@ def test_external_worker_exit_rechecks_exact_execution_before_failure(monkeypatc monkeypatch.setattr(scheduler, "get_execution", get, raising=False) process = Mock() process.poll.return_value = 0 + process.wait.return_value = 0 assert scheduler._wait_for_external_cron_worker( process, execution_id="exec-1" @@ -272,7 +283,6 @@ def test_external_worker_crash_recovers_uncertain_attempt(monkeypatch): statuses = iter( [ - {"id": "exec-1", "status": "running"}, {"id": "exec-1", "status": "running"}, {"id": "exec-1", "status": "unknown"}, ] @@ -285,12 +295,13 @@ def test_external_worker_crash_recovers_uncertain_attempt(monkeypatch): ) process = Mock() process.poll.return_value = 9 + process.wait.return_value = 9 assert scheduler._wait_for_external_cron_worker( process, execution_id="exec-1" ) is True recover.assert_called_once_with() - assert get.call_count == 3 + assert get.call_count == 2 def test_launch_external_worker_stays_in_process_outside_managed_gateway( diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index a989b01457eb..11fb4501b02c 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -392,7 +392,7 @@ def _run_job( monkeypatch.setattr(scheduler, "heartbeat_fire_claim", _heartbeat) monkeypatch.setattr(scheduler, "run_job", _run_job) monkeypatch.setattr(scheduler, "claim_dispatch", lambda job_id: True) - monkeypatch.setattr(scheduler, "mark_execution_running", lambda execution_id: None) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda execution_id: {}) monkeypatch.setattr(scheduler, "finish_execution", lambda *args, **kwargs: None) save_output = MagicMock() deliver_result = MagicMock() From 57dd3e9af73d0357af2ce4b23db7d1f17c93a719 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 3 Sep 2026 12:29:24 +0530 Subject: [PATCH 5/5] fix(cron): key worker deliveries by the job's own attempt; deferred send is not a failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review fold-in on the salvage of #101877: - `_deliver_result` routed to the durable queue whenever the worker's `_HERMES_CRON_EXTERNAL_WORKER` marker was set, regardless of WHICH job was delivering. A worker whose script dispatches another job in-process (`hermes cron run `) inherits that env and would have queued the nested job's message under the outer execution id — `INSERT OR IGNORE` then drops it silently. Match the marker against the delivering job's own `execution_id`, as `run_one_job` already does. Regression test added (mutation-checked: fails with the guard removed). - A `pending` row left queued at the worker's wait timeout was still reported as a delivery error, so `mark_job_run` recorded `last_status=delivery_failed` for a message the next gateway's drain goes on to send, and nothing ever corrects the job record. Log and return success instead; the deliveries row is the authority for the send. - Reuse `cron.executions._TERMINAL_STATES` in the parent wait loop instead of a second hardcoded terminal set. --- cron/delivery_queue.py | 18 ++++++---- cron/scheduler.py | 14 +++++--- tests/cron/test_delivery_queue.py | 3 +- tests/cron/test_restart_safe_worker.py | 49 ++++++++++++++++++++++++++ 4 files changed, 73 insertions(+), 11 deletions(-) diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index f03413b72714..635a295b2fc4 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -10,6 +10,7 @@ from __future__ import annotations import json +import logging import os import sqlite3 import threading @@ -25,6 +26,8 @@ from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now +logger = logging.getLogger(__name__) + DELIVERY_DB: Optional[Path] = None _PROCESS_ID = uuid.uuid4().hex _lock = threading.RLock() @@ -309,14 +312,12 @@ def _terminalize_wait_timeout(execution_id: str) -> str: A row still ``pending`` was provably never attempted, so it is left queued for whichever gateway comes up next (a restart that includes an update can - easily exceed the worker's wait budget). Only a row caught mid-send is + easily exceed the worker's wait budget). That is a deferral, not a + failure: report success so the job is not recorded ``delivery_failed`` for + a message the drain will still send. Only a row caught mid-send is uncertain and gets fenced ``unknown``. """ now = _hermes_now().isoformat() - pending_error = ( - "timed out waiting for a live gateway; delivery is still queued and " - "will be sent by the next gateway" - ) uncertain_error = ( "timed out while gateway delivery was in progress; outcome is unknown and " "was not retried" @@ -327,7 +328,12 @@ def _terminalize_wait_timeout(execution_id: str) -> str: (str(execution_id),), ).fetchone() if row is not None and row["status"] == "pending": - return pending_error + logger.warning( + "Cron delivery %s: no live gateway within the wait budget; " + "left queued for the next gateway", + execution_id, + ) + return "" conn.execute( """UPDATE deliveries SET status='unknown', finished_at=?, error=? WHERE execution_id=? AND status='delivering'""", diff --git a/cron/scheduler.py b/cron/scheduler.py index 65313c4a987f..267805a8f1a9 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -726,6 +726,7 @@ def _resolve_job_reasoning_config(job: dict, cfg: dict, model: str) -> dict | No use_cron_store, ) from cron.executions import ( + _TERMINAL_STATES, create_execution, finish_execution, get_execution, @@ -3222,8 +3223,15 @@ def _deliver_result( # Hand the send back through a durable queue so the current or replacement # gateway performs it with relay/E2EE parity. The execution id is the # idempotency key; the queue never retries an uncertain claimed send. + # Match on this job's own attempt: a worker's script may itself dispatch + # another job in-process (``hermes cron run``), and that nested delivery + # must not be keyed under the outer execution id. external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER", "") - if external_execution and adapters is None: + if ( + external_execution + and adapters is None + and external_execution == str(job.get("execution_id") or "") + ): from cron.delivery_queue import enqueue_and_wait return enqueue_and_wait( @@ -8068,11 +8076,9 @@ def _wait_for_external_cron_worker_body( A gateway replacement may kill this waiter; it does not kill the scoped worker or change its ledger ownership. """ - terminal_states = {"completed", "failed", "unknown"} - def _is_terminal() -> bool: current = get_execution(execution_id) - return bool(current and current.get("status") in terminal_states) + return bool(current and current.get("status") in _TERMINAL_STATES) # The worker commits its terminal row before its process exits, so exit is # the correct wakeup. Each ledger read opens a connection and re-runs diff --git a/tests/cron/test_delivery_queue.py b/tests/cron/test_delivery_queue.py index 66ad30bb3f24..b4c56edfafa2 100644 --- a/tests/cron/test_delivery_queue.py +++ b/tests/cron/test_delivery_queue.py @@ -175,7 +175,8 @@ def test_wait_timeout_leaves_unclaimed_delivery_queued_for_next_gateway( error = queue.enqueue_and_wait("exec-3", job, "result", timeout=0) - assert "timed out" in error and "still queued" in error + # Deferred, not failed: the worker must not record delivery_failed. + assert error is None status = queue.get_status("exec-3") assert status is not None assert status["status"] == "pending" diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index 4820c3e89990..1be7e30654ab 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -358,6 +358,55 @@ def test_shutdown_does_not_interrupt_restart_safe_waiter(): scheduler._interrupted_job_ids.discard(job_id) +def test_worker_delivery_queue_is_keyed_by_the_delivering_jobs_own_execution( + monkeypatch, tmp_path +): + """A nested in-process dispatch inside a worker (e.g. a script running + ``hermes cron run ``) must not queue under the OUTER execution id.""" + import cron.scheduler as scheduler + + queued = [] + monkeypatch.setattr( + "cron.delivery_queue.enqueue_and_wait", + lambda execution_id, job, content, **kw: ( + queued.append(execution_id) or "queued-marker" + ), + ) + monkeypatch.setattr( + scheduler, + "_resolve_delivery_targets", + lambda job, for_failure=False: [{"platform": "telegram", "chat_id": "123"}], + ) + + def _standalone(*_args, **_kwargs): + raise RuntimeError("standalone path reached") + + # First call the standalone (non-queue) path makes after the guard; the + # failure is reported as the delivery error string. + monkeypatch.setattr("gateway.config.load_gateway_config", _standalone) + monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", "exec-outer") + + # Own attempt: routed through the durable queue. + assert scheduler._deliver_result( + {"id": "job-1", "execution_id": "exec-outer", "deliver": "telegram:123"}, + "done", + adapters=None, + loop=None, + ) == "queued-marker" + assert queued == ["exec-outer"] + + # A different job's attempt: must NOT be queued under exec-outer; it falls + # through to the standalone path. + error = scheduler._deliver_result( + {"id": "job-2", "execution_id": "exec-inner", "deliver": "telegram:123"}, + "done", + adapters=None, + loop=None, + ) + assert error == "failed to load gateway config: standalone path reached" + assert queued == ["exec-outer"] + + def test_gateway_tool_run_without_adapter_objects_hands_off(monkeypatch): import cron.scheduler as scheduler