Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 36 additions & 18 deletions cron/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -3101,6 +3101,8 @@ def _launch_external_cron_worker(job: dict) -> bool:
handoff_dir = _get_hermes_home() / "cron" / "external-workers"
payload_path = handoff_dir / f"{execution_id}.json"
ack_path = handoff_dir / f"{execution_id}.ready"
# Captured so a worker that dies before its acknowledgement can name the cause (#112729).
stderr_path = handoff_dir / f"{execution_id}.stderr"
command = [
sys.executable,
"-m",
Expand Down Expand Up @@ -3190,18 +3192,23 @@ def _launch_external_cron_worker(job: dict) -> bool:
):
worker_env.pop(_presence_var, None)
try:
process = subprocess.Popen(
dispatch.argv,
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(),
)
stderr_fd = os.open(stderr_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
try:
process = subprocess.Popen(
dispatch.argv,
cwd=str(Path(__file__).resolve().parent.parent),
env=worker_env,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=stderr_fd,
start_new_session=True,
creationflags=windows_hide_flags(),
)
finally:
os.close(stderr_fd)
except BaseException:
payload_path.unlink(missing_ok=True)
stderr_path.unlink(missing_ok=True)
raise

with _running_lock:
Expand All @@ -3226,7 +3233,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
process,
execution_id=execution_id,
job_id=job_id,
handoff_files=(payload_path,),
handoff_files=(payload_path, stderr_path),
)
finally:
ack_path.unlink(missing_ok=True)
Expand All @@ -3243,7 +3250,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
process,
execution_id=execution_id,
job_id=job_id,
handoff_files=(payload_path,),
handoff_files=(payload_path, stderr_path),
)
logger.info(
"Cron job '%s' handed to restart-safe worker pid=%s execution=%s",
Expand All @@ -3257,15 +3264,19 @@ def _launch_external_cron_worker(job: dict) -> bool:
process,
execution_id=execution_id,
job_id=job_id,
handoff_files=(payload_path,),
handoff_files=(payload_path, stderr_path),
)
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)
from cron.scheduler_diagnostics import external_worker_stderr_tail
stderr_tail = external_worker_stderr_tail(stderr_path)
stderr_path.unlink(missing_ok=True)
if dispatch.mode == "scoped" and scoped_spawn_lost_user_bus(worker_env):
# systemd-run itself failed (stderr is DEVNULL): name the cause, not the exit code.
# systemd-run itself failed before any worker ran, so the captured stderr
# holds nothing useful: name the cause, not the exit code.
raise RuntimeError(
"restart-safe systemd scope could not be created: the user D-Bus session at "
f"/run/user/{os.getuid()}/bus disappeared after the gateway started. On a " # windows-footgun: ok — scoped dispatch exists only on Linux
Expand All @@ -3274,7 +3285,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
)
raise RuntimeError(
f"cron external worker exited before ownership acknowledgement "
f"(exit {returncode})"
f"(exit {returncode}){stderr_tail}"
)
time.sleep(0.05)

Expand All @@ -3292,7 +3303,7 @@ def _launch_external_cron_worker(job: dict) -> bool:
process,
execution_id=execution_id,
job_id=job_id,
handoff_files=(payload_path, ack_path),
handoff_files=(payload_path, ack_path, stderr_path),
)


Expand Down Expand Up @@ -3368,6 +3379,11 @@ def _run_external_worker_payload(payload_path: Path, ack_path: Path) -> bool:
os.environ.pop("_HERMES_CRON_EXTERNAL_WORKER", None)
else:
os.environ["_HERMES_CRON_EXTERNAL_WORKER"] = old_external_execution
# Post-ack the gateway never reads the stderr capture (it only
# serves the pre-ack death report) and may not outlive this run
# in the restart-safe topology, so the worker removes its own.
with contextlib.suppress(OSError):
ack_path.with_suffix(".stderr").unlink(missing_ok=True)
finally:
reset_secret_scope(secret_token)
set_multiplex_active(previous_multiplex)
Expand Down Expand Up @@ -3888,8 +3904,10 @@ def _process_job(job: dict) -> bool:
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.
# The gateway spawns this worker with stdout on DEVNULL and stderr on a
# capture file it only reads back if we die before the ack; without a
# log handler every adoption/ack failure below would otherwise be
# invisible to the persistent log.
try:
from hermes_logging import setup_logging

