fix(cron): late imports survive an in-place upgrade; a released claim records its outcome first

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

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__)
 

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__)
 

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"}

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.

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 ""

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]]:

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")
 

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]
 
 

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

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__)
 

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:

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()

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)))