fix(cron): keep direct runs alive after CLI timeout

cron/scheduler.py

@@ -45,7 +45,10 @@
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
 from hermes_constants import get_hermes_home
-from hermes_cli._subprocess_compat import windows_hide_flags
+from hermes_cli._subprocess_compat import (
+    windows_detach_popen_kwargs,
+    windows_hide_flags,
+)
 from hermes_cli.config import (
     _expand_env_vars,
     cron_model_drift_axes,
@@ -8133,13 +8136,19 @@ def _wait_for_external_cron_worker(
                 pass
 
 
-def _launch_external_cron_worker(job: dict) -> bool:
-    """Launch *job* outside a managed gateway cgroup when required.
+def _launch_external_cron_worker(
+    job: dict,
+    *,
+    detach: bool = False,
+    allow_unscoped: bool = False,
+    extra_prompt: Optional[str] = None,
+) -> bool:
+    """Launch *job* in a process that survives its dispatching caller.
 
-    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.
+    Managed gateways use a transient systemd scope and wait for the worker's
+    terminal ledger state. Direct CLI runs pass ``allow_unscoped`` and
+    ``detach``: they return after ownership acknowledgement with a durable
+    execution receipt. Other unscoped callers keep the in-process path.
     """
     execution_id = str(job["execution_id"])
     job_id = str(job["id"])
@@ -8165,7 +8174,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
         command,
         unit_suffix=f"cron-{job_id}-exec-{execution_id}",
     )
-    if scoped_command == command:
+    if scoped_command == command and not allow_unscoped:
         return False
 
     if mark_execution_handoff_pending(execution_id) is None:
@@ -8186,6 +8195,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
                     "job": job,
                     "profile_home": str(_get_hermes_home().resolve()),
                     "multiplex_active": multiplex_active,
+                    "extra_prompt": extra_prompt,
                 },
                 payload_file,
             )
@@ -8208,8 +8218,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
             stdin=subprocess.DEVNULL,
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
-            start_new_session=True,
-            creationflags=windows_hide_flags(),
+            **windows_detach_popen_kwargs(),
         )
     except BaseException:
         payload_path.unlink(missing_ok=True)
@@ -8258,6 +8267,8 @@ def _launch_external_cron_worker(job: dict) -> bool:
                 acknowledgement.get("pid"),
                 execution_id,
             )
+            if detach:
+                return True
             return _wait_for_external_cron_worker(
                 process,
                 execution_id=execution_id,
@@ -8358,7 +8369,13 @@ def _run_external_worker_payload(payload_path: Path, ack_path: Path) -> bool:
             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)
+                return run_one_job(
+                    job,
+                    adapters=None,
+                    loop=None,
+                    verbose=False,
+                    extra_prompt=payload.get("extra_prompt"),
+                )
             finally:
                 if old_external_execution is None:
                     os.environ.pop("_HERMES_CRON_EXTERNAL_WORKER", None)

hermes_cli/cron.py

@@ -931,7 +931,11 @@ def _stateless_reset() -> None:
         except Exception:
             _stateless_reset = None
     try:
-        result = _cron_api(action=action, job_id=job_id)
+        result = _cron_api(
+            action=action,
+            job_id=job_id,
+            **({"detach_run": True} if action == "run" else {}),
+        )
     finally:
         if _stateless_reset is not None:
             _stateless_reset()
@@ -953,7 +957,13 @@ def _stateless_reset() -> None:
         # success/failure verdict would be a lie (#83340). Report the
         # background dispatch instead of claiming the run failed.
         delegation_id = job.get("delegation_id")
-        if job.get("execution_mode") == "background" or delegation_id:
+        execution_id = job.get("execution_id")
+        if job.get("execution_mode") == "detached":
+            if execution_id:
+                print(f"  Running in detached worker (execution {execution_id}).")
+            else:
+                print("  Running in detached worker.")
+        elif job.get("execution_mode") == "background" or delegation_id:
             if delegation_id:
                 print(f"  Running in background (delegation {delegation_id}).")
             else:

tests/cron/test_cron_relay_run_forward.py

@@ -148,6 +148,31 @@ def fake_post(url, headers=None, json=None, timeout=None):
             cronjob_tools._forward_relay_fronted_run({"id": "j1"})
         assert sent["json"] == {}
 
+    def test_detached_direct_run_forwards_instead_of_launching_worker(self):
+        job = {"id": "j1", "name": "relay run", "deliver": "discord"}
+        with patch.object(
+            cronjob_tools, "resolve_job_ref", return_value=job
+        ), patch.object(
+            cronjob_tools,
+            "_relay_fronted_delivery_platforms",
+            return_value={"discord"},
+        ), patch("httpx.post", return_value=_Resp(200)) as post, patch.object(
+            cronjob_tools, "_try_dispatch_detached_run"
+        ) as detached:
+            out = json.loads(
+                cronjob_tools.cronjob(
+                    action="run",
+                    job_id="j1",
+                    prompt="one fire only",
+                    detach_run=True,
+                )
+            )
+
+        assert out["success"] is True
+        assert out["forwarded_to_gateway"] is True
+        assert post.call_args.kwargs["json"] == {"prompt": "one fire only"}
+        detached.assert_not_called()
+
 
 class TestManualRunPromptConsumption:
     """The stamped transient prompt reaches the fire that consumes the

tests/cron/test_restart_safe_worker.py

@@ -230,6 +230,10 @@ def popen(command, **kwargs):
     handoff = Mock(return_value={"id": "exec-1", "handoff_pending": 1})
     monkeypatch.setattr(scheduler, "mark_execution_handoff_pending", handoff)
     monkeypatch.setattr(scheduler.subprocess, "Popen", popen)
+    monkeypatch.setattr(
+        scheduler, "windows_detach_popen_kwargs",
+        lambda: {"start_new_session": True},
+    )
     observed_statuses = iter(
         [
             {"id": "exec-1", "status": "running"},
@@ -257,6 +261,56 @@ def popen(command, **kwargs):
     assert not (tmp_path / "cron/external-workers/exec-1.json").exists()
 
 
+def test_direct_cli_worker_detaches_without_waiting_for_terminal_state(
+    tmp_path, monkeypatch
+):
+    import cron.scheduler as scheduler
+
+    job = {"id": "job-direct", "execution_id": "exec-direct", "prompt": "work"}
+    monkeypatch.setattr(scheduler, "_get_hermes_home", lambda: tmp_path)
+    monkeypatch.setattr(
+        "tools.process_registry.restart_safe_gateway_child_argv",
+        lambda command, **_kwargs: command,
+    )
+    monkeypatch.setattr(
+        scheduler, "mark_execution_handoff_pending",
+        Mock(return_value={"id": "exec-direct", "handoff_pending": 1}),
+    )
+    monkeypatch.setattr(
+        scheduler, "windows_detach_popen_kwargs",
+        lambda: {"start_new_session": True},
+    )
+
+    process = Mock()
+    process.poll.return_value = None
+    process.wait.side_effect = AssertionError("detached launch waited for completion")
+
+    def popen(command, **_kwargs):
+        ack = Path(command[command.index("--ack-file") + 1])
+        ack.write_text(
+            json.dumps({"pid": 4321, "execution_id": "exec-direct"}),
+            encoding="utf-8",
+        )
+        return process
+
+    monkeypatch.setattr(scheduler.subprocess, "Popen", popen)
+
+    assert scheduler._launch_external_cron_worker(
+        job,
+        detach=True,
+        allow_unscoped=True,
+        extra_prompt="one fire only",
+    ) is True
+
+    process.wait.assert_not_called()
+    payload = json.loads(
+        (tmp_path / "cron/external-workers/exec-direct.json").read_text(
+            encoding="utf-8"
+        )
+    )
+    assert payload["extra_prompt"] == "one fire only"
+
+
 def test_external_worker_exit_rechecks_exact_execution_before_failure(monkeypatch):
     import cron.scheduler as scheduler
 

tests/hermes_cli/test_cron.py

@@ -443,6 +443,33 @@ def _run_cmd(self, capsys):
         rc = cron_command(Namespace(cron_command="run", job_id="job-1"))
         return rc, capsys.readouterr().out
 
+    def test_direct_cli_requests_detached_run_and_prints_execution_receipt(
+        self, monkeypatch, capsys
+    ):
+        observed = {}
+
+        def fake_cron_api(**kwargs):
+            observed.update(kwargs)
+            return {
+                "success": True,
+                "job": {
+                    "id": "job-1",
+                    "name": "Watchdog",
+                    "executed": True,
+                    "execution_mode": "detached",
+                    "execution_id": "exec-123",
+                },
+            }
+
+        monkeypatch.setattr(cron_cli, "_cron_api", fake_cron_api)
+
+        rc, out = self._run_cmd(capsys)
+
+        assert rc == 0
+        assert observed["detach_run"] is True
+        assert "Running in detached worker (execution exec-123)." in out
+        assert "Ran now" not in out
+
     def test_background_dispatch_with_delegation_id_does_not_report_failed(
         self, monkeypatch, capsys
     ):

tests/tools/test_cronjob_run_background.py

@@ -17,6 +17,7 @@
 from unittest.mock import patch
 
 from tools.cronjob_tools import (
+    _try_dispatch_detached_run,
     _try_dispatch_background_run,
     cronjob,
 )
@@ -195,6 +196,46 @@ def test_pool_at_capacity_runs_inline(self):
         m_run.assert_called_once()   # ran inline on this thread
 
 
+class TestDetachedCliDispatch:
+    def test_claims_creates_receipt_and_hands_ownership_to_worker(self):
+        claimed = {
+            **_job("job-direct-helper"),
+            "fire_claim": {"by": "direct-owner"},
+        }
+        launch_observed = {}
+
+        def launch(job, **kwargs):
+            launch_observed["job"] = job
+            launch_observed["kwargs"] = kwargs
+            return True
+
+        with patch(
+            "tools.cronjob_tools.claim_job_for_fire", return_value=claimed
+        ) as claim, patch(
+            "cron.executions.create_execution",
+            return_value={"id": "exec-direct-helper"},
+        ) as create, patch(
+            "cron.scheduler._launch_external_cron_worker", side_effect=launch
+        ):
+            result = _try_dispatch_detached_run(
+                _job("job-direct-helper"), extra_prompt="one fire"
+            )
+
+        assert result == {
+            "claimed": True,
+            "dispatched": True,
+            "execution_id": "exec-direct-helper",
+        }
+        claim.assert_called_once_with("job-direct-helper", return_job=True)
+        create.assert_called_once_with("job-direct-helper", source="direct")
+        assert launch_observed["job"]["execution_id"] == "exec-direct-helper"
+        assert launch_observed["kwargs"] == {
+            "detach": True,
+            "allow_unscoped": True,
+            "extra_prompt": "one fire",
+        }
+
+
 class TestInFlightDedupe:
     """Manual runs must not double-fire a job that is already mid-run
     (salvaged from #53395 by @izumi0uu): the fire claim's 300s TTL is
@@ -296,6 +337,30 @@ def test_ticker_guard_uses_shared_helpers(self):
 
 
 class TestCronjobRunToolIntegration:
+    def test_direct_cli_detach_returns_durable_execution_receipt(self):
+        detached = {
+            "claimed": True,
+            "dispatched": True,
+            "execution_id": "exec-direct-1",
+        }
+        with patch("tools.cronjob_tools.resolve_job_ref", return_value=_job("job-direct-1")), \
+             patch("tools.cronjob_tools._try_dispatch_detached_run", return_value=detached) as dispatch, \
+             patch("tools.cronjob_tools._try_dispatch_background_run") as background, \
+             patch("tools.cronjob_tools._execute_job_now") as inline, \
+             patch("tools.cronjob_tools.get_job", return_value={
+                 "id": "job-direct-1", "name": "direct run",
+             }):
+            out = json.loads(cronjob(
+                action="run", job_id="job-direct-1", detach_run=True
+            ))
+
+        assert out["success"] is True
+        assert out["job"]["execution_mode"] == "detached"
+        assert out["job"]["execution_id"] == "exec-direct-1"
+        dispatch.assert_called_once()
+        background.assert_not_called()
+        inline.assert_not_called()
+
     def test_run_action_returns_background_note(self):
         """cronjob(action='run') surfaces the handle + do-not-wait note."""
         with _bound_session_key():