From 9b26dcb9fb1f5425b198495b655d7ef54304da3f Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 4 Sep 2026 09:49:16 +0700 Subject: [PATCH 1/8] =?UTF-8?q?fix(cron):=20tri-state=20fire-claim=20heart?= =?UTF-8?q?beat=20=E2=80=94=20fence=20contention=20is=20not=20ownership=20?= =?UTF-8?q?loss=20(c-027)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit heartbeat_fire_claim() collapsed fire-fence acquisition failure into False. During slow fenced delivery the worker's own heartbeat thread cannot acquire the same process-local RLock, so a successfully delivering run was misclassified as ownership loss and terminalled as 'Interrupted by shutdown before terminal completion.' (TrustMRR exec 4219b48d while RSI recovery exec f502e929 was settling). Tri-state contract: - True = renewed/confirmed owner - False = authoritatively inspected: owner mismatch / absent claim - None = fence unavailable/unconfirmed (contention or store error) - initial validation stays fail-closed: run starts only on exactly True - heartbeat loop: None rides the existing grace window (180s prod vs 30s fence wait); immediate cancel only on explicit False - post-run probes (_fire_claim_ownership_lost, both interrupted blocks, terminal owner-CAS read): None can never adjudicate a confirmed loss; uncertain outcomes get distinct ledger errors - mark_job_run also tri-state: None = fence unavailable (CAS never ran), False stays reserved for confirmed owner mismatch - _side_effect_fence unchanged: still the exactly-once save/delivery barrier Tests (all written red-first, verified failing on unfixed code): slow_owned_delivery (parent handoff), sustained fence-contention grace, pre/post-delivery None probes, grace-exhausted None probe, terminal write None, mark_job_run fence-unavailable None, RSI recovery dispatch fails closed during original's fenced delivery, TrustMRR slow-owned delivery terminal success. Managed-gateway restart E2E (systemd scope) passes on host: one gateway, single side effect, single delivery. --- cron/jobs.py | 43 +- cron/scheduler.py | 149 +++++-- tests/cron/test_claim_job_for_fire.py | 29 ++ tests/cron/test_script_claim_heartbeat.py | 498 ++++++++++++++++++++++ 4 files changed, 675 insertions(+), 44 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 385d747619e8..5ea25cc4c7c3 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -3040,10 +3040,24 @@ def mark_job_run( status: Optional[str] = None, *, expected_fire_owner: Optional[str] = None, -) -> bool: +) -> Optional[bool]: + """Tri-state terminal write (c-027). + + Returns: + ``True`` — recorded; when ``expected_fire_owner`` is set, the + owner CAS matched and the completion is authoritative. + ``False`` — authoritatively rejected: the fire fence was acquired + and the owner CAS (when supplied) found a replacement + owner, so the stale completion was discarded. + ``None`` — unconfirmed: the fire fence could not be acquired, so + the write (and the CAS) never happened. Callers must + not treat this as ownership loss — a worker's own + fenced side effect may be the reason the fence was + busy. + """ with _fire_job_lock(job_id) as acquired: if not acquired: - return False + return None return _mark_job_run_locked( job_id, success, @@ -3785,17 +3799,36 @@ def _sweep_completed_oneshots( return removed -def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> bool: +def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> Optional[bool]: + """Tri-state fire-claim heartbeat (correction c-027). + + Returns: + ``True`` — renewed/confirmed: we hold the fire fence and the + persisted claim still names ``expected_owner``. + ``False`` — authoritatively inspected: the fence was acquired and + the claim is absent or owned by someone else. + ``None`` — unconfirmed: the fire fence could not be acquired (its + process-local RLock is contended, e.g. held across a + slow side-effect delivery) so ownership was never + actually inspected. + + Collapsing ``None`` into ``False`` made a worker's own heartbeat thread + report authoritative ownership loss during fenced delivery and cancel a + successfully completing run. Callers must treat the three outcomes + differently: only an explicit ``False`` is an ownership loss. + """ with _fire_job_lock(job_id) as acquired: if not acquired: - return False + return None return _heartbeat_fire_claim_locked( job_id, expected_owner=expected_owner, ) -def _heartbeat_fire_claim_locked(job_id: str, *, expected_owner: str) -> bool: +def _heartbeat_fire_claim_locked( + job_id: str, *, expected_owner: str +) -> Optional[bool]: """Refresh an active ``fire_claim`` without extending another owner's lease. A cron execution can legitimately outlive the fire-claim TTL. The shared diff --git a/cron/scheduler.py b/cron/scheduler.py index 267805a8f1a9..e78592c4d495 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -7321,8 +7321,12 @@ def _finish_unstarted(error: str) -> None: ) try: + # Tri-state (c-027): True=renewed/owned; False=confirmed loss; + # None=unconfirmed (fence contended). Initial validation stays + # fail-closed: the run may only start on exactly True. owns_fire_claim = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: + owns_fire_claim = None logger.warning( "Job '%s': initial fire_claim validation failed", job_id, @@ -7333,44 +7337,64 @@ def _finish_unstarted(error: str) -> None: ) return True - if owns_fire_claim is False: - logger.warning( - "Job '%s': fire claim ownership was already lost before execution", - job_id, - ) - _finish_unstarted("Fire claim ownership lost before execution started.") + if owns_fire_claim is not True: + # Fail closed (c-027): only an exactly-True renewal lets the run + # start. An explicit False is a confirmed loss; None (or an + # unexpected value) is unconfirmed ownership — the run must not + # begin without a validated claim. + if owns_fire_claim is False: + logger.warning( + "Job '%s': fire claim ownership was already lost before execution", + job_id, + ) + _finish_unstarted("Fire claim ownership lost before execution started.") + else: + logger.warning( + "Job '%s': fire claim ownership could not be confirmed before " + "execution; refusing to start", + job_id, + ) + _finish_unstarted( + "Fire claim ownership could not be validated before execution started." + ) return True def _heartbeat_loop() -> None: last_confirmed = time.monotonic() while not stop.wait(_RUN_CLAIM_HEARTBEAT_SECONDS): try: - if not heartbeat_fire_claim(job_id, expected_owner=owner): - lost_ownership.set() - logger.warning( - "Job '%s': fire claim ownership lost; interrupting stale run", - job_id, - ) - return - last_confirmed = time.monotonic() + # Tri-state (c-027): only an explicit False is an + # authoritative ownership loss. None means the fire fence + # was busy (e.g. our own delivery holds it) — ownership was + # not inspected, so the renewal is merely unconfirmed and + # falls under the grace window like any other exception. + renewed = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: - logger.debug( - "Job '%s': fire_claim heartbeat failed", + renewed = None + if renewed is False: + lost_ownership.set() + logger.warning( + "Job '%s': fire claim ownership lost; interrupting stale run", job_id, - exc_info=True, ) - if ( - time.monotonic() - last_confirmed - >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS - ): - lost_ownership.set() - logger.warning( - "Job '%s': fire_claim could not be renewed within %.1fs; " - "interrupting uncertain run", - job_id, - _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, - ) - return + return + if renewed is True: + last_confirmed = time.monotonic() + continue + logger.debug( + "Job '%s': fire_claim heartbeat unconfirmed (fence contended " + "or store error)", + job_id, + ) + if time.monotonic() - last_confirmed >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS: + lost_ownership.set() + logger.warning( + "Job '%s': fire_claim could not be renewed within %.1fs; " + "interrupting uncertain run", + job_id, + _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, + ) + return heartbeat_thread = threading.Thread( target=heartbeat_context.run, @@ -7525,7 +7549,15 @@ def _fire_claim_ownership_lost() -> bool: if fire_owner is None: return False try: - if heartbeat_fire_claim(job["id"], expected_owner=fire_owner): + # Tri-state (c-027): True = still owned; False = authoritative + # loss; None = fence contended (likely our own side effect + # holds it) — NOT a loss. None keeps the caller in + # unknown-but-not-lost territory; the fence acquisition below + # re-validates ownership before the terminal write. + renewed = heartbeat_fire_claim(job["id"], expected_owner=fire_owner) + if renewed is True: + return False + if renewed is None: return False except Exception: logger.debug( @@ -7658,9 +7690,15 @@ def _fire_claim_ownership_lost() -> bool: # would leave fire_claim lingering until TTL and last_status # stale. Probe ownership once; if still ours, record the # interruption through the owner-fenced terminal write. - if fire_owner is not None and heartbeat_fire_claim( - job["id"], expected_owner=fire_owner, - ): + # Tri-state (c-027): a None probe is unconfirmed — never + # adjudicate "ownership lost" on it; the None arm falls through + # to the uncertain-store recording path below. + probe = ( + heartbeat_fire_claim(job["id"], expected_owner=fire_owner) + if fire_owner is not None + else False + ) + if probe is True: mark_job_run( job["id"], False, @@ -7676,7 +7714,14 @@ def _fire_claim_ownership_lost() -> bool: finish_execution( execution_id, success=False, - error="Fire claim ownership lost; stale result was discarded.", + error=( + "Fire claim ownership lost; stale result was discarded." + if probe is False + else ( + "Fire claim could not be re-validated after " + "interruption; outcome uncertain." + ) + ), ) return True @@ -7849,9 +7894,14 @@ def _fire_claim_ownership_lost() -> bool: # Same transport-cancel distinction as the pre-side-effect path: # if WE still own the claim, record the interruption instead of # discarding silently (lingering claim + stale last_status). - if fire_owner is not None and heartbeat_fire_claim( - job["id"], expected_owner=fire_owner, - ): + # Tri-state (c-027): a None probe is unconfirmed — never + # adjudicate "ownership lost" on it. + probe = ( + heartbeat_fire_claim(job["id"], expected_owner=fire_owner) + if fire_owner is not None + else False + ) + if probe is True: mark_job_run( job["id"], False, @@ -7867,7 +7917,14 @@ def _fire_claim_ownership_lost() -> bool: finish_execution( execution_id, success=False, - error="Fire claim ownership lost; stale result was discarded.", + error=( + "Fire claim ownership lost; stale result was discarded." + if probe is False + else ( + "Fire claim could not be re-validated after " + "interruption; outcome uncertain." + ) + ), ) return True @@ -7910,13 +7967,27 @@ def _fire_claim_ownership_lost() -> bool: if blocked_config: mark_kwargs["status"] = "blocked_config" marked = mark_job_run(job["id"], success, error, **mark_kwargs) - if fire_owner is not None and not marked: + # Tri-state (c-027): only an explicit False is an authoritative + # owner-CAS rejection. None means the fire fence was unavailable — + # the CAS never ran, so ownership was NOT confirmed lost; record + # the unconfirmed-terminal outcome instead of a confirmed loss. + if fire_owner is not None and marked is False: finish_execution( execution_id, success=False, error="Fire claim ownership lost before terminal completion.", ) return True + if fire_owner is not None and marked is None: + finish_execution( + execution_id, + success=False, + error=( + "Fire claim fence unavailable before terminal completion; " + "outcome uncertain." + ), + ) + return True normalized_deliver = _normalize_deliver_value( _delivery_lane_value(job, for_failure=not success) ) diff --git a/tests/cron/test_claim_job_for_fire.py b/tests/cron/test_claim_job_for_fire.py index 200199efd7e6..c70d3cea77e2 100644 --- a/tests/cron/test_claim_job_for_fire.py +++ b/tests/cron/test_claim_job_for_fire.py @@ -7,6 +7,7 @@ These exercise the real store against a temp HERMES_HOME (no mocks) per the E2E-over-mocks discipline for file-touching code. """ +import contextlib import threading import time @@ -263,3 +264,31 @@ def reentrant_claimant(): assert result == {"outer": True, "inner": True} thread.join(timeout=2) assert thread.is_alive() is False + + +def test_mark_job_run_returns_none_when_fence_unavailable(temp_home, monkeypatch): + """Fence-unavailable is unconfirmed, not an authoritative CAS loss (c-027). + + ``False`` stays reserved for a confirmed owner mismatch; a contended + fire fence must return ``None`` so callers cannot mistake lock + contention for ownership loss. + """ + import cron.jobs as jobs + + job = jobs.create_job(prompt="x", schedule="every 5m", name="fence none") + assert jobs.claim_job_for_fire(job["id"]) is True + owner = jobs.get_job(job["id"])["fire_claim"]["by"] + + @contextlib.contextmanager + def contended_fence(*_args, **_kwargs): + yield False + + monkeypatch.setattr(jobs, "_fire_job_lock", contended_fence) + + assert jobs.mark_job_run( + job["id"], True, expected_fire_owner=owner + ) is None + + persisted = jobs.get_job(job["id"]) + assert persisted["fire_claim"]["by"] == owner + assert persisted.get("last_run_at") is None diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index 11fb4501b02c..5b7db0ed18fb 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -354,6 +354,81 @@ def _blocking_body(job, **kwargs): assert refreshed["by"] == original_claim["by"] +def test_slow_owned_delivery_does_not_false_interrupt_completed_run( + tmp_path, monkeypatch +): + """An owner-fenced delivery may outlast one heartbeat lock attempt. + + The worker still owns the unchanged fire claim throughout. A transient + failure to acquire its own delivery fence must not become ownership loss. + """ + import cron.executions as executions + import cron.jobs as jobs + import cron.scheduler as scheduler + + profile_home = tmp_path / "profile" + profile_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", profile_home / "cron" / "executions.db" + ) + + with jobs.use_cron_store(profile_home): + job = jobs.create_job(prompt="x", schedule="every 5m", name="slow delivery") + assert jobs.claim_job_for_fire(job["id"]) is True + claimed_job = jobs.get_job(job["id"]) + assert claimed_job is not None + execution = executions.create_execution(job["id"], source="builtin") + claimed_job["execution_id"] = execution["id"] + + monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", execution["id"]) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_args, **_kwargs: (True, "output", "completed", None), + ) + monkeypatch.setattr( + scheduler, "save_job_output", lambda *_args, **_kwargs: tmp_path / "output.md" + ) + + false_heartbeat_seen = threading.Event() + real_heartbeat = jobs.heartbeat_fire_claim + + expected_owner = claimed_job["fire_claim"]["by"] + + def _observed_heartbeat(job_id: str, *, expected_owner: str) -> bool: + renewed = real_heartbeat(job_id, expected_owner=expected_owner) + if not renewed: + false_heartbeat_seen.set() + return renewed + + def _slow_delivery(*_args, **_kwargs): + assert false_heartbeat_seen.wait(timeout=2) + still_owned = jobs.get_job(job["id"]) + assert still_owned is not None + assert still_owned["fire_claim"]["by"] == expected_owner + return None + + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", _observed_heartbeat) + monkeypatch.setattr(scheduler, "_deliver_result", _slow_delivery) + + assert scheduler.run_one_job(claimed_job) is True + + finished = executions.get_execution(execution["id"]) + persisted_job = jobs.get_job(job["id"]) + assert finished is not None + assert persisted_job is not None + + assert ( + finished["status"], + finished["error"], + persisted_job["last_status"], + ) == ("completed", None, "ok") + + def test_lost_fire_claim_stops_stale_delivery(monkeypatch): """A runner that loses its durable owner must not deliver its stale result.""" import cron.scheduler as scheduler @@ -586,3 +661,426 @@ def owned_fence(*_args, **_kwargs): success=False, error="Fire claim ownership lost before terminal completion.", ) + + +def test_heartbeat_lock_contention_is_unconfirmed_not_ownership_loss(monkeypatch): + """Fence contention (None) is unconfirmed ownership, not a loss (c-027). + + During slow fenced delivery the worker's own heartbeat thread cannot + acquire the process-local fire fence. That must not become an + authoritative ownership-loss cancellation: None rides the grace window + while only an explicit False interrupts immediately. + """ + import cron.scheduler as scheduler + + calls = [] + + def heartbeat(*_args, **_kwargs): + # Initial validation renews (True); afterwards the fence stays + # contended for the whole run: never False, never raises. + calls.append("none" if calls else "true") + return True if len(calls) == 1 else None + + ran = threading.Event() + + def run_body(_job, **kwargs): + # Far beyond several heartbeat ticks, well under the grace window. + assert not kwargs["fire_claim_lost"].wait(timeout=0.08) + ran.set() + return True + + job = { + "id": "fence-contended", + "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner"}, + } + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) + monkeypatch.setattr(scheduler, "_run_one_job_body", run_body) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + + assert scheduler.run_one_job(job) is True + + assert ran.is_set() + assert calls, "heartbeat loop never probed the fire claim" + + +def test_pre_delivery_ownership_probe_treats_none_as_not_lost(tmp_path, monkeypatch): + """A contended (None) post-run probe must not discard a completed run (c-027). + + Between run_job returning and delivery, ``_fire_claim_ownership_lost()`` + probes the claim. A None probe (fence contended, likely by our own + fenced save) is not a loss: delivery must proceed and the run completes. + """ + import cron.executions as executions + import cron.jobs as jobs + import cron.scheduler as scheduler + + profile_home = tmp_path / "profile" + profile_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", profile_home / "cron" / "executions.db" + ) + + with jobs.use_cron_store(profile_home): + job = jobs.create_job(prompt="x", schedule="every 5m", name="probe none") + assert jobs.claim_job_for_fire(job["id"]) is True + claimed_job = jobs.get_job(job["id"]) + execution = executions.create_execution(job["id"], source="builtin") + claimed_job["execution_id"] = execution["id"] + + monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", execution["id"]) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_args, **_kwargs: (True, "output", "done", None), + ) + monkeypatch.setattr( + scheduler, "save_job_output", lambda *_args, **_kwargs: "output.md" + ) + + real_probe = jobs.heartbeat_fire_claim + probes = [] + + def _contended_probe(job_id: str, *, expected_owner: str): + outcome = real_probe(job_id, expected_owner=expected_owner) + probes.append(outcome) + # First probe (initial validation) is clean; post-run probes hit + # a contended fence. + return True if len(probes) == 1 else None + + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", _contended_probe) + delivered = [] + + def _deliver(*_args, **_kwargs): + delivered.append(True) + return None + + monkeypatch.setattr(scheduler, "_deliver_result", _deliver) + + assert scheduler.run_one_job(claimed_job) is True + + assert delivered, "completed run was not delivered" + + finished = executions.get_execution(execution["id"]) + persisted_job = jobs.get_job(job["id"]) + assert finished is not None and persisted_job is not None + assert ( + finished["status"], + finished["error"], + persisted_job["last_status"], + ) == ("completed", None, "ok") + + +def test_post_delivery_ownership_probe_treats_none_as_not_lost(tmp_path, monkeypatch): + """A contended (None) probe after delivery must not overwrite the result (c-027).""" + import cron.executions as executions + import cron.jobs as jobs + import cron.scheduler as scheduler + + profile_home = tmp_path / "profile" + profile_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", profile_home / "cron" / "executions.db" + ) + + with jobs.use_cron_store(profile_home): + job = jobs.create_job(prompt="x", schedule="every 5m", name="post none") + assert jobs.claim_job_for_fire(job["id"]) is True + claimed_job = jobs.get_job(job["id"]) + execution = executions.create_execution(job["id"], source="builtin") + claimed_job["execution_id"] = execution["id"] + + monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", execution["id"]) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_args, **_kwargs: (True, "output", "done", None), + ) + monkeypatch.setattr( + scheduler, "save_job_output", lambda *_args, **_kwargs: "output.md" + ) + + real_probe = jobs.heartbeat_fire_claim + probes = [] + + def _contended_probe(job_id: str, *, expected_owner: str): + outcome = real_probe(job_id, expected_owner=expected_owner) + probes.append(outcome) + return True if len(probes) == 1 else None + + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", _contended_probe) + + def _deliver(*_args, **_kwargs): + return None + + monkeypatch.setattr(scheduler, "_deliver_result", _deliver) + + assert scheduler.run_one_job(claimed_job) is True + + finished = executions.get_execution(execution["id"]) + persisted_job = jobs.get_job(job["id"]) + assert finished is not None and persisted_job is not None + assert ( + finished["status"], + finished["error"], + persisted_job["last_status"], + ) == ("completed", None, "ok") + + +def test_interrupted_path_none_probe_does_not_claim_confirmed_loss(monkeypatch): + """Grace-exhausted stop + contended probe must not claim a confirmed loss (c-027).""" + import cron.scheduler as scheduler + + calls = [] + + def heartbeat(*_args, **_kwargs): + calls.append(1) + return True if len(calls) == 1 else None + + def _run_job(job, *, defer_agent_teardown=None, extra_prompt=None, + cancel_event=None, execution_id=None): + # Sustained None heartbeats exhaust the grace window -> lost_ownership + # set as "uncertain, stop". Never an explicit False. + assert cancel_event is not None + assert cancel_event.wait(timeout=1.0) + return True, "output", "response", None + + finish = MagicMock() + mark = MagicMock(return_value=True) + job = { + "id": "grace-none", + "execution_id": "grace-none-exec", + "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner"}, + } + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) + monkeypatch.setattr(scheduler, "run_job", _run_job) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.03) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) + monkeypatch.setattr(scheduler, "mark_job_run", mark) + monkeypatch.setattr(scheduler, "finish_execution", finish) + + with patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"): + assert scheduler.run_one_job(job) is True + + errors = [call.kwargs.get("error") for call in finish.call_args_list] + assert errors, "execution ledger never closed" + for err in errors: + assert "ownership lost" not in err.lower(), ( + f"None probe adjudicated as confirmed ownership loss: {err!r}" + ) + + +def test_terminal_write_none_does_not_mark_ownership_lost(monkeypatch): + """An unconfirmed (None) terminal write must not be read as CAS loss (c-027).""" + import cron.scheduler as scheduler + + @contextlib.contextmanager + def owned_fence(*_args, **_kwargs): + yield True + + finish = MagicMock() + job = { + "id": "terminal-none", + "execution_id": "terminal-none-exec", + "name": "terminal-none", + "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner"}, + } + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", lambda *a, **k: True) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_a, **_k: (True, "output", "response", None), + ) + monkeypatch.setattr(scheduler, "fire_claim_fence", owned_fence, raising=False) + monkeypatch.setattr(scheduler, "save_job_output", lambda *_a: "output.md") + monkeypatch.setattr(scheduler, "_deliver_result", lambda *_a, **_k: None) + # Terminal write hits a contended fence -> None (unconfirmed, not a CAS loss). + monkeypatch.setattr(scheduler, "mark_job_run", lambda *_a, **_k: None) + monkeypatch.setattr(scheduler, "finish_execution", finish) + + with patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"): + assert scheduler.run_one_job(job) is True + + for call in finish.call_args_list: + assert call.kwargs.get("error") != ( + "Fire claim ownership lost before terminal completion." + ), "unconfirmed terminal write adjudicated as ownership loss" + + +def test_recovery_execution_fails_closed_while_original_delivery_settles( + tmp_path, monkeypatch +): + """RSI-loop case (c-027): a recovery dispatch of the same job while the + original holds the fence in slow delivery must (a) fail closed without + starting a duplicate run and (b) not disturb the original's claim. + """ + import cron.executions as executions + import cron.jobs as jobs + import cron.scheduler as scheduler + + profile_home = tmp_path / "profile" + profile_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", profile_home / "cron" / "executions.db" + ) + + with jobs.use_cron_store(profile_home): + job = jobs.create_job(prompt="x", schedule="every 5m", name="rsi loop") + assert jobs.claim_job_for_fire(job["id"]) is True + claimed_job = jobs.get_job(job["id"]) + original_owner = claimed_job["fire_claim"]["by"] + execution = executions.create_execution(job["id"], source="builtin") + claimed_job["execution_id"] = execution["id"] + + monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", execution["id"]) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_args, **_kwargs: (True, "output", "done", None), + ) + monkeypatch.setattr( + scheduler, "save_job_output", lambda *_args, **_kwargs: "output.md" + ) + + in_delivery = threading.Event() + release_delivery = threading.Event() + + def _slow_delivery(*_args, **_kwargs): + in_delivery.set() + assert release_delivery.wait(timeout=5) + return None + + monkeypatch.setattr(scheduler, "_deliver_result", _slow_delivery) + # Short fence timeout so the recovery dispatch's contended validation + # resolves as None quickly instead of blocking on the real 30s wait. + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.02) + + runner_done = threading.Event() + runner_result = [] + + def _original_runner(): + runner_result.append(scheduler.run_one_job(dict(claimed_job))) + runner_done.set() + + runner = threading.Thread(target=_original_runner, daemon=True) + runner.start() + assert in_delivery.wait(timeout=5) + + # A recovery execution of the SAME job (new execution id, no + # external-worker marker) attempts to start while the fence is held. + monkeypatch.delenv("_HERMES_CRON_EXTERNAL_WORKER", raising=False) + recovery_execution = executions.create_execution(job["id"], source="builtin") + recovery_job = dict(claimed_job) + recovery_job["execution_id"] = recovery_execution["id"] + recovery_job["fire_claim"] = dict(claimed_job["fire_claim"]) + + recovery_ran = [] + + def _must_not_run(*_args, **_kwargs): + recovery_ran.append(True) + return True, "duplicate", "duplicate", None + + monkeypatch.setattr(scheduler, "run_job", _must_not_run) + + assert scheduler.run_one_job(recovery_job) is True + assert not recovery_ran, "recovery dispatch ran a duplicate side effect" + recovery_row = executions.get_execution(recovery_execution["id"]) + assert recovery_row is not None + assert recovery_row["status"] == "failed" + + # The original must be untouched and still complete its delivery. + current = jobs.get_job(job["id"]) + assert current is not None + assert current["fire_claim"]["by"] == original_owner + + release_delivery.set() + assert runner_done.wait(timeout=5) + assert runner_result == [True] + + finished = executions.get_execution(execution["id"]) + persisted_job = jobs.get_job(job["id"]) + assert finished is not None and persisted_job is not None + assert ( + finished["status"], + finished["error"], + persisted_job["last_status"], + ) == ("completed", None, "ok") + + +def test_terminal_write_after_slow_owned_delivery_succeeds(tmp_path, monkeypatch): + """TrustMRR case (c-027): the terminal write runs after fenced delivery + releases and must record the successful run, not a false loss. + """ + import cron.executions as executions + import cron.jobs as jobs + import cron.scheduler as scheduler + + profile_home = tmp_path / "profile" + profile_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", profile_home / "cron" / "executions.db" + ) + + with jobs.use_cron_store(profile_home): + job = jobs.create_job(prompt="x", schedule="every 5m", name="trustmrr") + assert jobs.claim_job_for_fire(job["id"]) is True + claimed_job = jobs.get_job(job["id"]) + execution = executions.create_execution(job["id"], source="builtin") + claimed_job["execution_id"] = execution["id"] + + monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", execution["id"]) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_args, **_kwargs: (True, "output", "done", None), + ) + monkeypatch.setattr( + scheduler, "save_job_output", lambda *_args, **_kwargs: "output.md" + ) + + def _slow_delivery(*_args, **_kwargs): + # Hold the fence past several heartbeat intervals so the real + # heartbeat thread contends on the lock the delivery itself + # holds (the production shape). + time.sleep(0.08) + return None + + monkeypatch.setattr(scheduler, "_deliver_result", _slow_delivery) + + mark_calls = [] + + real_mark = jobs.mark_job_run + + def _observed_mark(job_id, success, error=None, **kwargs): + result = real_mark(job_id, success, error, **kwargs) + mark_calls.append((success, error, result)) + return result + + monkeypatch.setattr(scheduler, "mark_job_run", _observed_mark) + + assert scheduler.run_one_job(claimed_job) is True + + finished = executions.get_execution(execution["id"]) + persisted_job = jobs.get_job(job["id"]) + assert finished is not None and persisted_job is not None + assert finished["status"] == "completed" + assert persisted_job["last_status"] == "ok" + terminal = [call for call in mark_calls if call[2] is not None] + assert terminal and terminal[-1][0] is True, ( + f"terminal write never recorded success: {mark_calls!r}" + ) From 51e201b67397fb596cab6af88d5ac799f8d5122e Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 4 Sep 2026 10:57:18 +0700 Subject: [PATCH 2/8] [verified] fix(cron): fence tri-state interruption delivery --- cron/scheduler.py | 206 +++++++++++++++------- tests/cron/test_script_claim_heartbeat.py | 51 ++++++ tests/cron/test_shutdown_interrupt.py | 136 +++++++++++++- 3 files changed, 323 insertions(+), 70 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index e78592c4d495..8a6724f8ecbf 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -829,6 +829,10 @@ def _is_cron_silence_response(text: str) -> bool: # every fire) must not inherit the stale flag. Legacy dispatch paths without # a registered fire owner fall back to storing the bare job ID. _interrupted_job_ids: set = set() +# Executions observed during shutdown but whose owner-fenced terminal write has +# not yet succeeded. Pending tokens suppress normal-result delivery, but do +# not authorize the worker to skip its own terminal write. +_interrupting_job_ids: set = set() class _CancelEventLike(Protocol): @@ -1277,10 +1281,13 @@ def mark_running_jobs_interrupted( same process and may go on to produce a plausible-looking final response from the now-truncated tool output. - Records the job IDs in ``_interrupted_job_ids`` BEFORE writing - ``last_status`` so ``run_one_job``'s own eventual completion for the - same job (racing in its own thread) sees the flag and skips its normal - write instead of clobbering this one — see the check near the end of + Records execution tokens in ``_interrupting_job_ids`` BEFORE writing + ``last_status`` so ``run_one_job`` suppresses a plausible normal result + while shutdown races its thread. A successful terminal write promotes + the token to ``_interrupted_job_ids``; only that confirmed state lets the + worker skip its own terminal write. If the write is unavailable, the + pending token still suppresses delivery but leaves the worker responsible + for eventually persisting terminal state. See the checks near the end of ``run_one_job``. This does not attempt to correlate the killed subprocess PID to a specific job ID (the process registry tracks PIDs, not cron job IDs); any job still dispatched at the moment of a forced @@ -1319,35 +1326,47 @@ def mark_running_jobs_interrupted( _running_job_ids - registered_ids - restart_safe_waiters ) ) - _interrupted_job_ids.update( + _interrupting_job_ids.update( token if token is not None else job_id for token, job_id, _owner, _profile_home in active_fires ) marked = [] - for _token, job_id, fire_owner, profile_home in active_fires: + for token, job_id, fire_owner, profile_home in active_fires: + interrupt_key = token if token is not None else job_id if not fire_owner: logger.warning( "Job '%s' interrupted before its durable fire owner was registered; " "leaving persisted state untouched", job_id, ) - # Still report the interruption to the caller: the gateway - # shutdown path uses the returned IDs to send the - # interrupted-cron notice while adapters are still connected - # (#82232). The in-memory interrupt flag WAS recorded above — - # only the persisted last_status write is skipped here. + # Report the interruption to the caller, but leave the token + # pending rather than confirmed: the worker must still perform + # its own terminal write because shutdown did not persist one. marked.append(job_id) continue try: with use_cron_store(profile_home): - if mark_job_run( + persisted = mark_job_run( job_id, False, reason, expected_fire_owner=fire_owner, - ): - marked.append(job_id) + ) + with _running_lock: + if persisted is True: + _interrupting_job_ids.discard(interrupt_key) + _interrupted_job_ids.add(interrupt_key) + elif persisted is False: + # Authoritative owner loss: no write by this worker is + # allowed or needed. A pending worker-side write is still + # fenced by its stale expected owner. + _interrupting_job_ids.discard(interrupt_key) + if persisted is True: + marked.append(job_id) except Exception as e: + # Keep the token pending. This continues to suppress delivery, + # while _consume_interrupted_flag returns False so the worker + # retries the terminal write rather than assuming this one landed. logger.warning("Failed to mark job %s interrupted: %s", job_id, e) return marked @@ -1369,9 +1388,14 @@ def _is_interrupted(job_id: str, token: Optional[object] = None) -> bool: for legacy dispatch paths with no registered fire owner. """ with _running_lock: - if token is not None and token in _interrupted_job_ids: + if token is not None and ( + token in _interrupted_job_ids or token in _interrupting_job_ids + ): return True - return job_id in _interrupted_job_ids + return ( + job_id in _interrupted_job_ids + or job_id in _interrupting_job_ids + ) def _consume_interrupted_flag(job_id: str, token: Optional[object] = None) -> bool: @@ -1379,18 +1403,23 @@ def _consume_interrupted_flag(job_id: str, token: Optional[object] = None) -> bo THIS execution interrupted (see ``mark_running_jobs_interrupted``). Called by ``run_one_job`` right before it would otherwise write its own - ``last_status``. Consuming (discarding) rather than just checking keeps - the flag from leaking across a later, unrelated run of the same job ID - (recurring jobs reuse their ID every fire).""" + ``last_status``. Return True only for a confirmed flag whose shutdown + terminal write succeeded. A pending flag still gets cleared, but returns + False so the worker remains responsible for durable terminal state. + Consuming rather than just checking keeps either flag from leaking across + a later, unrelated run of the same recurring job ID.""" with _running_lock: - hit = False - if token is not None and token in _interrupted_job_ids: - _interrupted_job_ids.discard(token) - hit = True + confirmed = False + if token is not None: + if token in _interrupted_job_ids: + _interrupted_job_ids.discard(token) + confirmed = True + _interrupting_job_ids.discard(token) if job_id in _interrupted_job_ids: _interrupted_job_ids.discard(job_id) - hit = True - return hit + confirmed = True + _interrupting_job_ids.discard(job_id) + return confirmed def _inactivity_watchdog_loop( @@ -7856,26 +7885,54 @@ def _fire_claim_ownership_lost() -> bool: ) if should_deliver: - unresolved_origin = ( - _normalize_deliver_value(_delivery_lane_value(job, for_failure=not success)) - == "origin" - and not _resolve_delivery_targets(job, for_failure=not success) - ) try: with _side_effect_fence() as owns_delivery: if not owns_delivery: raise _FireClaimLostDuringSideEffect - delivery_attempted = True - delivery_error = _deliver_result( - job, - deliver_content, - adapters=adapters, - loop=loop, - # Failure summaries (and drift/blocked-config alerts - # composed into deliver_content on the failure path) - # honor the job's failure_deliver override (NS-788). - for_failure=not success, - ) + # Shutdown publishes the pending interrupt token before + # contending for this same fence. Recheck while fenced + # to close the window after the earlier delivery-gate + # check: a plausible response must not escape merely + # because shutdown began while we were waiting here. + if success and _is_interrupted(job["id"], execution_token): + success = False + error = ( + "Interrupted by gateway shutdown before the run " + "finished (tool subprocess was killed mid-flight)." + ) + incident_acked, failure_incident_id = ( + _upsert_incident_for_failure( + job, error, output_file=output_file + ) + ) + deliver_content = ( + "" + if incident_acked + else _summarize_cron_failure_for_delivery(job, error) + + _failure_streak_nudge(job) + ) + should_deliver = bool(deliver_content.strip()) + if should_deliver: + unresolved_origin = ( + _normalize_deliver_value( + _delivery_lane_value(job, for_failure=not success) + ) + == "origin" + and not _resolve_delivery_targets( + job, for_failure=not success + ) + ) + delivery_attempted = True + delivery_error = _deliver_result( + job, + deliver_content, + adapters=adapters, + loop=loop, + # Failure summaries (and drift/blocked-config alerts + # composed into deliver_content on the failure path) + # honor the job's failure_deliver override (NS-788). + for_failure=not success, + ) except Exception as de: if isinstance(de, _FireClaimLostDuringSideEffect): raise @@ -8060,38 +8117,51 @@ def _fire_claim_ownership_lost() -> bool: delivery_outcome = "suppressed_acked" else: try: - delivery_attempted = True - delivery_error = _deliver_result( - job, - # Composed exactly like the normal failure delivery above. - # mark_job_run below records THIS run in failure_streak - # whichever layer failed, so a job that fails before the - # run body every tick builds a streak nobody is ever told - # about: its alerts only ever leave through here, and the - # nudge only ever left through there (#88655). - _summarize_cron_failure_for_delivery(job, _err_text) - + _failure_streak_nudge(job), - adapters=adapters, - loop=loop, - for_failure=True, - ) + # The ownership probe above is tri-state: None only means + # unconfirmed, never permission to emit. The side-effect + # fence is the exactly-once authority for delivery on this + # exception path, just as it is on the normal path. + with _side_effect_fence() as owns_delivery: + if owns_delivery: + delivery_attempted = True + delivery_error = _deliver_result( + job, + # Composed exactly like the normal failure delivery above. + # mark_job_run below records THIS run in failure_streak + # whichever layer failed, so a job that fails before the + # run body every tick builds a streak nobody is ever told + # about: its alerts only ever leave through here, and the + # nudge only ever left through there (#88655). + _summarize_cron_failure_for_delivery(job, _err_text) + + _failure_streak_nudge(job), + adapters=adapters, + loop=loop, + for_failure=True, + ) + else: + logger.warning( + "Job '%s': skipping exception delivery because " + "fire-claim ownership could not be fenced", + job["id"], + ) except Exception as delivery_exc: delivery_error = str(delivery_exc) logger.error( "Delivery failed for job %s: %s", job["id"], delivery_exc ) - if not delivery_error and normalized_deliver == "origin": - unresolved_origin = not _resolve_delivery_targets( - job, for_failure=True - ) - if delivery_error: - delivery_outcome = "failed" - elif unresolved_origin: - delivery_outcome = "not_configured" - elif normalized_deliver != "local": - delivery_outcome = "delivered" - if delivery_outcome in ("delivered", "not_configured"): - _mark_incident_alerted(failure_incident_id) + if delivery_attempted: + if not delivery_error and normalized_deliver == "origin": + unresolved_origin = not _resolve_delivery_targets( + job, for_failure=True + ) + if delivery_error: + delivery_outcome = "failed" + elif unresolved_origin: + delivery_outcome = "not_configured" + elif normalized_deliver != "local": + delivery_outcome = "delivered" + if delivery_outcome in ("delivered", "not_configured"): + _mark_incident_alerted(failure_incident_id) try: if not _consume_interrupted_flag(job["id"], execution_token): mark_kwargs = {} diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index 5b7db0ed18fb..f66136eb9eaf 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -1084,3 +1084,54 @@ def _observed_mark(job_id, success, error=None, **kwargs): assert terminal and terminal[-1][0] is True, ( f"terminal write never recorded success: {mark_calls!r}" ) + + +def test_none_probe_cannot_authorize_unfenced_exception_delivery( + tmp_path, monkeypatch +): + """A contended ownership probe is unconfirmed; only the side-effect + fence may authorize the exception-path failure notification.""" + import cron.scheduler as scheduler + + probes = iter([True, None]) + delivered = MagicMock(return_value=None) + fence = MagicMock( + side_effect=lambda *a, **k: contextlib.nullcontext(False) + ) + monkeypatch.setattr( + scheduler, "heartbeat_fire_claim", lambda *a, **k: next(probes) + ) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 60.0) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *a, **k: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *a, **k: {}) + monkeypatch.setattr( + scheduler, "run_job", MagicMock(side_effect=RuntimeError("run failure")) + ) + monkeypatch.setattr(scheduler, "_deliver_result", delivered) + monkeypatch.setattr(scheduler, "fire_claim_fence", fence) + monkeypatch.setattr(scheduler, "mark_job_run", lambda *a, **k: True) + monkeypatch.setattr(scheduler, "finish_execution", MagicMock()) + monkeypatch.setattr( + scheduler, "_upsert_incident_for_failure", lambda *a, **k: (False, None) + ) + monkeypatch.setattr(scheduler, "_mark_incident_alerted", lambda *a, **k: None) + monkeypatch.setattr(scheduler, "_get_hermes_home", lambda: tmp_path) + + job = { + "id": "stale-worker", + "name": "stale worker", + "execution_id": "exec-stale", + "prompt": "x", + "schedule_display": "manual", + "deliver": "telegram:123", + "fire_claim": {"by": "old-owner"}, + } + + with patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"), \ + patch("tools.terminal_scope.install_profile_terminal_scope", return_value=None): + assert scheduler.run_one_job(job) is False + + fence.assert_called_once_with("stale-worker", expected_owner="old-owner") + delivered.assert_not_called() diff --git a/tests/cron/test_shutdown_interrupt.py b/tests/cron/test_shutdown_interrupt.py index a8bb3bf28f13..c7ccd706a6c7 100644 --- a/tests/cron/test_shutdown_interrupt.py +++ b/tests/cron/test_shutdown_interrupt.py @@ -26,10 +26,12 @@ def _reset_scheduler_state(): sched._running_job_ids.clear() sched._running_fire_owners.clear() sched._interrupted_job_ids.clear() + sched._interrupting_job_ids.clear() yield sched._running_job_ids.clear() sched._running_fire_owners.clear() sched._interrupted_job_ids.clear() + sched._interrupting_job_ids.clear() class TestGetRunningJobIds: @@ -96,7 +98,7 @@ def test_marks_every_in_flight_job(self): assert "gateway shutdown" in c.args[2] assert c.kwargs["expected_fire_owner"] in {"owner-1", "owner-2"} - def test_sets_interrupted_flag_for_consumption_by_run_one_job(self): + def test_sets_pending_interrupt_for_worker_without_registered_owner(self): import cron.scheduler as sched sched._running_job_ids.add("job-1") @@ -104,7 +106,11 @@ def test_sets_interrupted_flag_for_consumption_by_run_one_job(self): with patch("cron.scheduler.mark_job_run"): sched.mark_running_jobs_interrupted("shutdown") - assert "job-1" in sched._interrupted_job_ids + assert "job-1" in sched._interrupting_job_ids + assert sched._is_interrupted("job-1") is True + # Shutdown could not persist without an owner, so the worker remains + # responsible for its own terminal write. + assert sched._consume_interrupted_flag("job-1") is False def test_one_job_marking_failure_does_not_block_the_others(self): """mark_job_run raising for one job (e.g. a jobs.json write race) @@ -131,6 +137,132 @@ def _side_effect(job_id, success, reason, **kwargs): assert marked == ["job-2"] + def test_unavailable_terminal_fence_leaves_worker_responsible(self, tmp_path): + """A None result means shutdown wrote nothing; do not install the + confirmed token that makes the worker skip its terminal write.""" + import cron.scheduler as sched + + token = object() + sched._running_job_ids.add("job-1") + sched._running_fire_owners["job-1"] = { + token: ("owner-1", tmp_path), + } + + with patch("cron.scheduler.mark_job_run", return_value=None): + marked = sched.mark_running_jobs_interrupted("shutdown") + + assert marked == [] + assert sched._is_interrupted("job-1", token) is True + assert sched._consume_interrupted_flag("job-1", token) is False + + def test_worker_terminalizes_after_shutdown_fence_contention(self, tmp_path): + """Shutdown fence contention suppresses normal output, but the worker + later records terminal state and emits at most one failure delivery.""" + import contextlib + from unittest.mock import MagicMock + + import cron.scheduler as sched + + token = object() + job = { + "id": "job-contended", + "name": "contended job", + "execution_id": "exec-contended", + "prompt": "do work", + "deliver": "telegram:123", + "fire_claim": {"by": "owner-1"}, + } + sched._running_job_ids.add(job["id"]) + sched._running_fire_owners[job["id"]] = { + token: ("owner-1", tmp_path), + } + marks = MagicMock(side_effect=[None, True]) + delivered = MagicMock(return_value=None) + + with patch("cron.scheduler.mark_job_run", marks): + assert sched.mark_running_jobs_interrupted("shutdown") == [] + with patch("cron.scheduler.claim_dispatch", return_value=True), \ + patch("cron.scheduler.mark_execution_running", return_value={}), \ + patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"), \ + patch( + "cron.scheduler.run_job", + return_value=(True, "full output", "plausible response", None), + ), \ + patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ + patch("cron.scheduler._deliver_result", delivered), \ + patch("cron.scheduler.heartbeat_fire_claim", return_value=True), \ + patch( + "cron.scheduler.fire_claim_fence", + side_effect=lambda *a, **k: contextlib.nullcontext(True), + ), \ + patch("cron.scheduler.finish_execution"): + assert sched._run_one_job_body(job, execution_token=token) is True + + assert marks.call_count == 2 + assert marks.call_args_list[1].args[:2] == (job["id"], False) + assert "Interrupted by gateway shutdown" in marks.call_args_list[1].args[2] + assert marks.call_args_list[1].kwargs["expected_fire_owner"] == "owner-1" + assert delivered.call_count == 1 + assert "plausible response" not in delivered.call_args.args[1] + + def test_shutdown_starting_at_delivery_fence_cannot_leak_normal_result(self): + """The fence-local interrupt recheck closes the race after the first + delivery-gate check but before the external side effect.""" + import contextlib + from unittest.mock import MagicMock + + import cron.scheduler as sched + + token = object() + job = { + "id": "job-delivery-race", + "name": "delivery race", + "execution_id": "exec-delivery-race", + "prompt": "do work", + "deliver": "telegram:123", + "fire_claim": {"by": "owner-1"}, + } + fence_entries = 0 + + def fence(*_args, **_kwargs): + @contextlib.contextmanager + def held(): + nonlocal fence_entries + fence_entries += 1 + if fence_entries == 2: + with sched._running_lock: + sched._interrupting_job_ids.add(token) + yield True + + return held() + + delivered = MagicMock(return_value=None) + marked = MagicMock(return_value=True) + with patch("cron.scheduler.claim_dispatch", return_value=True), \ + patch("cron.scheduler.mark_execution_running", return_value={}), \ + patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"), \ + patch( + "cron.scheduler.run_job", + return_value=(True, "full output", "plausible response", None), + ), \ + patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ + patch("cron.scheduler._deliver_result", delivered), \ + patch("cron.scheduler.heartbeat_fire_claim", return_value=True), \ + patch("cron.scheduler.fire_claim_fence", side_effect=fence), \ + patch("cron.scheduler.mark_job_run", marked), \ + patch("cron.scheduler.finish_execution"): + assert sched._run_one_job_body(job, execution_token=token) is True + + assert fence_entries == 2 + delivered.assert_called_once() + assert "plausible response" not in delivered.call_args.args[1] + assert marked.call_args.args[:2] == (job["id"], False) + assert sched._is_interrupted(job["id"], token) is False + def test_stale_shutdown_cannot_clear_replacement_owner(self, tmp_path): import cron.jobs as jobs import cron.scheduler as sched From 991095136967b0f2776d451bde02dcc05b5ad0ea Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 4 Sep 2026 11:19:09 +0700 Subject: [PATCH 3/8] [verified] test(cron): cover persisted shutdown contention --- tests/cron/test_shutdown_interrupt.py | 138 ++++++++++++++++++-------- 1 file changed, 94 insertions(+), 44 deletions(-) diff --git a/tests/cron/test_shutdown_interrupt.py b/tests/cron/test_shutdown_interrupt.py index c7ccd706a6c7..130ed7d743a0 100644 --- a/tests/cron/test_shutdown_interrupt.py +++ b/tests/cron/test_shutdown_interrupt.py @@ -155,57 +155,107 @@ def test_unavailable_terminal_fence_leaves_worker_responsible(self, tmp_path): assert sched._is_interrupted("job-1", token) is True assert sched._consume_interrupted_flag("job-1", token) is False - def test_worker_terminalizes_after_shutdown_fence_contention(self, tmp_path): - """Shutdown fence contention suppresses normal output, but the worker - later records terminal state and emits at most one failure delivery.""" - import contextlib + def test_worker_terminalizes_after_shutdown_fence_contention( + self, tmp_path, monkeypatch + ): + """A shutdown write blocked by the real fire fence leaves the worker + responsible for durable terminal state and at-most-once delivery.""" from unittest.mock import MagicMock + import cron.executions as executions + import cron.jobs as jobs import cron.scheduler as sched - token = object() - job = { - "id": "job-contended", - "name": "contended job", - "execution_id": "exec-contended", - "prompt": "do work", - "deliver": "telegram:123", - "fire_claim": {"by": "owner-1"}, - } - sched._running_job_ids.add(job["id"]) - sched._running_fire_owners[job["id"]] = { - token: ("owner-1", tmp_path), - } - marks = MagicMock(side_effect=[None, True]) - delivered = MagicMock(return_value=None) + profile_home = tmp_path / "profile" + monkeypatch.setenv("HERMES_HOME", str(profile_home)) + monkeypatch.setattr( + executions, + "EXECUTIONS_FILE", + profile_home / "cron" / "executions.db", + ) + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.05) + monkeypatch.setattr(sched, "_get_hermes_home", lambda: profile_home) - with patch("cron.scheduler.mark_job_run", marks): - assert sched.mark_running_jobs_interrupted("shutdown") == [] - with patch("cron.scheduler.claim_dispatch", return_value=True), \ - patch("cron.scheduler.mark_execution_running", return_value={}), \ - patch("agent.secret_scope.set_secret_scope", return_value=None), \ - patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ - patch("agent.secret_scope.reset_secret_scope"), \ - patch( - "cron.scheduler.run_job", - return_value=(True, "full output", "plausible response", None), - ), \ - patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ - patch("cron.scheduler._deliver_result", delivered), \ - patch("cron.scheduler.heartbeat_fire_claim", return_value=True), \ - patch( - "cron.scheduler.fire_claim_fence", - side_effect=lambda *a, **k: contextlib.nullcontext(True), - ), \ - patch("cron.scheduler.finish_execution"): - assert sched._run_one_job_body(job, execution_token=token) is True - - assert marks.call_count == 2 - assert marks.call_args_list[1].args[:2] == (job["id"], False) - assert "Interrupted by gateway shutdown" in marks.call_args_list[1].args[2] - assert marks.call_args_list[1].kwargs["expected_fire_owner"] == "owner-1" + with jobs.use_cron_store(profile_home): + job = jobs.create_job( + prompt="do work", + schedule="every 5m", + name="contended job", + deliver="telegram:123", + ) + claimed = jobs.claim_job_for_fire(job["id"], force=True, return_job=True) + assert isinstance(claimed, dict) + owner = claimed["fire_claim"]["by"] + execution = executions.create_execution(job["id"], source="builtin") + claimed["execution_id"] = execution["id"] + + token = object() + with sched._running_lock: + sched._running_job_ids.add(job["id"]) + sched._running_fire_owners[job["id"]] = { + token: (owner, profile_home), + } + + fence_acquired = threading.Event() + release_fence = threading.Event() + + def hold_fire_fence(): + with jobs.use_cron_store(profile_home): + with jobs.fire_claim_fence( + job["id"], expected_owner=owner + ) as owns_claim: + assert owns_claim is True + fence_acquired.set() + assert release_fence.wait(timeout=5) + + holder = threading.Thread(target=hold_fire_fence) + holder.start() + assert fence_acquired.wait(timeout=5) + try: + assert sched.mark_running_jobs_interrupted("shutdown") == [] + assert sched._is_interrupted(job["id"], token) is True + finally: + release_fence.set() + holder.join(timeout=5) + assert holder.is_alive() is False + + delivered = MagicMock(return_value=None) + with ( + patch( + "cron.scheduler.run_job", + return_value=( + True, + "full output", + "plausible response", + None, + ), + ), + patch("cron.scheduler._deliver_result", delivered), + patch("agent.secret_scope.set_secret_scope", return_value=None), + patch( + "agent.secret_scope.build_profile_secret_scope", return_value=None + ), + patch("agent.secret_scope.reset_secret_scope"), + patch( + "tools.terminal_scope.install_profile_terminal_scope", + return_value=None, + ), + patch("tools.terminal_scope.reset_terminal_scope"), + ): + assert sched._run_one_job_body(claimed, execution_token=token) is True + + persisted = jobs.get_job(job["id"]) + finished = executions.get_execution(execution["id"]) + + assert isinstance(persisted, dict) + assert isinstance(finished, dict) + assert persisted["last_status"] == "error" + assert "Interrupted by gateway shutdown" in persisted["last_error"] + assert persisted["fire_claim"] is None + assert finished["status"] == "failed" assert delivered.call_count == 1 assert "plausible response" not in delivered.call_args.args[1] + assert sched._is_interrupted(job["id"], token) is False def test_shutdown_starting_at_delivery_fence_cannot_leak_normal_result(self): """The fence-local interrupt recheck closes the race after the first From d74a8d157a5a4e3fabac716df2d9c14420ec34ae Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 4 Sep 2026 12:43:24 +0700 Subject: [PATCH 4/8] [verified] fix(cron): preserve long fenced deliveries --- cron/jobs.py | 71 ++++- cron/scheduler.py | 49 ++-- tests/cron/test_fire_fence_completion_race.py | 272 ++++++++++++++++++ tests/cron/test_script_claim_heartbeat.py | 51 ++-- 4 files changed, 379 insertions(+), 64 deletions(-) create mode 100644 tests/cron/test_fire_fence_completion_race.py diff --git a/cron/jobs.py b/cron/jobs.py index 5ea25cc4c7c3..5f8fbe57506d 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -9,6 +9,7 @@ import copy from contextvars import ContextVar from dataclasses import dataclass +import errno import json import logging import shutil @@ -375,13 +376,20 @@ def _jobs_lock(): _jobs_lock_state.load_stamp = None +def _is_fire_fence_contention(exc: OSError) -> bool: + """Return whether a nonblocking platform lock reports an active holder.""" + return exc.errno in {errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK} + + @contextlib.contextmanager -def _fire_job_lock(job_id: str): +def _fire_job_lock(job_id: str, *, raise_unavailable: bool = False): """Serialize one job's owner mutations and external side effects. Unlike the global jobs lock, this lock may be held across network delivery. It is scoped to one profile + job, so unrelated cron jobs keep progressing. - Fencing fails closed when cross-process locking is unavailable. + Fencing fails closed when cross-process locking is unavailable. Heartbeat + callers can request an exception for backend I/O failure so it remains + distinct from ordinary contention with an active fence holder. """ cron_dir = _current_cron_store().cron_dir lock_key = f"{cron_dir.resolve()}::{job_id}" @@ -410,6 +418,7 @@ def _fire_job_lock(job_id: str): lock_path = cron_dir / f".fire-{lock_name}.lock" lock_fd = None acquired = False + unavailable_error = None try: lock_fd = open(lock_path, "a+", encoding="utf-8") lock_fd.seek(0) @@ -420,24 +429,56 @@ def _fire_job_lock(job_id: str): fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) acquired = True break - except (OSError, IOError): - if time.monotonic() >= deadline: - logger.error( - "Timed out waiting for fire fence %s; failing closed", - lock_path, - ) - break - time.sleep(0.1) + except OSError as exc: + if _is_fire_fence_contention(exc): + if time.monotonic() >= deadline: + logger.error( + "Timed out waiting for fire fence %s; failing closed", + lock_path, + ) + break + time.sleep(0.1) + continue + unavailable_error = exc + logger.error("Cron fire fence unavailable for %s: %s", job_id, exc) + break elif msvcrt is not None: - getattr(msvcrt, "locking")( - lock_fd.fileno(), getattr(msvcrt, "LK_LOCK"), 1 - ) - acquired = True + deadline = time.monotonic() + _JOBS_LOCK_TIMEOUT_SECONDS + while True: + try: + getattr(msvcrt, "locking")( + lock_fd.fileno(), getattr(msvcrt, "LK_NBLCK"), 1 + ) + acquired = True + break + except OSError as exc: + if _is_fire_fence_contention(exc): + if time.monotonic() >= deadline: + logger.error( + "Timed out waiting for fire fence %s; failing closed", + lock_path, + ) + break + time.sleep(0.1) + continue + unavailable_error = exc + logger.error("Cron fire fence unavailable for %s: %s", job_id, exc) + break else: # pragma: no cover - supported platforms provide one backend + unavailable_error = RuntimeError( + "No cross-process lock backend for cron fire fence" + ) logger.error("No cross-process lock backend for cron fire fence") except (OSError, IOError) as exc: + unavailable_error = exc logger.error("Cron fire fence unavailable for %s: %s", job_id, exc) + if unavailable_error is not None and raise_unavailable: + if lock_fd is not None: + lock_fd.close() + lock_fd = None + raise unavailable_error + held_locks[lock_key] = acquired try: yield acquired @@ -3817,7 +3858,7 @@ def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> Optional[bool]: successfully completing run. Callers must treat the three outcomes differently: only an explicit ``False`` is an ownership loss. """ - with _fire_job_lock(job_id) as acquired: + with _fire_job_lock(job_id, raise_unavailable=True) as acquired: if not acquired: return None return _heartbeat_fire_claim_locked( diff --git a/cron/scheduler.py b/cron/scheduler.py index 8a6724f8ecbf..5811e8d36147 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -7389,17 +7389,38 @@ def _finish_unstarted(error: str) -> None: return True def _heartbeat_loop() -> None: - last_confirmed = time.monotonic() + backend_error_since = None while not stop.wait(_RUN_CLAIM_HEARTBEAT_SECONDS): try: # Tri-state (c-027): only an explicit False is an # authoritative ownership loss. None means the fire fence - # was busy (e.g. our own delivery holds it) — ownership was - # not inspected, so the renewal is merely unconfirmed and - # falls under the grace window like any other exception. + # was busy (e.g. our own delivery holds it). Every takeover + # also needs that fence, so contention itself excludes a + # replacement owner and must not consume store-error grace. renewed = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: - renewed = None + # Genuine store/I/O uncertainty is different from a returned + # None (known fence contention). Bound consecutive backend + # failures by their own grace clock; a prior owned side effect + # must not pre-spend that budget. + now = time.monotonic() + if backend_error_since is None: + backend_error_since = now + logger.debug( + "Job '%s': fire_claim heartbeat failed", + job_id, + exc_info=True, + ) + if now - backend_error_since >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS: + lost_ownership.set() + logger.warning( + "Job '%s': fire_claim could not be renewed within %.1fs; " + "interrupting uncertain run", + job_id, + _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, + ) + return + continue if renewed is False: lost_ownership.set() logger.warning( @@ -7408,22 +7429,16 @@ def _heartbeat_loop() -> None: ) return if renewed is True: - last_confirmed = time.monotonic() + backend_error_since = None continue + # A returned None specifically means an acquired lock could not be + # obtained because another holder owns the fence. That holder also + # excludes takeover, so it resets (rather than spends) I/O grace. + backend_error_since = None logger.debug( - "Job '%s': fire_claim heartbeat unconfirmed (fence contended " - "or store error)", + "Job '%s': fire_claim heartbeat skipped; fire fence busy", job_id, ) - if time.monotonic() - last_confirmed >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS: - lost_ownership.set() - logger.warning( - "Job '%s': fire_claim could not be renewed within %.1fs; " - "interrupting uncertain run", - job_id, - _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, - ) - return heartbeat_thread = threading.Thread( target=heartbeat_context.run, diff --git a/tests/cron/test_fire_fence_completion_race.py b/tests/cron/test_fire_fence_completion_race.py new file mode 100644 index 000000000000..53670d720388 --- /dev/null +++ b/tests/cron/test_fire_fence_completion_race.py @@ -0,0 +1,272 @@ +"""Regression tests for fire-fence heartbeat settlement races. + +A worker's own side-effect fence can make its heartbeat return ``None`` while +saving or delivering. That self-contention must not become ownership loss by +duration alone. Confirmed takeover and external cancellation still fail closed. +""" + +import contextlib +import errno +import threading +import time +from unittest.mock import patch + +import pytest + +import cron.scheduler as scheduler + + +@contextlib.contextmanager +def _owned_fence(*_args, **_kwargs): + yield True + + +@pytest.fixture +def temp_home(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + yield tmp_path + + +def _make_claimed_job(jobs, name): + job = jobs.create_job(prompt="x", schedule="every 5m", name=name) + assert jobs.claim_job_for_fire(job["id"]) is True + stored = jobs.get_job(job["id"]) + assert stored is not None and stored.get("fire_claim") + return job, stored["fire_claim"]["by"] + + +def test_fire_fence_contention_errno_is_portable(): + import cron.jobs as jobs + + assert jobs._is_fire_fence_contention(OSError(errno.EACCES, "busy")) + assert jobs._is_fire_fence_contention(OSError(errno.EAGAIN, "busy")) + assert not jobs._is_fire_fence_contention(OSError(errno.EMFILE, "unavailable")) + + +def test_heartbeat_lock_backend_error_propagates_as_store_uncertainty( + temp_home, monkeypatch +): + """A fence I/O failure is not the known-contention ``None`` verdict.""" + import cron.jobs as jobs + + job, owner = _make_claimed_job(jobs, "fence-io-error") + + def unavailable(*_args, **_kwargs): + raise OSError("lock backend unavailable") + + monkeypatch.setattr(jobs, "open", unavailable, raising=False) + + with pytest.raises(OSError, match="lock backend unavailable"): + jobs.heartbeat_fire_claim(job["id"], expected_owner=owner) + + +def test_store_error_gets_full_grace_after_prolonged_fence_contention(monkeypatch): + """Known contention does not pre-spend later backend-error grace.""" + first_error = threading.Event() + calls = 0 + + def heartbeat(*_args, **_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + return True + if calls <= 16: + return None + first_error.set() + raise OSError("store unavailable") + + def run_body(_job, **kwargs): + cancel = kwargs["fire_claim_lost"] + assert first_error.wait(timeout=1) + assert not cancel.wait(timeout=0.02) + assert cancel.wait(timeout=0.3) + return True + + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) + monkeypatch.setattr(scheduler, "_run_one_job_body", run_body) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.005) + monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.05) + + assert scheduler.run_one_job( + {"id": "mixed-uncertainty", "fire_claim": {"by": "owner"}} + ) is True + + +def test_successful_long_delivery_is_not_demoted_after_heartbeat_grace( + temp_home, monkeypatch +): + """A delivery holding its own fence past heartbeat grace stays successful.""" + import cron.jobs as jobs + from cron.executions import latest_execution + + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.005) + monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.02) + + job, owner = _make_claimed_job(jobs, "long-delivery") + jid = job["id"] + delivered = [] + + def slow_delivery(_job, content, **_kwargs): + # _deliver_result runs under the real side-effect fence. Keep it held + # long enough for repeated heartbeat lock timeouts to exceed grace. + time.sleep(0.12) + delivered.append(content) + return None + + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a, **_kw: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_a, **_kw: (True, "out", "final response", None), + ) + monkeypatch.setattr(scheduler, "save_job_output", lambda *_a: "out.md") + monkeypatch.setattr(scheduler, "_deliver_result", slow_delivery) + + with ( + patch("cron.scheduler._launch_external_cron_worker", return_value=False), + patch("agent.secret_scope.set_secret_scope", return_value=None), + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), + patch("agent.secret_scope.reset_secret_scope"), + patch("tools.terminal_scope.install_profile_terminal_scope", return_value=None), + patch("tools.terminal_scope.reset_terminal_scope"), + ): + assert scheduler.run_one_job( + { + "id": jid, + "name": "long-delivery", + "deliver": "all", + "fire_claim": {"at": "2026-09-01T00:00:00+00:00", "by": owner}, + } + ) is True + + assert delivered == ["final response"] + final = jobs.get_job(jid) + assert final is not None + assert final["last_status"] == "ok", final + assert not final.get("last_error"), final + assert not final.get("failure_streak"), final + row = latest_execution(jid) + assert row is not None and row["status"] == "completed", row + + +def test_shutdown_during_long_delivery_still_fails_closed(temp_home, monkeypatch): + """A real transport cancellation is not mistaken for fence self-contention.""" + import cron.jobs as jobs + from cron.executions import latest_execution + + monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.005) + monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.02) + + job, owner = _make_claimed_job(jobs, "cancelled-long-delivery") + jid = job["id"] + external_cancel = threading.Event() + + def cancelled_delivery(*_args, **_kwargs): + time.sleep(0.12) + external_cancel.set() + return None + + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a, **_kw: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) + monkeypatch.setattr( + scheduler, + "run_job", + lambda *_a, **_kw: (True, "out", "final response", None), + ) + monkeypatch.setattr(scheduler, "save_job_output", lambda *_a: "out.md") + monkeypatch.setattr(scheduler, "_deliver_result", cancelled_delivery) + + with ( + patch("cron.scheduler._launch_external_cron_worker", return_value=False), + patch("agent.secret_scope.set_secret_scope", return_value=None), + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), + patch("agent.secret_scope.reset_secret_scope"), + patch("tools.terminal_scope.install_profile_terminal_scope", return_value=None), + patch("tools.terminal_scope.reset_terminal_scope"), + ): + assert scheduler.run_one_job( + { + "id": jid, + "name": "cancelled-long-delivery", + "deliver": "all", + "fire_claim": {"at": "2026-09-01T00:00:00+00:00", "by": owner}, + }, + cancel_event=external_cancel, + ) is True + + final = jobs.get_job(jid) + assert final is not None + assert final["last_status"] == "error", final + assert final["failure_streak"] == 1, final + assert final["last_error"] == "Interrupted by shutdown before terminal completion." + row = latest_execution(jid) + assert row is not None and row["status"] == "failed", row + + +def test_genuine_ownership_loss_still_fails_closed(monkeypatch): + """A confirmed takeover discards the stale result without delivering it.""" + finished = [] + delivered = [] + + monkeypatch.setattr( + scheduler, "create_execution", lambda *_a, **_kw: {"id": "exec-loss"} + ) + monkeypatch.setattr(scheduler, "_launch_external_cron_worker", lambda *_a: False) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a, **_kw: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) + monkeypatch.setattr( + scheduler, "run_job", lambda *_a, **_kw: (True, "out", "stale", None) + ) + monkeypatch.setattr(scheduler, "fire_claim_fence", _owned_fence, raising=False) + monkeypatch.setattr(scheduler, "save_job_output", lambda *_a: "out.md") + monkeypatch.setattr( + scheduler, "_deliver_result", lambda *_a, **_kw: delivered.append(True) + ) + + probes = {"n": 0} + + def heartbeat(_job_id, *, expected_owner): + probes["n"] += 1 + return probes["n"] == 1 + + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(scheduler, "mark_job_run", lambda *_a, **_kw: True) + monkeypatch.setattr( + scheduler, + "finish_execution", + lambda execution_id, **kwargs: finished.append((execution_id, kwargs)), + ) + + with ( + patch("agent.secret_scope.set_secret_scope", return_value=None), + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), + patch("agent.secret_scope.reset_secret_scope"), + patch("tools.terminal_scope.install_profile_terminal_scope", return_value=None), + patch("tools.terminal_scope.reset_terminal_scope"), + ): + assert scheduler.run_one_job( + { + "id": "taken-over", + "name": "taken-over", + "deliver": "all", + "fire_claim": { + "at": "2026-09-01T00:00:00+00:00", + "by": "owner-A", + }, + } + ) is True + + assert delivered == [] + assert finished == [ + ( + "exec-loss", + { + "success": False, + "error": "Fire claim ownership lost; stale result was discarded.", + }, + ) + ] diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index f66136eb9eaf..87c38fdb12a4 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -830,8 +830,8 @@ def _deliver(*_args, **_kwargs): ) == ("completed", None, "ok") -def test_interrupted_path_none_probe_does_not_claim_confirmed_loss(monkeypatch): - """Grace-exhausted stop + contended probe must not claim a confirmed loss (c-027).""" +def test_sustained_fence_contention_does_not_consume_store_error_grace(monkeypatch): + """Known fence contention cannot manufacture cancellation by duration.""" import cron.scheduler as scheduler calls = [] @@ -840,41 +840,24 @@ def heartbeat(*_args, **_kwargs): calls.append(1) return True if len(calls) == 1 else None - def _run_job(job, *, defer_agent_teardown=None, extra_prompt=None, - cancel_event=None, execution_id=None): - # Sustained None heartbeats exhaust the grace window -> lost_ownership - # set as "uncertain, stop". Never an explicit False. - assert cancel_event is not None - assert cancel_event.wait(timeout=1.0) - return True, "output", "response", None + def run_body(_job, **kwargs): + # Stay contended well beyond the shortened grace window. Because every + # takeover also needs this fence, returned None is safe to defer until + # the side effect releases it; only exceptions consume error grace. + assert not kwargs["fire_claim_lost"].wait(timeout=0.12) + return True - finish = MagicMock() - mark = MagicMock(return_value=True) job = { "id": "grace-none", - "execution_id": "grace-none-exec", "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner"}, } monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) - monkeypatch.setattr(scheduler, "run_job", _run_job) + monkeypatch.setattr(scheduler, "_run_one_job_body", run_body) monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.03) - monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a: True) - monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) - monkeypatch.setattr(scheduler, "mark_job_run", mark) - monkeypatch.setattr(scheduler, "finish_execution", finish) - - with patch("agent.secret_scope.set_secret_scope", return_value=None), \ - patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ - patch("agent.secret_scope.reset_secret_scope"): - assert scheduler.run_one_job(job) is True - errors = [call.kwargs.get("error") for call in finish.call_args_list] - assert errors, "execution ledger never closed" - for err in errors: - assert "ownership lost" not in err.lower(), ( - f"None probe adjudicated as confirmed ownership loss: {err!r}" - ) + assert scheduler.run_one_job(job) is True + assert len(calls) >= 4 def test_terminal_write_none_does_not_mark_ownership_lost(monkeypatch): @@ -912,10 +895,14 @@ def owned_fence(*_args, **_kwargs): patch("agent.secret_scope.reset_secret_scope"): assert scheduler.run_one_job(job) is True - for call in finish.call_args_list: - assert call.kwargs.get("error") != ( - "Fire claim ownership lost before terminal completion." - ), "unconfirmed terminal write adjudicated as ownership loss" + finish.assert_called_once_with( + "terminal-none-exec", + success=False, + error=( + "Fire claim fence unavailable before terminal completion; " + "outcome uncertain." + ), + ) def test_recovery_execution_fails_closed_while_original_delivery_settles( From 189daf8c971354bf10ced5b128fb9b1db88ad0eb Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 4 Sep 2026 09:49:16 +0700 Subject: [PATCH 5/8] =?UTF-8?q?fix(cron):=20tri-state=20fire-claim=20heart?= =?UTF-8?q?beat=20=E2=80=94=20fence=20contention=20is=20not=20ownership=20?= =?UTF-8?q?loss=20(c-027)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit heartbeat_fire_claim() collapsed fire-fence acquisition failure into False. During slow fenced delivery the worker's own heartbeat thread cannot acquire the same process-local RLock, so a successfully delivering run was misclassified as ownership loss and terminalled as 'Interrupted by shutdown before terminal completion.' (TrustMRR exec 4219b48d while RSI recovery exec f502e929 was settling). Tri-state contract: - True = renewed/confirmed owner - False = authoritatively inspected: owner mismatch / absent claim - None = fence unavailable/unconfirmed (contention or store error) - initial validation stays fail-closed: run starts only on exactly True - heartbeat loop: None rides the existing grace window (180s prod vs 30s fence wait); immediate cancel only on explicit False - post-run probes (_fire_claim_ownership_lost, both interrupted blocks, terminal owner-CAS read): None can never adjudicate a confirmed loss; uncertain outcomes get distinct ledger errors - mark_job_run also tri-state: None = fence unavailable (CAS never ran), False stays reserved for confirmed owner mismatch - _side_effect_fence unchanged: still the exactly-once save/delivery barrier Tests (all written red-first, verified failing on unfixed code): slow_owned_delivery (parent handoff), sustained fence-contention grace, pre/post-delivery None probes, grace-exhausted None probe, terminal write None, mark_job_run fence-unavailable None, RSI recovery dispatch fails closed during original's fenced delivery, TrustMRR slow-owned delivery terminal success. Managed-gateway restart E2E (systemd scope) passes on host: one gateway, single side effect, single delivery. --- cron/jobs.py | 42 +++++-- cron/scheduler.py | 143 +++++++++++++++++----- tests/cron/test_script_claim_heartbeat.py | 48 ++++++++ 3 files changed, 197 insertions(+), 36 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 044d0bee54ec..3c5bff7b7846 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -2229,14 +2229,19 @@ def mark_job_run( status: Optional[str] = None, *, expected_fire_owner: Optional[str] = None, -) -> bool: +) -> Optional[bool]: """Mark a job as run: update last_run_at/last_status, bump completed, recompute next_run_at, and retire the record as a terminal completion when the repeat limit is reached. ``delivery_error`` is separate from the agent error: agent succeeded but delivery failed records ``last_status = "delivery_failed"`` (never "ok") while ``failure_streak`` is left alone. An - explicit ``status`` (e.g. "blocked_config") overrides the derived value. False when the fence - can't be taken, the job is missing, or ``expected_fire_owner`` no longer holds the fire claim. + explicit ``status`` (e.g. "blocked_config") overrides the derived value. + + Tri-state (c-027): ``True`` — recorded (owner CAS matched when supplied); ``False`` — + authoritatively rejected (job missing, or ``expected_fire_owner`` no longer holds the fire + claim); ``None`` — unconfirmed: the fire fence could not be acquired, so the write (and the + CAS) never happened. Callers must not treat ``None`` as ownership loss — the worker's own + fenced side effect may be the reason the fence was busy. """ def apply(jobs, _i, job): if expected_fire_owner is not None: @@ -2259,7 +2264,10 @@ def locked(): return False return found - return _under_fire_fence(job_id, locked) + with _fire_job_lock(job_id) as acquired: + if not acquired: + return None + return locked() def _write_oneshot_diagnostic(job: Dict[str, Any], text: str, what: str) -> bool: @@ -2511,13 +2519,31 @@ def apply(jobs, _i, job): return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False)) -def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> bool: - """Refresh an active ``fire_claim`` without extending another owner's lease: an execution may - outlive the TTL, and the owner check stops a stale runner from refreshing a recovered claim.""" +def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> Optional[bool]: + """Tri-state fire-claim heartbeat (c-027). + + Returns: + ``True`` — renewed/confirmed: we hold the fire fence and the + persisted claim still names ``expected_owner``. + ``False`` — authoritatively inspected: the fence was acquired and + the claim is absent or owned by someone else. + ``None`` — unconfirmed: the fire fence could not be acquired (its + process-local RLock is contended, e.g. held across a + slow side-effect delivery) so ownership was never + actually inspected. + + Collapsing ``None`` into ``False`` made a worker's own heartbeat thread + report authoritative ownership loss during fenced delivery and cancel a + successfully completing run. Callers must treat the three outcomes + differently: only an explicit ``False`` is an ownership loss. + """ def apply(jobs, _i, job): return _refresh_claim(jobs, job.get("fire_claim"), expected_owner) - return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False)) + with _fire_job_lock(job_id) as acquired: + if not acquired: + return None + return _with_job(job_id, apply, False) # Completed one-shots are retained in jobs.json (final status stays inspectable) and pruned by diff --git a/cron/scheduler.py b/cron/scheduler.py index 78b57f525380..11a5ab8bdc60 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2436,41 +2436,80 @@ def _finish_unstarted(error: str) -> None: exc_info=True) try: + # Tri-state (c-027): True=renewed/owned; False=confirmed loss; + # None=unconfirmed (fence contended). Initial validation stays + # fail-closed: the run may only start on exactly True. owns_fire_claim = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: - logger.warning("Job '%s': initial fire_claim validation failed", job_id, exc_info=True) - _finish_unstarted("Fire claim ownership could not be validated before execution started.") + owns_fire_claim = None + logger.warning( + "Job '%s': initial fire_claim validation failed", + job_id, + exc_info=True, + ) + _finish_unstarted( + "Fire claim ownership could not be validated before execution started." + ) return True - if owns_fire_claim is False: - logger.warning("Job '%s': fire claim ownership was already lost before execution", job_id) - _finish_unstarted("Fire claim ownership lost before execution started.") + if owns_fire_claim is not True: + # Fail closed (c-027): only an exactly-True renewal lets the run + # start. An explicit False is a confirmed loss; None (or an + # unexpected value) is unconfirmed ownership — the run must not + # begin without a validated claim. + if owns_fire_claim is False: + logger.warning( + "Job '%s': fire claim ownership was already lost before execution", + job_id, + ) + _finish_unstarted("Fire claim ownership lost before execution started.") + else: + logger.warning( + "Job '%s': fire claim ownership could not be confirmed before " + "execution; refusing to start", + job_id, + ) + _finish_unstarted( + "Fire claim ownership could not be validated before execution started." + ) return True def _heartbeat_loop() -> None: last_confirmed = time.monotonic() while not stop.wait(_RUN_CLAIM_HEARTBEAT_SECONDS): try: - if not heartbeat_fire_claim(job_id, expected_owner=owner): - lost_ownership.set() - logger.warning( - "Job '%s': fire claim ownership lost; interrupting stale run", - job_id) - return - last_confirmed = time.monotonic() + # Tri-state (c-027): only an explicit False is an + # authoritative ownership loss. None means the fire fence + # was busy (e.g. our own delivery holds it) — ownership was + # not inspected, so the renewal is merely unconfirmed and + # falls under the grace window like any other exception. + renewed = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: - logger.debug("Job '%s': fire_claim heartbeat failed", job_id, exc_info=True) - if ( - time.monotonic() - last_confirmed - >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS - ): - lost_ownership.set() - logger.warning( - "Job '%s': fire_claim could not be renewed within %.1fs; " - "interrupting uncertain run", - job_id, - _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS) - return + renewed = None + if renewed is False: + lost_ownership.set() + logger.warning( + "Job '%s': fire claim ownership lost; interrupting stale run", + job_id, + ) + return + if renewed is True: + last_confirmed = time.monotonic() + continue + logger.debug( + "Job '%s': fire_claim heartbeat unconfirmed (fence contended " + "or store error)", + job_id, + ) + if time.monotonic() - last_confirmed >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS: + lost_ownership.set() + logger.warning( + "Job '%s': fire_claim could not be renewed within %.1fs; " + "interrupting uncertain run", + job_id, + _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, + ) + return heartbeat_thread = _start_heartbeat_thread( _heartbeat_loop, "cron-fire-claim-heartbeat", @@ -2567,10 +2606,23 @@ def run_one_job( def _record_fire_ownership_lost(job_id: str, fire_owner: Optional[str], execution_id: str) -> None: """Bookkeeping after fire-claim ownership loss. A transport-level cancel (dashboard drain) is not a real loss — we still own the claim, so record the interruption via the owner-fenced - terminal write instead of leaving fire_claim/last_status stale; otherwise discard.""" - if fire_owner is not None and heartbeat_fire_claim(job_id, expected_owner=fire_owner): + terminal write instead of leaving fire_claim/last_status stale. + + Tri-state (c-027): a None probe is unconfirmed — never adjudicate "ownership lost" on it; + the uncertain arm records the distinct uncertain-outcome error instead. + """ + probe = ( + heartbeat_fire_claim(job_id, expected_owner=fire_owner) + if fire_owner is not None + else False + ) + if probe is True: mark_job_run(job_id, False, _OWNERSHIP_LOST_INTERRUPTED, expected_fire_owner=fire_owner) finish_execution(execution_id, success=False, error=_OWNERSHIP_LOST_INTERRUPTED) + elif probe is None: + finish_execution( + execution_id, success=False, + error="Fire claim could not be re-validated after interruption; outcome uncertain.") else: finish_execution( execution_id, success=False, @@ -2664,7 +2716,15 @@ def lost(self) -> bool: if self.owner is None: return False try: - if heartbeat_fire_claim(self.job["id"], expected_owner=self.owner): + # Tri-state (c-027): True = still owned; False = authoritative + # loss; None = fence contended (likely our own side effect + # holds it) — NOT a loss. None keeps the caller in + # unknown-but-not-lost territory; the fence acquisition below + # re-validates ownership before the terminal write. + renewed = heartbeat_fire_claim(self.job["id"], expected_owner=self.owner) + if renewed is True: + return False + if renewed is None: return False except Exception: logger.debug( @@ -2790,11 +2850,20 @@ def _finish_completed_run(d: _RunDelivery, fire_owner: Optional[str], execution_ if d.blocked_config: mark_kwargs["status"] = "blocked_config" marked = mark_job_run(job["id"], d.success, d.error, **mark_kwargs) - if fire_owner is not None and not marked: + # Tri-state (c-027): only an explicit False is an authoritative owner-CAS + # rejection. None means the fire fence was unavailable — the CAS never + # ran, so ownership was NOT confirmed lost; record the unconfirmed-terminal + # outcome instead of a confirmed loss. + if fire_owner is not None and marked is False: finish_execution( execution_id, success=False, error="Fire claim ownership lost before terminal completion.") return True + if fire_owner is not None and marked is None: + finish_execution( + execution_id, success=False, + error="Fire claim fence unavailable before terminal completion; outcome uncertain.") + return True delivery_outcome = _classify_delivery_outcome( delivery_error=d.delivery_error, should_deliver=d.should_deliver, @@ -2949,6 +3018,15 @@ def _teardown_deferred() -> None: if _fire_claim_ownership_lost(): _teardown_deferred() + # Distinguish a real ownership loss (TTL expiry / replacement + # claim) from a transport-level cancel (dashboard drain): in the + # latter case WE still own the claim, and silently discarding + # would leave fire_claim lingering until TTL and last_status + # stale. Probe ownership once; if still ours, record the + # interruption through the owner-fenced terminal write. + # Tri-state (c-027): a None probe is unconfirmed — never + # adjudicate "ownership lost" on it; the None arm falls through + # to the uncertain-store recording path below. _record_fire_ownership_lost(job["id"], fire_owner, execution_id) return True @@ -2967,6 +3045,11 @@ def _teardown_deferred() -> None: _teardown_deferred() if d.side_effect_ownership_lost or _fire_claim_ownership_lost(): + # Same transport-cancel distinction as the pre-side-effect path: + # if WE still own the claim, record the interruption instead of + # discarding silently (lingering claim + stale last_status). + # Tri-state (c-027): a None probe is unconfirmed — never + # adjudicate "ownership lost" on it. _record_fire_ownership_lost(job["id"], fire_owner, execution_id) return True @@ -2979,6 +3062,10 @@ def _teardown_deferred() -> None: _finish_interrupted_run(job, execution_id, delivery_error) return True + # Tri-state (c-027), inside _finish_completed_run: only an explicit False from + # mark_job_run is an authoritative owner-CAS rejection. None means the fire + # fence was unavailable — the CAS never ran, so ownership was NOT confirmed + # lost; the unconfirmed-terminal outcome is recorded instead of a confirmed loss. return _finish_completed_run(d, fire_owner, execution_id) except BaseException as e: # noqa: BLE001 — deliberate: see below diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index da5a27e160d8..35d441177e00 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -873,6 +873,53 @@ def run_body(_job, **kwargs): assert len(calls) >= 4 +def test_interrupted_path_none_probe_does_not_claim_confirmed_loss(monkeypatch): + """Grace-exhausted stop + contended probe must not claim a confirmed loss (c-027).""" + import cron.scheduler as scheduler + + calls = [] + + def heartbeat(*_args, **_kwargs): + calls.append(1) + return True if len(calls) == 1 else None + + def _run_job(job, *, defer_agent_teardown=None, extra_prompt=None, + cancel_event=None, execution_id=None): + # Sustained None heartbeats exhaust the grace window -> lost_ownership + # set as "uncertain, stop". Never an explicit False. + assert cancel_event is not None + assert cancel_event.wait(timeout=1.0) + return True, "output", "response", None + + finish = MagicMock() + mark = MagicMock(return_value=True) + job = { + "id": "grace-none", + "execution_id": "grace-none-exec", + "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner"}, + } + monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) + monkeypatch.setattr(scheduler, "run_job", _run_job) + monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.03) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a: True) + monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) + monkeypatch.setattr(scheduler, "mark_job_run", mark) + monkeypatch.setattr(scheduler, "finish_execution", finish) + + with patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"): + assert scheduler.run_one_job(job) is True + + errors = [call.kwargs.get("error") for call in finish.call_args_list] + assert errors, "execution ledger never closed" + for err in errors: + assert "ownership lost" not in err.lower(), ( + f"None probe adjudicated as confirmed ownership loss: {err!r}" + ) + + def test_terminal_write_none_does_not_mark_ownership_lost(monkeypatch): """An unconfirmed (None) terminal write must not be read as CAS loss (c-027).""" import cron.scheduler as scheduler @@ -1135,3 +1182,4 @@ def test_none_probe_cannot_authorize_unfenced_exception_delivery( fence.assert_called_once_with("stale-worker", expected_owner="old-owner") delivered.assert_not_called() + From 1bc7b76256b6715a15a7fd56df118e38f89eaa50 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 4 Sep 2026 10:57:18 +0700 Subject: [PATCH 6/8] [verified] fix(cron): fence tri-state interruption delivery --- cron/scheduler.py | 110 +++++++++++++++++----- tests/cron/test_script_claim_heartbeat.py | 1 - 2 files changed, 87 insertions(+), 24 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 11a5ab8bdc60..f5d64049c882 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -545,6 +545,10 @@ def _is_cron_silence_response(text: str) -> bool: # false "ok" (#60432). Token keying keeps an interruption scoped to that exact execution: a later run of the # same job ID (recurring jobs reuse the ID every fire) must not inherit the stale flag. _interrupted_job_ids: set = set() +# Executions observed during shutdown but whose owner-fenced terminal write has +# not yet succeeded. Pending tokens suppress normal-result delivery, but do +# not authorize the worker to skip its own terminal write. +_interrupting_job_ids: set = set() class _CancelEventLike(Protocol): @@ -854,10 +858,36 @@ def mark_running_jobs_interrupted( ) -> list: """Best-effort: mark every in-flight cron job interrupted; returns the job IDs marked. - Called by gateway shutdown right after ``process_registry.kill_all()``: a job whose tool was - killed must never report success. ``only_owners`` (``(job_id, fire_owner)`` pairs) restricts - marking. Tokens go into ``_interrupted_job_ids`` BEFORE ``last_status`` is written so - ``run_one_job`` sees them. + Called by the gateway shutdown path immediately after it force-kills + tool subprocesses (``process_registry.kill_all()``). A job whose tool + subprocess was just killed out from under it must never be allowed to + report success — even though its agent thread is still alive in this + same process and may go on to produce a plausible-looking final + response from the now-truncated tool output. + + Records execution tokens in ``_interrupting_job_ids`` BEFORE writing + ``last_status`` so ``run_one_job`` suppresses a plausible normal result + while shutdown races its thread. A successful terminal write promotes + the token to ``_interrupted_job_ids``; only that confirmed state lets the + worker skip its own terminal write. If the write is unavailable, the + pending token still suppresses delivery but leaves the worker responsible + for eventually persisting terminal state. See the checks near the end of + ``run_one_job``. This does not attempt to correlate the killed + subprocess PID to a specific job ID (the process registry tracks PIDs, + not cron job IDs); any job still dispatched at the moment of a forced + kill is treated as interrupted, matching the coarser precedent already + set by ``GatewayRunner._interrupt_running_agents``, which interrupts + every entry in ``_running_agents`` on a drain timeout without + per-agent correlation either. + + ``only_owners``: optional set of ``(job_id, fire_owner)`` pairs. When + given (dashboard webhook drain), ONLY those exact executions are + marked — unrelated runs sharing the process (e.g. the desktop ticker's + own jobs) are left untouched. Interruption flags are recorded per + execution token, so a later run of the same job ID never consumes a + stale flag that targeted its dead predecessor. + + Returns the list of job IDs marked, for the caller to log. """ with _running_lock: restart_safe_waiters = set(_restart_safe_waiter_job_ids) @@ -877,28 +907,47 @@ def mark_running_jobs_interrupted( _running_job_ids - registered_ids - restart_safe_waiters ) ) - _interrupted_job_ids.update( + _interrupting_job_ids.update( token if token is not None else job_id for token, job_id, _owner, _profile_home in active_fires ) marked = [] - for _token, job_id, fire_owner, profile_home in active_fires: + for token, job_id, fire_owner, profile_home in active_fires: + interrupt_key = token if token is not None else job_id if not fire_owner: logger.warning( "Job '%s' interrupted before its durable fire owner was registered; " "leaving persisted state untouched", - job_id) - # Still report it: shutdown uses the returned IDs for the interrupted-cron notice. The - # in-memory flag WAS recorded above; only the persisted last_status write is skipped. - # See #82232. + job_id, + ) + # Report the interruption to the caller, but leave the token + # pending rather than confirmed: the worker must still perform + # its own terminal write because shutdown did not persist one. marked.append(job_id) continue try: with use_cron_store(profile_home): - if mark_job_run( - job_id, False, reason, expected_fire_owner=fire_owner): - marked.append(job_id) + persisted = mark_job_run( + job_id, + False, + reason, + expected_fire_owner=fire_owner, + ) + with _running_lock: + if persisted is True: + _interrupting_job_ids.discard(interrupt_key) + _interrupted_job_ids.add(interrupt_key) + elif persisted is False: + # Authoritative owner loss: no write by this worker is + # allowed or needed. A pending worker-side write is still + # fenced by its stale expected owner. + _interrupting_job_ids.discard(interrupt_key) + if persisted is True: + marked.append(job_id) except Exception as e: + # Keep the token pending. This continues to suppress delivery, + # while _consume_interrupted_flag returns False so the worker + # retries the terminal write rather than assuming this one landed. logger.warning("Failed to mark job %s interrupted: %s", job_id, e) return marked @@ -908,23 +957,38 @@ def _is_interrupted(job_id: str, token: Optional[object] = None) -> bool: what to deliver; does not clear the flag (the authoritative pre-``last_status`` check needs it). ``token`` scopes to one execution so a fresh run reusing the job ID isn't poisoned.""" with _running_lock: - if token is not None and token in _interrupted_job_ids: + if token is not None and ( + token in _interrupted_job_ids or token in _interrupting_job_ids + ): return True - return job_id in _interrupted_job_ids + return ( + job_id in _interrupted_job_ids + or job_id in _interrupting_job_ids + ) def _consume_interrupted_flag(job_id: str, token: Optional[object] = None) -> bool: - """Return True and clear the flag if shutdown marked THIS execution interrupted. Called right - before ``last_status`` is written; consuming stops the flag leaking into a later run.""" + """Return True and clear the flag if the shutdown path already marked + THIS execution interrupted (see ``mark_running_jobs_interrupted``). + + Called by ``run_one_job`` right before it would otherwise write its own + ``last_status``. Return True only for a confirmed flag whose shutdown + terminal write succeeded. A pending flag still gets cleared, but returns + False so the worker remains responsible for durable terminal state. + Consuming rather than just checking keeps either flag from leaking across + a later, unrelated run of the same recurring job ID.""" with _running_lock: - hit = False - if token is not None and token in _interrupted_job_ids: - _interrupted_job_ids.discard(token) - hit = True + confirmed = False + if token is not None: + if token in _interrupted_job_ids: + _interrupted_job_ids.discard(token) + confirmed = True + _interrupting_job_ids.discard(token) if job_id in _interrupted_job_ids: _interrupted_job_ids.discard(job_id) - hit = True - return hit + confirmed = True + _interrupting_job_ids.discard(job_id) + return confirmed def _inactivity_watchdog_loop( diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index 35d441177e00..ce9e499ea5f6 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -1182,4 +1182,3 @@ def test_none_probe_cannot_authorize_unfenced_exception_delivery( fence.assert_called_once_with("stale-worker", expected_owner="old-owner") delivered.assert_not_called() - From 4c37e427b76f94a09a51fc0304b95c7b25f3d9c0 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 4 Sep 2026 12:43:24 +0700 Subject: [PATCH 7/8] [verified] fix(cron): preserve long fenced deliveries --- cron/jobs.py | 29 +++++++++++--- cron/scheduler.py | 49 +++++++++++++++-------- tests/cron/test_script_claim_heartbeat.py | 35 +++++----------- 3 files changed, 65 insertions(+), 48 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 3c5bff7b7846..e95c99e28a1b 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -5,6 +5,7 @@ import copy from contextvars import ContextVar from dataclasses import dataclass, field +import errno import json import logging import shutil @@ -288,11 +289,21 @@ def _jobs_lock(): _jobs_lock_state.load_stamp = None +def _is_fire_fence_contention(exc: OSError) -> bool: + """Return whether a nonblocking platform lock reports an active holder.""" + return exc.errno in {errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK} + + @contextlib.contextmanager -def _fire_job_lock(job_id: str): - """Serialize one job's owner mutations and external side effects. Unlike the global jobs lock - this may be held across network delivery; scoped to one profile + job so unrelated jobs keep - progressing. Fails closed when cross-process locking is unavailable.""" +def _fire_job_lock(job_id: str, *, raise_unavailable: bool = False): + """Serialize one job's owner mutations and external side effects. + + Unlike the global jobs lock, this lock may be held across network delivery. + It is scoped to one profile + job, so unrelated cron jobs keep progressing. + Fencing fails closed when cross-process locking is unavailable. Heartbeat + callers can request an exception for backend I/O failure so it remains + distinct from ordinary contention with an active fence holder. + """ cron_dir = _current_cron_store().cron_dir lock_key = f"{cron_dir.resolve()}::{job_id}" with _fire_fence_locks_guard: @@ -316,6 +327,7 @@ def _fire_job_lock(job_id: str): lock_path = cron_dir / f".fire-{uuid.uuid5(uuid.NAMESPACE_URL, lock_key).hex}.lock" lock_fd = None acquired = False + unavailable_error = None try: lock_fd = open(lock_path, "a+", encoding="utf-8") lock_fd.seek(0) @@ -326,8 +338,15 @@ def _fire_job_lock(job_id: str): logger.error("Timed out waiting for fire fence %s; failing closed", lock_path) acquired = bool(result) except (OSError, IOError) as exc: + unavailable_error = exc logger.error("Cron fire fence unavailable for %s: %s", job_id, exc) + if unavailable_error is not None and raise_unavailable: + if lock_fd is not None: + lock_fd.close() + lock_fd = None + raise unavailable_error + held_locks[lock_key] = acquired try: yield acquired @@ -2540,7 +2559,7 @@ def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> Optional[bool]: def apply(jobs, _i, job): return _refresh_claim(jobs, job.get("fire_claim"), expected_owner) - with _fire_job_lock(job_id) as acquired: + with _fire_job_lock(job_id, raise_unavailable=True) as acquired: if not acquired: return None return _with_job(job_id, apply, False) diff --git a/cron/scheduler.py b/cron/scheduler.py index f5d64049c882..5b7ba49633e7 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2539,17 +2539,38 @@ def _finish_unstarted(error: str) -> None: return True def _heartbeat_loop() -> None: - last_confirmed = time.monotonic() + backend_error_since = None while not stop.wait(_RUN_CLAIM_HEARTBEAT_SECONDS): try: # Tri-state (c-027): only an explicit False is an # authoritative ownership loss. None means the fire fence - # was busy (e.g. our own delivery holds it) — ownership was - # not inspected, so the renewal is merely unconfirmed and - # falls under the grace window like any other exception. + # was busy (e.g. our own delivery holds it). Every takeover + # also needs that fence, so contention itself excludes a + # replacement owner and must not consume store-error grace. renewed = heartbeat_fire_claim(job_id, expected_owner=owner) except Exception: - renewed = None + # Genuine store/I/O uncertainty is different from a returned + # None (known fence contention). Bound consecutive backend + # failures by their own grace clock; a prior owned side effect + # must not pre-spend that budget. + now = time.monotonic() + if backend_error_since is None: + backend_error_since = now + logger.debug( + "Job '%s': fire_claim heartbeat failed", + job_id, + exc_info=True, + ) + if now - backend_error_since >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS: + lost_ownership.set() + logger.warning( + "Job '%s': fire_claim could not be renewed within %.1fs; " + "interrupting uncertain run", + job_id, + _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, + ) + return + continue if renewed is False: lost_ownership.set() logger.warning( @@ -2558,22 +2579,16 @@ def _heartbeat_loop() -> None: ) return if renewed is True: - last_confirmed = time.monotonic() + backend_error_since = None continue + # A returned None specifically means an acquired lock could not be + # obtained because another holder owns the fence. That holder also + # excludes takeover, so it resets (rather than spends) I/O grace. + backend_error_since = None logger.debug( - "Job '%s': fire_claim heartbeat unconfirmed (fence contended " - "or store error)", + "Job '%s': fire_claim heartbeat skipped; fire fence busy", job_id, ) - if time.monotonic() - last_confirmed >= _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS: - lost_ownership.set() - logger.warning( - "Job '%s': fire_claim could not be renewed within %.1fs; " - "interrupting uncertain run", - job_id, - _FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS, - ) - return heartbeat_thread = _start_heartbeat_thread( _heartbeat_loop, "cron-fire-claim-heartbeat", diff --git a/tests/cron/test_script_claim_heartbeat.py b/tests/cron/test_script_claim_heartbeat.py index ce9e499ea5f6..5f2532108d70 100644 --- a/tests/cron/test_script_claim_heartbeat.py +++ b/tests/cron/test_script_claim_heartbeat.py @@ -883,41 +883,24 @@ def heartbeat(*_args, **_kwargs): calls.append(1) return True if len(calls) == 1 else None - def _run_job(job, *, defer_agent_teardown=None, extra_prompt=None, - cancel_event=None, execution_id=None): - # Sustained None heartbeats exhaust the grace window -> lost_ownership - # set as "uncertain, stop". Never an explicit False. - assert cancel_event is not None - assert cancel_event.wait(timeout=1.0) - return True, "output", "response", None + def run_body(_job, **kwargs): + # Stay contended well beyond the shortened grace window. Because every + # takeover also needs this fence, returned None is safe to defer until + # the side effect releases it; only exceptions consume error grace. + assert not kwargs["fire_claim_lost"].wait(timeout=0.12) + return True - finish = MagicMock() - mark = MagicMock(return_value=True) job = { "id": "grace-none", - "execution_id": "grace-none-exec", "fire_claim": {"at": "2026-07-12T12:00:00+00:00", "by": "owner"}, } monkeypatch.setattr(scheduler, "heartbeat_fire_claim", heartbeat) - monkeypatch.setattr(scheduler, "run_job", _run_job) + monkeypatch.setattr(scheduler, "_run_one_job_body", run_body) monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01) monkeypatch.setattr(scheduler, "_FIRE_CLAIM_HEARTBEAT_GRACE_SECONDS", 0.03) - monkeypatch.setattr(scheduler, "claim_dispatch", lambda *_a: True) - monkeypatch.setattr(scheduler, "mark_execution_running", lambda *_a: {}) - monkeypatch.setattr(scheduler, "mark_job_run", mark) - monkeypatch.setattr(scheduler, "finish_execution", finish) - - with patch("agent.secret_scope.set_secret_scope", return_value=None), \ - patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ - patch("agent.secret_scope.reset_secret_scope"): - assert scheduler.run_one_job(job) is True - errors = [call.kwargs.get("error") for call in finish.call_args_list] - assert errors, "execution ledger never closed" - for err in errors: - assert "ownership lost" not in err.lower(), ( - f"None probe adjudicated as confirmed ownership loss: {err!r}" - ) + assert scheduler.run_one_job(job) is True + assert len(calls) >= 4 def test_terminal_write_none_does_not_mark_ownership_lost(monkeypatch): From 7cf86bab8b6b16a87d03397226a93e8f7883a5fd Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Sat, 5 Sep 2026 12:29:17 +0700 Subject: [PATCH 8/8] fix(cron): restore fence-gated delivery invariants on refactored main paths Forward-port completion on top of the main-side cron refactor (scheduler phase helpers): the exception-path failure notification and the fence-local interrupted recheck before _deliver_result were lost in the refactor's _deliver_crash_failure extraction. Restored so a stale/uncertain worker can never emit a failure alert the replacement run will also send, and a shutdown that begins during the delivery fence can never leak the plausible final response as a success. --- cron/scheduler.py | 36 ++++++++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 5b7ba49633e7..1877f66e1d53 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2883,6 +2883,26 @@ def _save_compose_deliver( with fence.side_effect_fence() as owns_delivery: if not owns_delivery: raise _FireClaimLostDuringSideEffect + # Shutdown publishes the pending interrupt token before contending for this + # same fence. Recheck while fenced to close the window after the earlier + # delivery-gate check: a plausible response must not escape merely because + # shutdown began while we were waiting here. + if d.success and _is_interrupted(job["id"], execution_token): + d.success = False + d.error = ( + "Interrupted by gateway shutdown before the run finished " + "(tool subprocess was killed mid-flight)." + ) + d.incident_acked, d.failure_incident_id = _upsert_incident_for_failure( + job, d.error, output_file=output_file) + failure_content = ( + "" if d.incident_acked + else _summarize_cron_failure_for_delivery(job, d.error) + + _failure_streak_nudge(job)) + if not failure_content.strip(): + d.should_deliver = False + return + deliver_content = failure_content d.delivery_attempted = True d.delivery_error = _deliver_result( job, @@ -3174,8 +3194,20 @@ def _teardown_deferred() -> None: and not isinstance(e, _FireClaimLostDuringSideEffect) and not _fire_claim_ownership_lost() ): - delivery_error, delivery_outcome = _deliver_crash_failure( - job, _err_text, adapters=adapters, loop=loop) + # The ownership probe above is tri-state: None only means + # unconfirmed, never permission to emit. The side-effect + # fence is the exactly-once authority for delivery on this + # exception path, just as it is on the normal path. + with fence.side_effect_fence() as owns_delivery: + if owns_delivery: + delivery_error, delivery_outcome = _deliver_crash_failure( + job, _err_text, adapters=adapters, loop=loop) + else: + logger.warning( + "Job '%s': skipping exception delivery because " + "fire-claim ownership could not be fenced", + job["id"], + ) try: if not _consume_interrupted_flag(job["id"], execution_token): mark_kwargs = {}