Cron external workers no longer relaunch after their ack and leave the run unknown (#124827, salvage #133720)

contributors/emails/mzykr@yahoo.co.jp

@@ -0,0 +1,2 @@
+mzyas
+# PR #133720 salvage

cron/AGENTS.md

@@ -79,7 +79,11 @@ Hardening invariants — each guards a real failure; don't weaken without answer
   activates it before the first third-party import. A failed activation is fatal: the worker
   exits before its ownership ack (a reported dispatch failure) rather than run on an unleased
   generation the collector may delete. The gateway never re-runs the boot — `hermes_bootstrap`
-  already did at its own launch (#122222).
+  already did at its own launch (#122222). The worker's `__main__` imports `hermes_bootstrap`
+  (`finish_worker_boot`) and runs `run_agent`'s interrupted-pull recovery before it reads the
+  payload, and only then drops the marker: a source-update relaunch (`-I`, no `PYTHONPATH`) or a
+  post-restore re-exec must replay the whole worker, never land after the ack with the payload
+  gone and no dependency boot.
 - Cron sessions pass `skip_memory=True`; memory providers intentionally do not run during cron.
 - Cron execution has its own session. Eligible continuable deliveries may mirror or seed the
   reply-facing conversation: origin, origin-less home fallback, user-written bare-platform home,

cron/scheduler.py

@@ -36,7 +36,7 @@
 # `hermes update`) otherwise fail with ModuleNotFoundError for hermes_time et al.
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
-from cron.worker_bootstrap import WORKER_MARKER
+from cron.worker_bootstrap import WORKER_MARKER, finish_worker_boot
 from hermes_constants import get_hermes_home, hermes_home_key
 from hermes_cli.observability.shared_metrics_gateway import note_cron_execution, note_cron_skipped
 from cron.env_settings import cron_env_setting
@@ -4435,6 +4435,7 @@ def _on_done(_f: concurrent.futures.Future) -> None:
 # tick paths see every name they need.
 if __name__ == "__main__":
     if "--external-worker-file" in sys.argv:
+        finish_worker_boot()  # may relaunch: before the payload is read and the ack published
         import argparse
 
         parser = argparse.ArgumentParser(add_help=False)

cron/worker_bootstrap.py

@@ -2,7 +2,7 @@
 
 ``sys.executable -m cron.scheduler --external-worker-file ...``
 (``cron/scheduler.py::_launch_external_cron_worker``) is a full Hermes entry point that
-does not go through ``hermes_bootstrap``, and it imports Hermes packages the moment it
+reaches ``hermes_bootstrap`` only from its ``__main__``, and it imports Hermes packages the moment it
 starts. ``cron/scheduler_worker_env.py`` restores the committed generation's
 ``site-packages`` on its ``PYTHONPATH`` so those imports resolve, but a pinned path is not a
 boot: the worker holds no lease on the generation, so the PM collector may remove it
@@ -23,23 +23,45 @@
 acknowledgement, which ``_launch_external_cron_worker`` already reports as a failed dispatch with
 the worker's stderr. PM's legitimate no-ops (no committed generation under a venv, wheel, Nix)
 return normally.
+
+That boot covers dependencies only. ``hermes_bootstrap`` also finishes a pending source update
+and may relaunch the process into a fresh one (``hermes_cli/venv_sync.py::relaunch_command``, an
+``-I`` interpreter that ignores the pinned ``PYTHONPATH``), and re-executes it after restoring a
+tree a killed ``hermes update`` half-wrote. ``finish_worker_boot()`` runs it from
+the worker's ``__main__`` before the payload is read, so a relaunch replays the whole worker: the
+payload is still on disk and the marker still set, and the new process boots its dependencies
+again. ``hermes_bootstrap`` re-runs the (idempotent) dependency boot; the package-init call above
+must stay, because ``cron.jobs`` needs dependencies before ``__main__`` can run.
 """
 
 from __future__ import annotations
 
 import os
 from pathlib import Path
 
-# Set by ``_launch_external_cron_worker`` in the child's env and consumed here, so it never
-# reaches the worker's own children -- they inherit the activated PYTHONPATH instead.
+# Set by ``_launch_external_cron_worker`` in the child's env and consumed by
+# ``finish_worker_boot()`` -- after any relaunch, before the ack -- so it never reaches the
+# worker's own children; they inherit the activated PYTHONPATH instead.
 # Distinct from ``_HERMES_CRON_EXTERNAL_WORKER`` (the owning execution id) in scheduler.py.
 WORKER_MARKER = "_HERMES_CRON_WORKER_BOOT"
 
 
 def worker_bootstrap() -> None:
-    """Run PM's dependency boot in the marked external worker, once."""
-    if not os.environ.pop(WORKER_MARKER, None):
+    """Run PM's dependency boot in the marked external worker (again after a relaunch)."""
+    if not os.environ.get(WORKER_MARKER):
         return
     from pm.environments import activate_dependencies
 
     activate_dependencies(Path(__file__).resolve().parent.parent)
+
+
+def finish_worker_boot() -> None:
+    """Run ``hermes_bootstrap`` before the worker reads its payload; may relaunch the worker.
+
+    Not from ``worker_bootstrap()``: while ``-m cron.scheduler`` executes ``cron/__init__.py``,
+    ``sys.argv[0]`` is still ``-m`` and ``__main__`` has no spec, so the relaunch command could
+    not name the module to re-run. In ``__main__`` both are set.
+    """
+    import hermes_bootstrap  # noqa: F401
+
+    os.environ.pop(WORKER_MARKER, None)

tests/cron/test_restart_safe_worker.py

@@ -11,6 +11,7 @@
 import threading
 import time
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import Mock
 
 import pytest
@@ -757,9 +758,9 @@ def test_pin_restores_the_committed_generation_site_packages(tmp_path):
 def test_marked_worker_boots_dependencies_before_cron_jobs(marked):
     """#122222: ``-m cron.scheduler`` executes ``cron/__init__.py`` first, whose first import
     (``cron.jobs`` -> ``hermes_yaml`` -> ``ruamel``) is already a dependency, so the marked
-    worker must boot before it -- exactly once, consuming the marker so the worker's own
-    children do not inherit it. An unmarked importer (the gateway already booted through
-    ``hermes_bootstrap``) is never re-booted."""
+    worker must boot before it -- exactly once. The marker survives the package import so a
+    relaunched worker boots again (``finish_worker_boot`` consumes it). An unmarked importer
+    (the gateway already booted through ``hermes_bootstrap``) is never re-booted."""
     import cron.worker_bootstrap as worker_bootstrap
 
     repo_root = Path(worker_bootstrap.__file__).resolve().parent.parent
@@ -773,7 +774,83 @@ def test_marked_worker_boots_dependencies_before_cron_jobs(marked):
     )
     assert child.returncode == 0, child.stderr
     result = json.loads(child.stdout.strip().splitlines()[-1])
-    assert result == {"boots": [False] if marked else [], "marker": None}
+    assert result == {"boots": [False] if marked else [], "marker": "1" if marked else None}
+
+
+_RELAUNCH_PROBE = """
+import json, os, runpy, subprocess, sys
+from pathlib import Path
+
+import pm.environments
+pm.environments.activate_dependencies = lambda root: None
+import hermes_cli._early_recovery as early_recovery
+import hermes_cli.venv_sync as venv_sync
+
+payload, ack, record, marker = Path(sys.argv[1]), Path(sys.argv[2]), Path(sys.argv[3]), sys.argv[4]
+if sys.argv[5] == "source_update":
+    venv_sync.prepare_launch = lambda root, argv: Path(sys.executable)
+else:
+    venv_sync.prepare_launch = lambda root, argv: None
+    early_recovery.restore_interrupted_pull = lambda *args, **kwargs: True
+
+
+def relaunch(*args):
+    record.write_text(json.dumps({
+        "payload": payload.exists(), "ack": ack.exists(),
+        "marker": os.environ.get(marker), "command": args[-1],
+    }))
+    raise SystemExit(0)
+
+
+os.execv = relaunch
+subprocess.call = relaunch
+sys.argv = ["cron.scheduler", "--external-worker-file", str(payload), "--ack-file", str(ack)]
+runpy.run_module("cron.scheduler", run_name="__main__", alter_sys=True)
+"""
+
+
+@pytest.mark.parametrize("cause", ["source_update", "interrupted_pull"])
+def test_relaunch_replays_the_worker_before_its_ack(tmp_path, cause):
+    """Importing ``hermes_bootstrap`` may relaunch the process: finishing a source
+    update (into an ``-I`` interpreter that ignores the pinned PYTHONPATH), or the restore of a
+    tree a killed ``hermes update`` half-wrote. Reached inside ``run_one_job``, either came after
+    the ack: the payload was deleted, the marker consumed, and the new process died on ``ruamel``
+    or found no payload, leaving the adopted run ``unknown``. The relaunch must instead replay
+    the whole worker: payload unread, no ack, marker set, the same worker re-run."""
+    import cron.worker_bootstrap as worker_bootstrap
+
+    repo_root = Path(worker_bootstrap.__file__).resolve().parent.parent
+    payload = tmp_path / "exec-1.json"
+    payload.write_text(json.dumps({
+        "job": {"id": "job-1", "execution_id": "exec-1"},
+        "profile_home": str(tmp_path / "profile"),
+    }), encoding="utf-8")
+    ack, record = tmp_path / "exec-1.ready", tmp_path / "relaunch.json"
+    env = dict(os.environ, PYTHONPATH=str(repo_root))
+    env[worker_bootstrap.WORKER_MARKER] = "1"
+    child = subprocess.run(
+        [sys.executable, "-c", _RELAUNCH_PROBE, str(payload), str(ack), str(record),
+         worker_bootstrap.WORKER_MARKER, cause],
+        cwd=repo_root, env=env, capture_output=True, text=True, timeout=120,
+    )
+    assert record.exists(), f"worker never relaunched (exit {child.returncode}): {child.stderr}"
+    seen = json.loads(record.read_text())
+    assert (seen["payload"], seen["ack"], seen["marker"]) == (True, False, "1")
+    if cause == "source_update":
+        assert "runpy.run_module('cron.scheduler', run_name='__main__'" in seen["command"][-1]
+    else:
+        assert str(payload) in seen["command"]
+
+
+def test_booted_worker_drops_the_marker_before_any_job_child(monkeypatch):
+    """Without a relaunch, ``finish_worker_boot`` must consume the marker: job children spawned
+    later inherit this environment and would otherwise re-run the worker dependency boot."""
+    import cron.worker_bootstrap as worker_bootstrap
+
+    monkeypatch.setitem(sys.modules, "hermes_bootstrap", SimpleNamespace())
+    monkeypatch.setenv(worker_bootstrap.WORKER_MARKER, "1")
+    worker_bootstrap.finish_worker_boot()
+    assert worker_bootstrap.WORKER_MARKER not in os.environ
 
 
 _REAL_BOOT_PROBE = """