fix(cron): record a lost manual run instead of reporting a restart that never happened (#128509)

A restart-safe cron worker that dies AFTER adopting its attempt was reported as
"Scheduler restarted after this execution's owner exited" and treated as success.

recover_interrupted_executions() only knows that an owner is gone, so that is all it
can write -- but the waiter holding the worker's Popen knows the owner was that
worker and knows its exit status. Worse, the sweep leaves the row terminal, so the
waiter returned True, run_one_job short-circuited, and mark_job_run never ran: the
job's fire claim outlived its lost execution and refused the next manual fire with
"Job is already being fired" for the whole 300s lease, with an empty last_error.

- executions.terminalize_dead_owner(): terminalize ONE attempt with an observed
  cause. Refuses a live owner, this process, and a handoff inside the adoption
  grace, so it can never steal a running worker's attempt; returns False so the
  caller keeps the generic sweep.
- The waiter records that cause and RAISES, routing the existing post-handoff
  bookkeeping: mark_job_run records the run and retires the claim, no dispatch
  incident and no false crash notice. Also folds the worker's captured stderr into
  the error, which the pre-ack path already did (#112729) and this path dropped.

(cherry picked from commit d3f3571f8c6f4d2f188e90151d8f8393f049a573)

cron/executions.py

@@ -386,6 +386,60 @@ def recover_interrupted_executions() -> int:
     return changed
 
 
+def terminalize_dead_owner(execution_id: str, *, reason: str) -> bool:
+    """Record one attempt as ``unknown`` with a cause this process actually observed.
+
+    ``recover_interrupted_executions`` sweeps every attempt whose owner is provably
+    dead, and knows nothing but that absence — so all it can write is
+    ``_OWNER_GONE_REASON``, which asserts a scheduler restart. A waiter that held the
+    worker's ``Popen`` knows more: the owner was that external worker, and it exited
+    with a known status. Without this, a manual run whose worker dies is filed as
+    "Scheduler restarted ..." (a restart that never happened) and, because the sweep
+    leaves the row terminal, the waiter reports success and never records the run —
+    the job's ``fire_claim`` then blocks the next manual fire for the whole lease
+    (#128509).
+
+    The attempt stays ``unknown``, not ``failed``: whether side effects ran is still
+    unknown. Only the CAUSE becomes truthful. Returns False — leaving the caller to
+    fall back to the generic sweep — when the row is absent, already terminal, owned by
+    this process, inside the handoff adoption grace, or owned by a live process: a
+    worker that is still running must never be terminalized out from under itself.
+    """
+    now = _hermes_now().isoformat()
+    with _transaction() as conn:
+        row = conn.execute(
+            """SELECT id, status, process_id, pid, process_started_at,
+                      handoff_pending, handoff_started_at
+               FROM executions WHERE id=?""",
+            (execution_id,),
+        ).fetchone()
+        if row is None or row["status"] not in ("claimed", "running"):
+            return False
+        if row["process_id"] == _PROCESS_ID:
+            return False
+        if _owner_is_live(int(row["pid"]), row["process_started_at"]):
+            return False
+        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
+        ):
+            return False
+        cur = conn.execute(
+            """UPDATE executions
+               SET status='unknown', finished_at=?, error=?,
+                   handoff_pending=0, handoff_started_at=NULL
+               WHERE id=? AND status=? AND process_id=? AND pid=?""",
+            (now, reason, row["id"], row["status"], row["process_id"], row["pid"]),
+        )
+        if cur.rowcount != 1:
+            return False
+        record = _fetch(conn, execution_id)
+    _emit_execution_state(record)
+    return True
+
+
 def list_executions(
     *, job_id: Optional[str] = None, limit: int = 50, before_claimed_at: Optional[str] = None,
 ) -> List[Dict[str, Any]]:

cron/scheduler.py

@@ -543,7 +543,7 @@ from cron.jobs import (
 from cron.executions import (
     _TERMINAL_STATES, HANDOFF_ADOPTION_GRACE_SECONDS, create_execution, finish_execution,
     get_execution, mark_execution_handoff_pending, mark_execution_running,
-    recover_interrupted_executions)
+    recover_interrupted_executions, terminalize_dead_owner)
 
 # Response marker that suppresses delivery (output is still saved locally for audit).
 SILENT_MARKER = "[SILENT]"
@@ -3415,10 +3415,27 @@ def _run_one_job_body(
             reset_terminal_scope(_terminal_scope_token)
 
 
+def _external_worker_exited_reason(returncode: Optional[int]) -> str:
+    """Cause for an attempt whose restart-safe worker this process watched exit.
+
+    The generic dead-owner sweep can only say the owner vanished, so it names a
+    scheduler restart — but this waiter never lost its scheduler: it handed the
+    attempt to the external worker and then saw that worker exit with this status.
+    The run is still ``unknown`` rather than ``failed`` (whether side effects ran is
+    still unknown); only the cause becomes truthful (#128509).
+    """
+    return (
+        f"Restart-safe cron worker exited with status {returncode} after adopting this "
+        "execution, without writing a durable terminal state; whether side effects ran "
+        "is unknown."
+    )
+
+
 def _wait_for_external_cron_worker_body(
     process: subprocess.Popen,
     *,
     execution_id: str,
+    stderr_path: Optional[Path] = None,
 ) -> bool:
     """Preserve ``run_one_job``'s synchronous contract after handoff.
 
@@ -3455,12 +3472,27 @@ def _wait_for_external_cron_worker_body(
         # 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
+        #
+        # Record the cause THIS waiter observed instead of letting the blanket
+        # sweep stamp "Scheduler restarted ...", and then RAISE: the sweep leaves
+        # the row terminal, so returning success here made run_one_job skip
+        # mark_job_run entirely and the job's fire claim outlived its lost
+        # execution, refusing the next manual fire with "Job is already being
+        # fired" (#128509). Raising routes the existing post-handoff bookkeeping
+        # instead — record the run, no dispatch incident, no crash notice.
+        reason = _external_worker_exited_reason(returncode)
+        if not terminalize_dead_owner(execution_id, reason=reason):
+            recover_interrupted_executions()
+            if _is_terminal():
+                return True
+        stderr_tail = ""
+        if stderr_path is not None:
+            from cron.scheduler_diagnostics import external_worker_stderr_tail
+
+            stderr_tail = external_worker_stderr_tail(stderr_path)
         raise RuntimeError(
-            "cron external worker exited before durable recovery could "
-            f"terminalize its execution state (exit {returncode})"
+            "cron external worker exited before durable recovery could terminalize "
+            f"its execution state (exit {returncode}): {reason}{stderr_tail}"
         )
 
 
@@ -3474,10 +3506,11 @@ def _wait_for_external_cron_worker(
     execution_id: str,
     job_id: Optional[str] = None,
     handoff_files: tuple[Path, ...] = (),
+    stderr_path: Optional[Path] = None,
 ) -> bool:
     try:
         return _wait_for_external_cron_worker_body(
-            process, execution_id=execution_id
+            process, execution_id=execution_id, stderr_path=stderr_path
         )
     except Exception as wait_error:
         raise _ExternalWorkerPostHandoffError(str(wait_error)) from wait_error
@@ -3657,6 +3690,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
                     execution_id=execution_id,
                     job_id=job_id,
                     handoff_files=(payload_path, stderr_path),
+                    stderr_path=stderr_path,
                 )
             finally:
                 ack_path.unlink(missing_ok=True)
@@ -3674,6 +3708,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
                     execution_id=execution_id,
                     job_id=job_id,
                     handoff_files=(payload_path, stderr_path),
+                    stderr_path=stderr_path,
                 )
             logger.info(
                 "Cron job '%s' handed to restart-safe worker pid=%s execution=%s",
@@ -3689,6 +3724,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
                 execution_id=execution_id,
                 job_id=job_id,
                 handoff_files=(payload_path, stderr_path),
+                stderr_path=stderr_path,
             )
         returncode = process.poll()
         if returncode is not None:
@@ -3728,6 +3764,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
         execution_id=execution_id,
         job_id=job_id,
         handoff_files=(payload_path, ack_path, stderr_path),
+        stderr_path=stderr_path,
     )
 
 

tests/cron/test_restart_safe_worker.py

@@ -1172,7 +1172,7 @@ def test_post_handoff_waiter_failure_records_bookkeeping_without_alert(
     import cron.incidents as incidents
     import cron.scheduler as scheduler
 
-    def _body_boom(_process, *, execution_id):
+    def _body_boom(_process, *, execution_id, **_kwargs):
         raise RuntimeError("cron external worker exited before durable recovery")
 
     monkeypatch.setattr(scheduler, "_wait_for_external_cron_worker_body", _body_boom)

tests/cron/test_restart_safe_worker_death_evidence.py

@@ -0,0 +1,157 @@
+"""Durable evidence when a restart-safe cron worker dies AFTER adopting its attempt.
+
+Regression for #128509. A ``cronjob(action='run')`` fired inside a gateway agent turn
+returns ``executed=true / execution_mode=background``, then the attempt dies. The waiter
+learns the adopted worker exited and calls ``recover_interrupted_executions()``, which only
+knows "the owner is gone" and stamps the generic ``_OWNER_GONE_REASON`` -- "Scheduler
+restarted after this execution's owner exited" -- a restart that never happened. The waiter
+then returns success, so ``run_one_job`` short-circuits and never calls ``mark_job_run``:
+
+* the job's ``fire_claim`` is left behind, so the next manual fire is refused with
+  "Job is already being fired by the scheduler" until the 300s lease expires, and
+* ``last_error`` stays empty, so the operator has no error and no cause anywhere.
+
+The worker is not in the calling turn's process group (``start_new_session=True`` plus a
+``systemd-run --user --scope`` cgroup), so turn teardown never signalled it: the child
+crashed on its own. What the gateway still owes the operator is a truthful cause and a
+released claim.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+# A real worker process: it adopts the durable row (the ownership rewrite that makes the
+# reported status line reachable), publishes the acknowledgement, then dies on its own.
+# The sleep keeps the parent's ack poll (every 50ms) deterministically ahead of the exit.
+_WORKER_THAT_DIES_AFTER_ADOPTION = """
+import json, os, sys, time
+from pathlib import Path
+
+argv = sys.argv[1:]
+payload_path = Path(argv[argv.index("--external-worker-file") + 1])
+ack_path = Path(argv[argv.index("--ack-file") + 1])
+
+import cron.executions as executions
+
+job = json.loads(payload_path.read_text(encoding="utf-8-sig"))["job"]
+assert executions.adopt_claimed_execution(job["execution_id"]) is not None
+ack_path.write_text(
+    json.dumps({"pid": os.getpid(), "execution_id": job["execution_id"]}),
+    encoding="utf-8",
+)
+time.sleep(1.5)
+os._exit(9)
+"""
+
+
+@pytest.fixture
+def profile_store(tmp_path, monkeypatch):
+    """A real profile-scoped cron store, ledger and scheduler home, all on one temp path."""
+    from cron.jobs import create_job, use_cron_store
+
+    home = tmp_path / "home"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    import cron.executions as executions
+    import cron.scheduler as scheduler
+
+    monkeypatch.setattr(scheduler, "_hermes_home", home)
+    monkeypatch.setattr(executions, "EXECUTIONS_FILE", home / "cron" / "executions.db")
+    monkeypatch.setattr(scheduler, "load_config_readonly", lambda: {})
+    return home
+
+
+def test_worker_dying_after_adoption_reports_its_cause_and_releases_the_claim(
+    tmp_path, monkeypatch, profile_store
+):
+    """The run is lost either way; the ledger must not blame a restart that never
+    happened, the job must be claimable again, and the operator must get a cause."""
+    import cron.executions as executions
+    import cron.scheduler as scheduler
+    from cron.jobs import claim_job_for_fire, create_job, get_job, use_cron_store
+    from tools.process_registry import GatewayChildDispatch
+
+    home = profile_store
+    with use_cron_store(home):
+        job = create_job(prompt="summarise the day", schedule="every 1h", name="daily brief")
+        claimed = claim_job_for_fire(job["id"], manual=True, return_job=True)
+    assert claimed, "the manual claim must succeed or this test proves nothing"
+
+    worker = tmp_path / "worker_that_dies.py"
+    worker.write_text(_WORKER_THAT_DIES_AFTER_ADOPTION, encoding="utf-8")
+
+    def dispatch(command, *, unit_suffix, require_restart_safe_scope=False):
+        # Keep the launcher's real payload/ack argv; swap only the child's body so the
+        # spawn, the acknowledgement poll and the wait all run for real.
+        flags = command[command.index("--external-worker-file"):]
+        return GatewayChildDispatch(
+            "degraded", [sys.executable, str(worker), *flags])
+
+    monkeypatch.setattr(
+        "tools.process_registry.restart_safe_gateway_child_argv", dispatch)
+
+    assert scheduler.run_one_job(claimed, adapters=None) is True
+    execution_id = claimed["execution_id"]
+
+    with use_cron_store(home):
+        record = get_job(job["id"])
+        # The claim is what makes the next manual fire say "Job is already being fired";
+        # only mark_job_run retires it, and the lost waiter used to never call it.
+        assert record["fire_claim"] is None, "the fire claim outlived its lost execution"
+        assert claim_job_for_fire(job["id"], manual=True) is True
+
+        # A truthful cause: this waiter never lost its scheduler, it watched ITS worker
+        # exit. The generic dead-owner sweep can only assert a restart.
+        assert record["last_error"], "a lost run recorded no error at all"
+        assert "Scheduler restarted" not in record["last_error"]
+        assert "exit 9" in record["last_error"]
+
+    # Still `unknown`, not a fabricated durable state: side effects really are unknown.
+    row = executions.get_execution(execution_id)
+    assert row["status"] == "unknown"
+    assert "Scheduler restarted" not in (row["error"] or "")
+
+
+def test_terminalizer_refuses_an_attempt_whose_worker_is_still_alive(
+    profile_store,
+):
+    """The targeted terminalizer must not steal a LIVE worker's attempt: refusing a live
+    owner is what keeps the fix from becoming a second way to lose side effects."""
+    import cron.executions as executions
+
+    script = (
+        "import sys, time\n"
+        "from pathlib import Path\n"
+        "import cron.executions as executions\n"
+        f"executions.EXECUTIONS_FILE = Path({str(executions.EXECUTIONS_FILE)!r})\n"
+        "assert executions.adopt_claimed_execution(sys.argv[1]) is not None\n"
+        "sys.stdout.write('adopted\\n')\n"
+        "sys.stdout.flush()\n"
+        "time.sleep(60)\n"
+    )
+    record = executions.create_execution("job-live", source="direct")
+    assert executions.mark_execution_handoff_pending(record["id"]) is not None
+    live_worker = subprocess.Popen(
+        [sys.executable, "-c", script, record["id"]],
+        stdout=subprocess.PIPE, text=True)
+    try:
+        deadline = time.monotonic() + 30.0
+        while time.monotonic() < deadline:
+            if executions.get_execution(record["id"])["status"] == "running":
+                break
+            time.sleep(0.05)
+        assert executions.get_execution(record["id"])["status"] == "running"
+
+        assert executions.terminalize_dead_owner(
+            record["id"], reason="exit 9") is False
+        assert executions.get_execution(record["id"])["status"] == "running"
+    finally:
+        live_worker.kill()
+        live_worker.wait(timeout=30)