Expand Down
25 changes: 25 additions & 0 deletions cron/scheduler_diagnostics.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,38 @@
"""Private run-document diagnostics, kept separate from delivery summaries."""

from pathlib import Path
from traceback import format_exception

from agent.redact import redact_sensitive_text

# Enough for a Python traceback's last frames plus the exception line.
_WORKER_STDERR_TAIL_CHARS = 1500


def format_run_error(exc: BaseException) -> str:
"""Retain chained causes without capturing locals or exposing URL credentials."""
traceback_text = redact_sensitive_text(
"".join(format_exception(exc)), force=True, redact_url_credentials=True,
)
return f"## Error\n\n```\n{traceback_text}\n```\n"


def external_worker_stderr_tail(path: Path) -> str:
"""Redacted tail of a restart-safe cron worker's captured stderr, as a message suffix.

The gateway used to spawn the worker with ``stderr=DEVNULL``, so a worker that died before
publishing its acknowledgement (an import error, a missing module in the interpreter it was
handed) only ever reported ``exit 1`` and the reporter had to guess the cause (#112729).
Returns an empty string when nothing was captured.
"""
try:
text = path.read_text(encoding="utf-8", errors="replace").strip()
except OSError:
return ""
if not text:
return ""
tail = text[-_WORKER_STDERR_TAIL_CHARS:]
if len(text) > _WORKER_STDERR_TAIL_CHARS:
tail = "…" + tail
tail = redact_sensitive_text(tail, force=True, redact_url_credentials=True)
return f"; worker stderr: {tail}"
29 changes: 28 additions & 1 deletion tests/cron/test_restart_safe_worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,9 @@ def test_external_worker_adopts_execution_and_runs_payload_once(
import cron.scheduler as scheduler

payload = tmp_path / "payload.json"
ack = tmp_path / "ready.json"
ack = tmp_path / "exec-1.ready"
stderr_capture = tmp_path / "exec-1.stderr"
stderr_capture.write_text("", encoding="utf-8")
payload.write_text(
json.dumps({
"job": {"id": "job-1", "execution_id": "exec-1"},
Expand Down Expand Up @@ -187,6 +189,9 @@ def test_external_worker_adopts_execution_and_runs_payload_once(
assert observed_homes == [expected_home, expected_home]
assert ack.exists()
assert not payload.exists()
# Post-ack the worker owns the stderr capture: a gateway that restarted mid-run
# would otherwise leave one orphan per surviving run.
assert not stderr_capture.exists()


def test_external_worker_refuses_to_run_without_durable_ownership(
Expand Down Expand Up @@ -407,6 +412,28 @@ def wait(self, timeout=None):
assert scheduler._running_worker_pids == {"job-cold": 4321}


def test_worker_dying_before_ack_names_its_stderr_cause(tmp_path, monkeypatch):
"""A worker that exits before acknowledging used to report only ``exit 1`` because its stderr
went to DEVNULL (#112729); the dispatch error must carry the worker's own traceback and the
capture file must not outlive the attempt."""
import cron.scheduler as scheduler
from tools.process_registry import GatewayChildDispatch

monkeypatch.setattr(scheduler, "_get_hermes_home", lambda: tmp_path)
monkeypatch.setattr(scheduler, "mark_execution_handoff_pending", lambda execution_id: {"id": execution_id})
monkeypatch.setattr(
"tools.process_registry.restart_safe_gateway_child_argv",
lambda command, *, unit_suffix, require_restart_safe_scope=False: GatewayChildDispatch(
"direct", [sys.executable, "-c", "import cron_module_that_does_not_exist"]),
)

with pytest.raises(RuntimeError) as excinfo:
scheduler._launch_external_cron_worker({"id": "job-1", "execution_id": "exec-1", "prompt": "work"})
assert "exit 1" in str(excinfo.value)
assert "No module named 'cron_module_that_does_not_exist'" in str(excinfo.value)
assert not list((tmp_path / "cron" / "external-workers").glob("exec-1.*"))


def test_external_worker_exit_rechecks_exact_execution_before_failure(monkeypatch):
import cron.scheduler as scheduler

Expand Down
Loading