From 31aba735b0ef0ac60aec68dc66a4f5d34485e9eb Mon Sep 17 00:00:00 2001 From: Sulaiman Ghori Date: Wed, 23 Sep 2026 21:55:19 -0700 Subject: [PATCH] fix(cron): keep late imports working across an in-place upgrade; record a released claim's outcome first MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two cron executions (b20a472b7e80 at 17:00, f98d49b4524e at 17:07) were marked failed with "cannot import name 'safe_strftime' from 'hermes_time'" while both workers kept running to completion (output written 18:19 / 18:57). A long-running process had hermes_time cached from the revision that predated safe_strftime, then imported a consumer for the first time after the tree was upgraded in place at 18:12, so the import needed a NEW name from a module it already had cached. Move safe_strftime/_repair_surrogates into their own stdlib-only leaf module (hermes_time re-exports them for existing callers) so a post-upgrade import loads them from disk instead of from the stale cached module. Second: sweep_stale_inflight dropped the in-flight claim in its classification loop and only afterwards wrote the release record; for an "age" release of a finite-repeat job it wrote no durable record at all. A released claim could therefore leave no answer to "did side effects run?" (#115692). Write the durable outcome BEFORE releasing — terminalising the claim's still-open ledger row via cron.executions.mark_claim_unknown — then release, re-checking the claim's start stamp so a fresh fire in the gap survives; if the write fails the claim is retained so the next sweep retries instead of losing the attempt. Tests: tests/cron/test_upgrade_module_skew.py reproduces the exact ImportError against a pre-upgrade hermes_time (and pins the failure signature), tests/cron/test_stale_claim_durable_terminal.py covers write-before-release ordering, fail-closed retention, and the durable ledger terminal state. --- agent/account_usage.py | 2 +- agent/auxiliary_unavailable.py | 2 +- agent/billing_usage.py | 2 +- agent/insights.py | 2 +- agent/learning_graph_render.py | 2 +- agent/system_prompt.py | 5 +- cron/executions.py | 58 +++++++ cron/quota_hold.py | 3 +- cron/scheduler.py | 47 +++++- gateway/message_timestamps.py | 2 +- hermes_cli/goals.py | 2 +- hermes_cli/status_auth.py | 2 +- hermes_time.py | 46 +----- hermes_time_format.py | 55 +++++++ .../cron/test_stale_claim_durable_terminal.py | 141 ++++++++++++++++++ tests/cron/test_upgrade_module_skew.py | 89 +++++++++++ tools/session_search_tool.py | 2 +- 17 files changed, 405 insertions(+), 57 deletions(-) create mode 100644 hermes_time_format.py create mode 100644 tests/cron/test_stale_claim_durable_terminal.py diff --git a/agent/account_usage.py b/agent/account_usage.py index e5989e26b92a0..97a700327a3e7 100644 --- a/agent/account_usage.py +++ b/agent/account_usage.py @@ -11,7 +11,7 @@ from agent.anthropic_credentials import _is_oauth_token, resolve_anthropic_token from hermes_cli.auth import AuthError, _read_codex_tokens, resolve_codex_runtime_credentials from hermes_cli.runtime_provider import resolve_runtime_provider -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime if TYPE_CHECKING: from typing import TypeGuard diff --git a/agent/auxiliary_unavailable.py b/agent/auxiliary_unavailable.py index bf8ee5ddbffd6..0b31e58432bfe 100644 --- a/agent/auxiliary_unavailable.py +++ b/agent/auxiliary_unavailable.py @@ -14,7 +14,7 @@ from datetime import datetime from typing import Optional -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime logger = logging.getLogger(__name__) diff --git a/agent/billing_usage.py b/agent/billing_usage.py index e3fc1ba7af077..66c1feb4db4ab 100644 --- a/agent/billing_usage.py +++ b/agent/billing_usage.py @@ -16,7 +16,7 @@ from dataclasses import dataclass from typing import Any, Optional -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime logger = logging.getLogger(__name__) diff --git a/agent/insights.py b/agent/insights.py index 53ac36a7c9a41..a251cb0730a5a 100644 --- a/agent/insights.py +++ b/agent/insights.py @@ -11,7 +11,7 @@ from agent.usage_pricing import CanonicalUsage, estimate_usage_cost, format_cost_label, format_duration_compact, has_known_pricing from hermes_cli.timefmt import coerce_epoch -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime _TOKEN_KEYS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens") _SKILL_TOOLS = {"skill_view", "skill_manage"} diff --git a/agent/learning_graph_render.py b/agent/learning_graph_render.py index cc735abfc95d9..faa7fdcc8406e 100644 --- a/agent/learning_graph_render.py +++ b/agent/learning_graph_render.py @@ -11,7 +11,7 @@ from datetime import datetime, timezone from typing import Any, Iterable, Optional -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime LEAD_IN = 0.06 # time-axis.ts LEAD_IN: the oldest node sits just off recency 0. # constants.ts AGE_GRADIENT — old quiet, recent bright. diff --git a/agent/system_prompt.py b/agent/system_prompt.py index 853462571cc29..70d6b34c9ebe9 100644 --- a/agent/system_prompt.py +++ b/agent/system_prompt.py @@ -474,7 +474,7 @@ def _zone_bits(now: Any, tz: Any) -> List[str]: """IANA key, abbreviation (if different) and UTC offset — all constant for the day, so the byte-stable date line stays cacheable.""" _iana = getattr(tz, "key", None) - from hermes_time import safe_strftime + from hermes_time_format import safe_strftime _abbrev = safe_strftime(now, "%Z") _offset = safe_strftime(now, "%z") # '-0400' -> 'UTC-04:00' bits = [_iana] if _iana else [] @@ -489,7 +489,8 @@ def _timestamp_line(agent: Any) -> str: """Date-only so the prompt is byte-stable for the day; zone + offset so tools needn't guess EST vs EDT. Long-lived sessions get an "as of" line on rebuild days (the cache prefix is already invalidated at that boundary).""" - from hermes_time import get_timezone as _hermes_tz, now as _hermes_now, safe_strftime + from hermes_time import get_timezone as _hermes_tz, now as _hermes_now + from hermes_time_format import safe_strftime now = _hermes_now() _bits = _zone_bits(now, _hermes_tz()) _zone_suffix = f" ({', '.join(_bits)})" if _bits else "" diff --git a/cron/executions.py b/cron/executions.py index 2f53c3b7e1ce0..43e9f779b5231 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -383,6 +383,64 @@ def recover_interrupted_executions() -> int: return changed +_STALE_CLAIM_RELEASED_REASON = ( + "Scheduler released this claim as stale without a terminal result from its owner; " + "whether side effects ran is unknown (#115692)." +) + + +def mark_claim_unknown( + job_id: str, *, claimed_after: Optional[float] = None, + reason: str = _STALE_CLAIM_RELEASED_REASON, +) -> Optional[Dict[str, Any]]: + """Durably terminalise the open row for ``job_id`` before its in-flight claim is released. + + ``cron.scheduler.sweep_stale_inflight`` drops the in-memory guard of a stale claim; if the + ledger still says 'claimed'/'running' afterwards, the attempt has no durable outcome and no one + can tell whether its side effects ran. This writes that terminal state first. Only the newest + open row for ``job_id`` is touched, and only when it was claimed at/after ``claimed_after`` (the + in-flight claim's start) so a row owned by a different attempt is left alone. Returns the + updated record, or None when nothing matched. + """ + now = _hermes_now().isoformat() + with _transaction() as conn: + rows = conn.execute( + """SELECT id, claimed_at FROM executions + WHERE job_id=? AND status IN ('claimed','running') + ORDER BY claimed_at DESC, id DESC""", + (str(job_id),), + ).fetchall() + target_id = None + for row in rows: + if claimed_after is None: + target_id = row["id"] + break + try: + claimed = datetime.fromisoformat(row["claimed_at"]) + if claimed.tzinfo is None: + claimed = claimed.astimezone() + if claimed.timestamp() >= claimed_after: + target_id = row["id"] + break + except (TypeError, ValueError): + continue + if target_id is None: + return None + cur = conn.execute( + """UPDATE executions + SET status='unknown', finished_at=?, error=?, handoff_pending=0, + handoff_started_at=NULL + WHERE id=? AND status IN ('claimed','running')""", + (now, reason, target_id), + ) + if cur.rowcount != 1: + return None + _prune_unlocked(conn) + record = _fetch(conn, target_id) + _emit_execution_state(record) + return record + + def list_executions( *, job_id: Optional[str] = None, limit: int = 50, before_claimed_at: Optional[str] = None, ) -> List[Dict[str, Any]]: diff --git a/cron/quota_hold.py b/cron/quota_hold.py index 6069cc1328b8b..75c79ab0c4d43 100644 --- a/cron/quota_hold.py +++ b/cron/quota_hold.py @@ -19,7 +19,8 @@ from datetime import datetime from typing import Any, Dict, Optional -from hermes_time import now as _hermes_now, safe_strftime +from hermes_time import now as _hermes_now +from hermes_time_format import safe_strftime logger = logging.getLogger("cron.scheduler") diff --git a/cron/scheduler.py b/cron/scheduler.py index 9e16f57c95517..0d05dfb8b2541 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -41,7 +41,8 @@ from hermes_cli.config import ( load_config, load_config_readonly) from hermes_cli.fallback_config import get_fallback_chain, scoped_fallback_chain -from hermes_time import now as _hermes_now, safe_strftime +from hermes_time import now as _hermes_now +from hermes_time_format import safe_strftime from agent.interrupt_compat import request_hard_interrupt from agent.delegation_context import ( enter_non_dispatcher_owned_context, exit_non_dispatcher_owned_context) @@ -985,13 +986,26 @@ def _record_stale_release(job: dict, job_id: str, age: float, allowance: float, logger.warning("Could not record forced release for job %s: %s", job_id, e) +def _terminalise_claim_if_open(job_id: str, claim_started: float) -> None: + """Write the terminal outcome an abandoned claim's owner never wrote, before it is released. + + Best-effort by contract: ``sweep_stale_inflight`` calls this ahead of the WARNING/JSONL/ + ``last_error`` record and treats any raise as "do not release the claim yet".""" + from cron.executions import mark_claim_unknown + + mark_claim_unknown(job_id, claimed_after=claim_started) + + def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list: """Force-release in-flight claims that can no longer be making progress; returns released ids. Stale = older than ``max(2 * interval, floor)`` AND (no live future — submit path hung before ``pool.submit`` returned — or finished without discarding the id) — or the claim's OWN ledger - row is terminal regardless of age. Each release logs WARNING ``event=forced_release``, bumps - the probe counter, mirrors JSONL, and writes ``last_error``. + row is terminal regardless of age. The durable outcome is written BEFORE the claim is released + (for an age release the still-open ledger row is terminalised first), so a released claim always + has an answer on disk for "did side effects run?"; a claim whose record cannot be written stays + held. Each release logs WARNING ``event=forced_release``, bumps the probe counter, mirrors JSONL, + and writes ``last_error``. """ global _forced_release_count @@ -1006,6 +1020,10 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list: # Compute intervals OUTSIDE _running_lock so croniter doesn't block try_register/release. _intervals = {jid: _job_interval_minutes(j) for jid, j in by_id.items()} + # Phase 1 — classify only, no mutation. The durable outcome of each release is written in + # phase 2 BEFORE the claim is dropped: after the release nobody can inspect the run any more, + # so "did side effects run?" must already be answerable from disk (#115692). + candidates: list = [] with _running_lock: for key in [k for k in _running_job_ids if k[0] == _local_home]: job_id = key[1] @@ -1039,15 +1057,32 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list: reason = "age" else: continue + candidates.append((key, started, job_id, age, allowance, fut, reason)) + + stale: list = [] + for key, started, job_id, age, allowance, fut, reason in candidates: + # Durable terminal state FIRST, claim release second. This ordering is also fail-closed: a + # claim whose outcome could not be written stays held (retried next sweep) instead of being + # dropped with no record of how it ended. + try: + if reason == "age": + _terminalise_claim_if_open(job_id, started) + _record_stale_release(by_id.get(job_id) or {}, job_id, age, allowance, fut, reason) + except Exception: + logger.exception( + "cron.inflight.forced_release.record_failed job_id=%s — claim retained", job_id) + continue + with _running_lock: + # Release only the claim that was classified: a fresh fire that registered while the + # record was being written must survive, so re-check its start stamp under the lock. + if _running_since.get(key) != started: + continue _running_job_ids.discard(key) _running_since.pop(key, None) _running_allowance_s.pop(key, None) _running_futures.pop(key, None) _forced_release_count += 1 stale.append((job_id, age, allowance, fut, reason)) - - for job_id, age, allowance, fut, _reason in stale: - _record_stale_release(by_id.get(job_id) or {}, job_id, age, allowance, fut, _reason) return [s[0] for s in stale] diff --git a/gateway/message_timestamps.py b/gateway/message_timestamps.py index 7f32fc0b94a60..dfb20731b4651 100644 --- a/gateway/message_timestamps.py +++ b/gateway/message_timestamps.py @@ -11,7 +11,7 @@ from datetime import datetime from typing import Any, Optional, Tuple -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime # Leading timestamp prefix, either the current human format diff --git a/hermes_cli/goals.py b/hermes_cli/goals.py index 500096ef87c85..40a3b34034f42 100644 --- a/hermes_cli/goals.py +++ b/hermes_cli/goals.py @@ -22,7 +22,7 @@ from typing import Any, Dict, List, Optional, Tuple from hermes_cli._subprocess_compat import noninteractive_git_env -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime logger = logging.getLogger(__name__) diff --git a/hermes_cli/status_auth.py b/hermes_cli/status_auth.py index 204af458789ef..a10bbbcb3010b 100644 --- a/hermes_cli/status_auth.py +++ b/hermes_cli/status_auth.py @@ -10,7 +10,7 @@ from hermes_cli.nous_subscription import get_nous_subscription_features from tools.tool_backend_helpers import managed_nous_tools_enabled from hermes_cli import config -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime def _format_iso_timestamp(value) -> str: diff --git a/hermes_time.py b/hermes_time.py index a71f2206294ef..6f28db00be1fe 100644 --- a/hermes_time.py +++ b/hermes_time.py @@ -5,16 +5,20 @@ time. Invalid timezone values log a warning and fall back — never crash. """ -import locale import logging import os -import re import threading from datetime import datetime from typing import Dict, Optional, Tuple from zoneinfo import ZoneInfo from hermes_constants import get_config_path +# ``safe_strftime``/``_repair_surrogates`` live in their own stdlib-only leaf module. A process +# that crosses an on-disk upgrade keeps THIS module cached from the pre-upgrade revision, so a +# consumer imported for the first time afterwards must not need a name from here that the cached +# copy lacks. Re-exported for compatibility with existing ``from hermes_time import +# safe_strftime`` callers. +from hermes_time_format import _repair_surrogates, safe_strftime # noqa: F401 logger = logging.getLogger(__name__) @@ -27,42 +31,6 @@ _cache_lock = threading.Lock() _tz_cache: Dict[Tuple[str, str], Tuple[str, Optional[ZoneInfo]]] = {} -_SURROGATE_RE = re.compile(r"[\ud800-\udfff]") -# ASCII plus surrogateescape'd bytes only: the shape of native text decoded with the wrong codec. -_ESCAPED_BYTES_RE = re.compile(r"[\x00-\x7f\udc80-\udcff]*") - - -def _repair_surrogates(text: str, encoding: Optional[str] = None) -> str: - """Make locale text JSON/UTF-8 safe; a no-op (same object) on valid text. - - Windows hands back zone names in the ANSI code page (``heure d'\\xe9t\\xe9``) but a UTF-8 - ``LC_CTYPE`` (UTF-8 mode, or a library flipping the process locale mid-run) decodes those - bytes with ``surrogateescape`` into lone surrogates. Recover the bytes and decode them with - the ANSI code page; when that is not possible, replace each surrogate with U+FFFD.""" - if text.isascii() or not _SURROGATE_RE.search(text): - return text - if _ESCAPED_BYTES_RE.fullmatch(text): - try: - return text.encode("ascii", "surrogateescape").decode(encoding or locale.getencoding()) - except (LookupError, UnicodeError): - pass - return _SURROGATE_RE.sub("\ufffd", text) - - -def safe_strftime(value: datetime, fmt: str) -> str: - """``value.strftime(fmt)`` that never raises or returns lone surrogates over locale text. - - ``datetime.strftime`` splices ``tzname()`` into the format as UTF-8, so a zone name carrying - surrogates (see ``_repair_surrogates``) raises ``UnicodeEncodeError`` before any string exists - (#102910). On that failure ``%Z`` is rendered from the repaired name instead. Output for valid - locale text is byte-identical to ``strftime`` (the system prompt must stay cache-stable).""" - try: - return _repair_surrogates(value.strftime(fmt)) - except UnicodeEncodeError: - zone = _repair_surrogates(value.tzname() or "").replace("%", "%%") - return _repair_surrogates(value.strftime( - re.sub(r"%([%Z])", lambda m: zone if m.group(1) == "Z" else "%%", fmt))) - def _env_timezone() -> str: """``HERMES_TIMEZONE`` when it may speak for the active profile. Under the multiplexed @@ -146,4 +114,4 @@ def reset_cache() -> None: def now() -> datetime: """Current time as a tz-aware datetime: configured zone, else server-local.""" tz = get_timezone() - return datetime.now(tz) if tz is not None else datetime.now().astimezone() + return datetime.now(tz) if tz is not None else datetime.now().astimezone() \ No newline at end of file diff --git a/hermes_time_format.py b/hermes_time_format.py new file mode 100644 index 0000000000000..1e1010d6c21a4 --- /dev/null +++ b/hermes_time_format.py @@ -0,0 +1,55 @@ +"""Locale-safe datetime formatting helpers for Hermes. + +Split out of :mod:`hermes_time` so a long-running process that crossed an on-disk upgrade can +import this module fresh and still get ``safe_strftime`` even though ``hermes_time`` is already +cached in ``sys.modules`` from the revision that predated it. ``hermes_time`` re-exports these +names for compatibility. + +Keep this module a stdlib-only leaf: importing ``hermes_time`` (or anything that imports it) +here would recreate the stale-cached-module hazard this split exists to remove — a module +imported for the first time *after* an upgrade must never need new names from a module the +process already has cached. +""" + +from __future__ import annotations + +import locale +import re +from datetime import datetime +from typing import Optional + +_SURROGATE_RE = re.compile(r"[\ud800-\udfff]") +# ASCII plus surrogateescape'd bytes only: the shape of native text decoded with the wrong codec. +_ESCAPED_BYTES_RE = re.compile(r"[\x00-\x7f\udc80-\udcff]*") + + +def _repair_surrogates(text: str, encoding: Optional[str] = None) -> str: + """Make locale text JSON/UTF-8 safe; a no-op (same object) on valid text. + + Windows hands back zone names in the ANSI code page (``heure d'\\xe9t\\xe9``) but a UTF-8 + ``LC_CTYPE`` (UTF-8 mode, or a library flipping the process locale mid-run) decodes those + bytes with ``surrogateescape`` into lone surrogates. Recover the bytes and decode them with + the ANSI code page; when that is not possible, replace each surrogate with U+FFFD.""" + if text.isascii() or not _SURROGATE_RE.search(text): + return text + if _ESCAPED_BYTES_RE.fullmatch(text): + try: + return text.encode("ascii", "surrogateescape").decode(encoding or locale.getencoding()) + except (LookupError, UnicodeError): + pass + return _SURROGATE_RE.sub("\ufffd", text) + + +def safe_strftime(value: datetime, fmt: str) -> str: + """``value.strftime(fmt)`` that never raises or returns lone surrogates over locale text. + + ``datetime.strftime`` splices ``tzname()`` into the format as UTF-8, so a zone name carrying + surrogates (see ``_repair_surrogates``) raises ``UnicodeEncodeError`` before any string exists + (#102910). On that failure ``%Z`` is rendered from the repaired name instead. Output for valid + locale text is byte-identical to ``strftime`` (the system prompt must stay cache-stable).""" + try: + return _repair_surrogates(value.strftime(fmt)) + except UnicodeEncodeError: + zone = _repair_surrogates(value.tzname() or "").replace("%", "%%") + return _repair_surrogates(value.strftime( + re.sub(r"%([%Z])", lambda m: zone if m.group(1) == "Z" else "%%", fmt))) \ No newline at end of file diff --git a/tests/cron/test_stale_claim_durable_terminal.py b/tests/cron/test_stale_claim_durable_terminal.py new file mode 100644 index 0000000000000..e993355042aa4 --- /dev/null +++ b/tests/cron/test_stale_claim_durable_terminal.py @@ -0,0 +1,141 @@ +"""A released stale claim must already have a durable terminal outcome on disk. + +``sweep_stale_inflight`` used to mutate the in-flight guard (``_running_job_ids`` and friends) in +its classification loop and only afterwards write the release record — and for an ``age`` release of +a job with a finite repeat budget it deliberately wrote no ``last_error`` at all. A released claim +whose ledger row was still ``claimed``/``running`` therefore left nobody able to say whether its +side effects ran (#115692: "Owner process is still alive but the claim outlived the derived stale +bound ... whether side effects ran is unknown"). + +These tests pin the ordering: the durable outcome is written BEFORE the claim is dropped, an open +ledger row is terminalised first, and a claim whose outcome cannot be written stays held (fail +closed) instead of being silently released. +""" + +from __future__ import annotations + +import time +from unittest.mock import patch + +import pytest + +import cron.scheduler as sched +from cron import executions as executions_mod + + +def _job(job_id="wedged", minutes=60): + return { + "id": job_id, + "name": job_id, + "schedule": {"kind": "interval", "minutes": minutes, "display": f"every {minutes}m"}, + } + + +@pytest.fixture(autouse=True) +def _clean_inflight(): + sched._running_job_ids.clear() + sched._running_since.clear() + sched._running_futures.clear() + yield + sched._running_job_ids.clear() + sched._running_since.clear() + sched._running_futures.clear() + + +def _claim(job, home, *, age_seconds=5 * 60 * 60): + """Register an aged, submit-hung claim for ``job`` and return (key, started).""" + key = sched._inflight_key(job["id"], home) + started = time.time() - age_seconds + sched._running_job_ids.add(key) + sched._running_since[key] = started + sched._running_futures[key] = sched._FUTURE_PENDING + return key, started + + +def test_durable_outcome_is_written_before_the_claim_is_released(tmp_path): + """The claim is still held while its outcome is recorded — after the release nothing can + inspect the run, so the answer must already be on disk.""" + job = _job() + key, started = _claim(job, tmp_path) + seen = {} + + def _record(job_id, claim_started): + seen["held"] = key in sched._running_job_ids + seen["started"] = sched._running_since.get(key) + seen["job_id"] = job_id + seen["claim_started"] = claim_started + + with patch.object(sched, "_terminalise_claim_if_open", side_effect=_record), \ + patch.object(sched, "mark_job_run"), \ + patch.object(sched, "_get_hermes_home", return_value=tmp_path): + assert sched.sweep_stale_inflight([job]) == [job["id"]] + + assert seen == {"held": True, "started": started, "job_id": job["id"], + "claim_started": started} + assert key not in sched._running_job_ids + + +@pytest.mark.parametrize("failing", ["_terminalise_claim_if_open", "_record_stale_release"]) +def test_claim_is_retained_when_the_durable_outcome_cannot_be_written(tmp_path, failing): + """Fail closed: a claim whose outcome could not be persisted is NOT released — the next sweep + retries rather than losing the attempt with no record of how it ended.""" + job = _job() + key, _started = _claim(job, tmp_path) + + patches = [ + patch.object(sched, "mark_job_run"), + patch.object(sched, "_get_hermes_home", return_value=tmp_path), + ] + if failing == "_terminalise_claim_if_open": + patches.append(patch.object(sched, "_terminalise_claim_if_open", + side_effect=RuntimeError("ledger locked"))) + else: + patches.append(patch.object(sched, "_terminalise_claim_if_open")) + patches.append(patch.object(sched, "_record_stale_release", + side_effect=RuntimeError("jsonl unwritable"))) + + from contextlib import ExitStack + with ExitStack() as stack: + for p in patches: + stack.enter_context(p) + assert sched.sweep_stale_inflight([job]) == [] + + assert key in sched._running_job_ids + assert job["id"] in sched.get_running_job_ids() + + +def test_age_release_terminalises_the_open_ledger_row(tmp_path): + """An age release leaves the ledger with a terminal 'unknown' row and the #115692 reason, so a + reader can tell the attempt ended and that its side effects are unknown.""" + job = _job() + row = executions_mod.create_execution(job["id"], source="builtin") + assert row["status"] == "claimed" + key, _started = _claim(job, tmp_path) + + with patch.object(sched, "mark_job_run"), \ + patch.object(sched, "_get_hermes_home", return_value=tmp_path): + assert sched.sweep_stale_inflight([job]) == [job["id"]] + + after = executions_mod.get_execution(row["id"]) + assert after["status"] == "unknown" + assert after["finished_at"] + assert "115692" in after["error"] + assert key not in sched._running_job_ids + + +def test_mark_claim_unknown_leaves_a_row_from_a_different_claim_alone(tmp_path): + """Only the claim's own row is terminalised: a newer attempt's row must not be reaped.""" + row = executions_mod.create_execution("job-x", source="builtin") + + assert executions_mod.mark_claim_unknown("job-x", claimed_after=time.time() + 3600) is None + assert executions_mod.get_execution(row["id"])["status"] == "claimed" + + # ... and the matching claim's row is the one that gets the terminal state. + updated = executions_mod.mark_claim_unknown("job-x", claimed_after=time.time() - 3600, + reason="released stale (#115692)") + assert updated is not None and updated["id"] == row["id"] + assert updated["status"] == "unknown" + assert updated["error"] == "released stale (#115692)" + # Terminal rows are immutable: a second call finds nothing open to touch. + assert executions_mod.mark_claim_unknown("job-x", claimed_after=time.time() - 3600) is None + assert executions_mod.get_execution(row["id"])["error"] == "released stale (#115692)" \ No newline at end of file diff --git a/tests/cron/test_upgrade_module_skew.py b/tests/cron/test_upgrade_module_skew.py index bd0bcdfbc1392..d19ff9fb02895 100644 --- a/tests/cron/test_upgrade_module_skew.py +++ b/tests/cron/test_upgrade_module_skew.py @@ -74,3 +74,92 @@ def test_occurrences_resolve_fire_claim_constants_without_cached_jobs_exports(): ) assert result.returncode == 0, result.stderr + + +# A daemon that crossed the 2026-09-23 upgrade mid-run kept ``hermes_time`` cached from the revision +# that predated ``safe_strftime`` (it had no such name), then lazy-imported a consumer from the newly +# checked-out tree: ``ImportError: cannot import name 'safe_strftime' from 'hermes_time' +# (/var/lib/hermes/src/hermes_time.py)`` marked two cron runs failed (jobs b20a472b7e80 / +# f98d49b4524e) even though the workers kept running to completion. +_SAFE_STRFTIME_SKEW_SCRIPT = """ +import sys + +import hermes_time + +# The pre-upgrade hermes_time had no formatting helpers at all. +for name in ("safe_strftime", "_repair_surrogates"): + if hasattr(hermes_time, name): + delattr(hermes_time, name) +assert not hasattr(hermes_time, "safe_strftime") + +consumers = [ + "cron.scheduler", + "cron.quota_hold", + "agent.learning_graph_render", + "agent.insights", + "agent.billing_usage", + "agent.auxiliary_unavailable", + "agent.account_usage", + "agent.system_prompt", + "hermes_cli.status_auth", + "hermes_cli.goals", + "gateway.message_timestamps", + "tools.session_search_tool", +] +for module in consumers: + sys.modules.pop(module, None) + try: + __import__(module) + except ImportError as exc: + # An unrelated missing optional dep must not mask the regression under test. + if "safe_strftime" in str(exc): + raise + print("skipped", module, exc) + +from hermes_time_format import _repair_surrogates, safe_strftime + +assert safe_strftime.__module__ == "hermes_time_format" +assert _repair_surrogates is not None +print("ok") +""" + + +def test_safe_strftime_consumers_import_against_pre_upgrade_hermes_time(): + """The helper lives in its own leaf module, so a consumer imported after the upgrade loads it + from disk even though the cached ``hermes_time`` predates it.""" + repo_root = Path(__file__).resolve().parents[2] + result = subprocess.run( + [sys.executable, "-c", _SAFE_STRFTIME_SKEW_SCRIPT], + cwd=repo_root, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + + +def test_pre_upgrade_hermes_time_still_fails_when_a_consumer_imports_it_directly(): + """Guard the guard: the test above is only meaningful while importing the name from the cached + module really does raise, i.e. the failure signature being reproduced is the real one.""" + script = """ +import hermes_time +delattr(hermes_time, "safe_strftime") +try: + from hermes_time import safe_strftime # noqa: F401 +except ImportError as exc: + assert "cannot import name 'safe_strftime'" in str(exc), exc + print("reproduced") +else: + raise AssertionError("stale hermes_time should not export safe_strftime") +""" + repo_root = Path(__file__).resolve().parents[2] + result = subprocess.run( + [sys.executable, "-c", script], + cwd=repo_root, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr diff --git a/tools/session_search_tool.py b/tools/session_search_tool.py index fd73bd3fc588f..ff7e591ee80ab 100644 --- a/tools/session_search_tool.py +++ b/tools/session_search_tool.py @@ -16,7 +16,7 @@ from typing import Any, Dict, List, Optional, Union from hermes_state_common import _BOUNDARY_END_REASONS -from hermes_time import safe_strftime +from hermes_time_format import safe_strftime # Hidden from browsing/searching — integrations (HERMES_SESSION_SOURCE=tool), delegate # subagent runs, kanban workers are not the user's history.