fix(cron): pair output docs with attempts after sessions are known

An attempt claimed before or in the same second as another run's output
doc (a manual `hermes cron run` overlapping a tick) became the doc's
newest candidate. When it failed without a session, the doc rendered as
an extra cron_output row with that attempt's status, and the attempt's
own failed row stayed hidden because representation was decided before
reconciliation.

_list_cron_output_runs now returns docs and terminal ledger attempts
separately; _reconcile_cron_runs pairs each doc with its likeliest
writer once every attempt's session is known (an attempt closed before
the doc only reaches it through the grace; one whose session spans the
doc beats an overlapping one without; then the newest claim), so each
attempt gets exactly one row: its session, else its doc, else its
ledger row. The internal _execution_candidates/_execution_id row keys
are gone.

Also: the cron request partial index is built from CRON_PRINCIPAL_ID;
the admission SQL derives its substr offsets from CRON_REQUEST_PREFIX;
a prefix_range() helper replaces three hand-rolled half-open bounds;
the two cron run-history suites share tests/fakes/cron_owner.py.

Refs #134889 (CRON-1)
Refs #139162
Refs #138812
Addresses review by @kshitijk4poor on #139273.

hermes_cli/web_routers/cron.py

@@ -353,20 +353,20 @@ def _list_cron_output_runs(
     canonical_job_id: str,
     profile: Optional[str],
     limit: int,
-) -> list[dict[str, Any]]:
-    """SessionDB-less run history for jobs that never create agent sessions.
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """SessionDB-less run records for jobs that never create agent sessions: ``(docs, attempts)``.
 
     Script-only (no_agent) jobs deliberately skip SessionDB (cron/scheduler.run_job),
     so their completed runs are invisible to the run-history endpoint. Their
-    output docs — one .md per fire under cron/output/<job_id>/ — are the only
-    per-run record. Rows mirror /api/sessions shape with source='cron_output'
-    so the frontend reuses SessionInfo; ids use a cron_output: prefix that can
-    never collide with a SessionDB session id.
-
-    Status comes from the execution ledger (one row per fire), matched to a doc
-    by claim window — never from ``last_run_at`` proximity, which mislabels an
-    older doc when the newest run's doc is missing. Terminal attempts with no
-    surviving doc get their own metadata rows.
+    output docs (one .md per fire under cron/output/<job_id>/) and the execution
+    ledger (one row per fire) are the only per-run records. ``docs`` are rows in
+    /api/sessions shape with source='cron_output', newest first, so the frontend
+    reuses SessionInfo; ids use a cron_output: prefix that can never collide with a
+    SessionDB session id. ``attempts`` are the job's terminal ledger rows, newest
+    first. A doc never records its execution, so which attempt it stands for is
+    decided by ``_reconcile_cron_runs`` beside the sessions, never here, and never
+    from ``last_run_at`` proximity, which mislabels an older doc when the newest
+    run's doc is missing.
     """
     output_dir = _cron_output_runs_dir(profile, canonical_job_id)
     try:
@@ -378,10 +378,8 @@ def _list_cron_output_runs(
     except OSError:
         files = []
 
-    executions = _owner_profile_executions(canonical_job_id)
-    represented: set = set()
-    runs: list[dict[str, Any]] = []
-
+    attempts = _owner_profile_executions(canonical_job_id)
+    docs: list[dict[str, Any]] = []
     for path in files[:limit]:
         started_at = _cron_output_run_timestamp(path)
         if started_at is None:
@@ -390,57 +388,20 @@ def _list_cron_output_runs(
             except OSError:
                 started_at = 0.0
         preview = _cron_output_run_preview(path)
-        title = preview or "Script-only run"
-        candidates = [
-            (index, attempt) for index, attempt in enumerate(executions)
-            if attempt["status"] in ("completed", "failed") and _execution_contains(attempt, started_at)
-        ]
-        if candidates:
-            # Newest first: the latest attempt claimed by the doc's second is the likeliest writer.
-            index, attempt = candidates[0]
-            represented.add(index)
-            title = _execution_status_title(attempt["status"], attempt["error"], title)
-        runs.append({
+        docs.append({
             "id": f"cron_output:{canonical_job_id}:{path.stem}",
-            **_cron_output_run_row(started_at, title, preview),
-            # Internal: a doc never records its execution, so this is a window guess — every
-            # attempt it could be, for the caller to pair with a session by time, never by id.
-            "_execution_candidates": tuple(attempt["id"] for _, attempt in candidates),
+            **_cron_output_run_row(started_at, preview or "Script-only run", preview),
         })
 
-    # Terminal attempts whose output doc is gone (pruned, or the run never
-    # wrote one — e.g. a failed fire) are still real executions: give each its
-    # own metadata row from the ledger instead of letting the latest attempt's
-    # status bleed onto an older surviving document.
-    for index, attempt in enumerate(executions):
-        if index in represented or attempt["status"] not in ("completed", "failed"):
-            continue
-        if len(runs) >= limit:
-            break
-        when = attempt["finished_at"] or attempt["claimed_at"]
-        error = attempt["error"]
-        runs.append({
-            "id": f"cron_output:{canonical_job_id}:exec:{index}",
-            **_cron_output_run_row(
-                when,
-                _execution_status_title(attempt["status"], error, "Script-only run"),
-                error or None,
-            ),
-            # Internal: the attempt itself (durable), so the caller drops this row
-            # exactly when a session already represents the same execution.
-            "_execution_id": attempt["id"],
-        })
-    runs.sort(key=lambda r: float(r.get("started_at") or 0), reverse=True)
-
-    if runs:
-        return runs
+    if docs or any(attempt["status"] in ("completed", "failed") for attempt in attempts):
+        return docs, attempts
 
     # No output docs survived (pruned or never written) but the job HAS run:
     # surface one metadata-only row from last_run_at/last_status instead of the
     # bare "No runs" the issue reports.
     latest_ts = _cron_job_last_run_timestamp(job)
     if latest_ts is None:
-        return []
+        return [], attempts
 
     preview = ""
     if isinstance(job, dict):
@@ -451,7 +412,7 @@ def _list_cron_output_runs(
     return [{
         "id": f"cron_output:{canonical_job_id}:latest",
         **_cron_output_run_row(latest_ts, title, preview),
-    }]
+    }], attempts
 
 
 # A legacy run's session row is written just AFTER its execution claim; the grace only
@@ -565,62 +526,94 @@ def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit:
             if selected:
                 s["profile"] = selected
 
-        doc_runs = _list_cron_output_runs(job, canonical, selected, limit_n)
+        docs, attempts = _list_cron_output_runs(job, canonical, selected, limit_n)
 
-    return {"runs": _reconcile_cron_runs(session_runs, doc_runs, limit_n), "limit": limit_n}
+    return {"runs": _reconcile_cron_runs(session_runs, docs, attempts, canonical, limit_n), "limit": limit_n}
+
+
+# A doc within this of a session's span can be that session's run's doc (see _doc_matches_session).
+_SESSION_DOC_MATCH_SECONDS = 300.0
 
 
 def _reconcile_cron_runs(
     session_runs: list[dict[str, Any]],
-    doc_runs: list[dict[str, Any]],
+    docs: list[dict[str, Any]],
+    attempts: list[dict[str, Any]],
+    canonical_job_id: str,
     limit: int,
 ) -> list[dict[str, Any]]:
-    """Merge session rows and output-doc rows per execution, newest first.
+    """One row per execution, newest first: each terminal ledger attempt is its session, else the
+    output doc it wrote, else a metadata row from the ledger.
 
-    An agent fire writes BOTH a session and an output doc for the same
-    execution; keeping both would double-render the run. Sessions carry the
-    richer record (tokens, message counts, chat navigation), so the session
-    wins. Only a durable association pairs by id: a unified-runtime session
-    names its execution and a ledger row IS its attempt, so a neighbouring fire
-    inside the time window (one that failed before it created a session) keeps
-    its own row. An output doc never records its execution, only a window guess
-    (``_execution_candidates``), so it pairs by time: a doc within
-    ``_SESSION_DOC_MATCH_SECONDS`` of a legacy session, or of the session whose
-    execution is its likeliest writer (the newest candidate, whose ledger row the
-    doc already stands in for), is that session's execution (a doc no attempt
-    claims is matched against every session). Doc rows matching no
-    session are distinct executions — script-only fires — and are kept. A session
-    row must never be dropped in favor of a doc: the doc is the fallback
-    representation.
+    An agent fire writes BOTH a session and an output doc; sessions carry the richer
+    record (tokens, message counts, chat navigation), so the session wins and is never
+    dropped. A unified-runtime session names its execution, so that pairing is exact.
+    A doc only has a window guess (``_execution_contains``), so docs pair with attempts
+    only once every attempt's session is known (``_doc_writer``), one doc per attempt:
+    the doc folds into its writer's session, or into a legacy session (which names no
+    execution) by time, and otherwise is its writer's row with the ledger status. A doc
+    no attempt claims is matched against every session by time, and kept if none matches.
     """
-    _SESSION_DOC_MATCH_SECONDS = 300.0
+    def matches(doc_ts: float, sessions: list[dict[str, Any]]) -> bool:
+        return any(_doc_matches_session(doc_ts, s, _SESSION_DOC_MATCH_SECONDS) for s in sessions)
 
-    executions = {s["cron_execution_id"] for s in session_runs if s.get("cron_execution_id")}
+    by_execution = {s["cron_execution_id"]: s for s in session_runs if s.get("cron_execution_id")}
     legacy = [s for s in session_runs if not s.get("cron_execution_id")]
+    terminal = [(index, a) for index, a in enumerate(attempts) if a["status"] in ("completed", "failed")]
+    represented = {index for index, a in terminal if a["id"] in by_execution}
+    paired: set = set()
     merged = list(session_runs)
-    for doc in doc_runs:
-        candidates = doc.get("_execution_candidates")
-        execution_id = doc.get("_execution_id")
-        if candidates:
-            # The likeliest writer is the doc's newest candidate, the attempt _list_cron_output_runs
-            # folded into it. When no unified session names that attempt, the doc is its only row:
-            # only a legacy session can still claim it by time.
-            window = legacy + [s for s in session_runs if s.get("cron_execution_id") == candidates[0]]
-        elif candidates is None and execution_id:
-            if execution_id in executions:
-                continue
-            window = legacy
-        else:
-            window = session_runs
-        doc = {k: v for k, v in doc.items() if not k.startswith("_")}
+
+    for doc in docs:
         doc_ts = float(doc.get("started_at") or 0)
-        if any(_doc_matches_session(doc_ts, s, _SESSION_DOC_MATCH_SECONDS) for s in window):
+        candidates = [(index, a) for index, a in terminal
+                      if index not in paired and _execution_contains(a, doc_ts)]
+        if not candidates:
+            if not matches(doc_ts, session_runs):
+                merged.append(doc)
             continue
-        merged.append(doc)
+        index, attempt = min(candidates, key=lambda c: _doc_writer(doc_ts, c[1], by_execution.get(c[1]["id"])))
+        paired.add(index)
+        represented.add(index)
+        if attempt["id"] in by_execution or matches(doc_ts, legacy):
+            continue
+        merged.append({**doc, "title": _execution_status_title(attempt["status"], attempt["error"], doc["title"])})
+
+    # Attempts with neither a session nor a surviving doc (pruned, or the run never wrote one,
+    # e.g. a failed fire) are still real executions: each keeps its own ledger row instead of
+    # letting a neighbour's status bleed onto its doc. Only a legacy session can claim one by time.
+    for index, attempt in terminal:
+        if index in represented:
+            continue
+        when = attempt["finished_at"] or attempt["claimed_at"]
+        if matches(when, legacy):
+            continue
+        error = attempt["error"]
+        merged.append({
+            "id": f"cron_output:{canonical_job_id}:exec:{index}",
+            **_cron_output_run_row(
+                when, _execution_status_title(attempt["status"], error, "Script-only run"), error or None),
+        })
     merged.sort(key=lambda r: float(r.get("started_at") or 0), reverse=True)
     return merged[:limit]
 
 
+def _doc_writer(doc_ts: float, attempt: dict[str, Any], session: Optional[dict[str, Any]]) -> tuple:
+    """Sort key over a doc's candidate attempts, likeliest writer first.
+
+    ``save_job_output`` runs before ``finish_execution``, so an attempt that closed before
+    the doc's second only reaches it through the grace. Among the rest, an attempt whose
+    own session spans the doc beats an overlapping one that has none (a manual run claimed
+    while a tick's run was still going); then the newest claim wins.
+    """
+    finish = attempt["finished_at"]
+    return (
+        finish is not None and doc_ts > finish,
+        session is None or not _doc_matches_session(doc_ts, session, _SESSION_DOC_MATCH_SECONDS),
+        -attempt["claimed_at"],
+    )
+
+
 def _doc_matches_session(doc_ts: float, session: dict[str, Any], grace_seconds: float) -> bool:
     """Whether an output doc belongs to a session's run.
 

hermes_state_common.py

@@ -11,6 +11,7 @@ import time
 from typing import Any
 
 from hermes_cli.timefmt import EPOCH_MAX, EPOCH_MIN
+from hermes_state_keys import CRON_PRINCIPAL_ID
 from hermes_state_pidns import persistent_record_pidns_checkable, pid_namespace_id
 from agent.skill_commands import AUTO_LOAD_SCAFFOLD_SQL_LIKE, SKILL_EXCERPT_JOINT, SKILL_SCAFFOLD_SQL_LIKE, describe_skill_invocation
 from agent.context_compressor import (LEGACY_SUMMARY_PREFIX, SUMMARY_PREFIX, _MERGED_PRIOR_CONTEXT_HEADER,
@@ -528,12 +529,13 @@ CREATE TABLE IF NOT EXISTS session_admissions (
 CREATE INDEX IF NOT EXISTS session_admissions_pending
     ON session_admissions(target_session_id, status, seq);
 -- Cron run history finds a job's unified-runtime runs by the scheduler principal's request ids
--- (``cron:<job>:<execution>``, hermes_state_keys.CRON_PRINCIPAL_ID); without this a lookup walked
+-- (``cron:<job>:<execution>``); without this a lookup walked
 -- every cron admission of every job. Partial: interactive turns, the bulk of admissions, pay
 -- nothing for it. The constant leading column gives the planner equality + range over a covering
 -- index, so it beats the UNIQUE autoindex's bare principal_id prefix even without ANALYZE stats.
+-- The WHERE is built from CRON_PRINCIPAL_ID, the literal the run queries inline, so they cannot drift.
 CREATE INDEX IF NOT EXISTS idx_session_admissions_cron_request
-    ON session_admissions(principal_id, request_id, target_session_id) WHERE principal_id = 'cron-owner';
+    ON session_admissions(principal_id, request_id, target_session_id) WHERE principal_id = '""" + CRON_PRINCIPAL_ID + """';
 CREATE TABLE IF NOT EXISTS worker_executions (
     execution_id TEXT PRIMARY KEY,
     session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE RESTRICT,

hermes_state_keys.py

@@ -13,19 +13,24 @@ import re
 # id, so its admission row is the run's durable job/execution association. Runs from before the
 # unified runtime were named ``cron_{job_id}_{YYYYmmdd_HHMMSS}`` (``cron/scheduler.py::run_job``).
 CRON_PRINCIPAL_ID = 'cron-owner'
-_CRON_REQUEST_PREFIX = 'cron:'
+CRON_REQUEST_PREFIX = 'cron:'
 _LEGACY_CRON_RUN_ID = re.compile(r'^cron_(.+)_\d{8}_\d{6}$')
 
 
 def cron_request_id(job_id, execution_id):
     """The admission request id of one cron fire (``execution_id`` is its ledger row id)."""
-    return f'{_CRON_REQUEST_PREFIX}{job_id}:{execution_id}'
+    return f'{CRON_REQUEST_PREFIX}{job_id}:{execution_id}'
+
+
+def prefix_range(prefix):
+    """Half-open ``[prefix, hi)`` bounds of every key starting with *prefix*, for an index range
+    scan: bump the final character so the range covers exactly the prefix."""
+    return prefix, prefix[:-1] + chr(ord(prefix[-1]) + 1)
 
 
 def cron_request_range(job_id):
     """Half-open ``[lo, hi)`` bounds of *job_id*'s cron request ids, for an index range scan."""
-    lo = cron_request_id(job_id, '')
-    return lo, lo[:-1] + chr(ord(lo[-1]) + 1)
+    return prefix_range(cron_request_id(job_id, ''))
 
 
 def cron_run_from_request(request_id, job_id=None):
@@ -35,10 +40,10 @@ def cron_run_from_request(request_id, job_id=None):
     prefix splits there. More than one is ambiguous (a job id holding ``:``, or an execution id
     admitted with one before that was refused): only the *job_id* its stored policy names splits
     it, and must prefix it."""
-    if not isinstance(request_id, str) or not request_id.startswith(_CRON_REQUEST_PREFIX):
+    if not isinstance(request_id, str) or not request_id.startswith(CRON_REQUEST_PREFIX):
         return None
     if job_id is None:
-        rest = request_id[len(_CRON_REQUEST_PREFIX):]
+        rest = request_id[len(CRON_REQUEST_PREFIX):]
         if rest.count(':') != 1:
             return None
         job_id, _, execution_id = rest.partition(':')

hermes_state_media.py

@@ -2,6 +2,8 @@
 import hashlib
 import json
 
+from hermes_state_keys import prefix_range
+
 PREFIX = 'gateway.media.retired.v1.'
 
 
@@ -22,7 +24,7 @@ def collect_retired_media(db):
     with db._read_ctx() as conn:
         # Primary-key range, not LIKE: the common no-candidate case costs one index probe.
         rows = conn.execute('SELECT key,value FROM state_meta WHERE key>=? AND key<?',
-                            (PREFIX, PREFIX[:-1] + chr(ord(PREFIX[-1]) + 1))).fetchall()
+                            prefix_range(PREFIX)).fetchall()
     if rows:
         release_unheld_media(db, [json.loads(row['value']) for row in rows])
         from pathlib import Path

hermes_state_portability.py

@@ -16,7 +16,10 @@ from agent.skill_commands import AUTO_LOAD_SCAFFOLD_SQL_LIKE, SKILL_SCAFFOLD_SQL
 from utils import safe_json_loads
 from hermes_cli.timefmt import coerce_epoch
 from hermes_state_ids import new_session_id
-from hermes_state_keys import CRON_PRINCIPAL_ID, cron_request_range, cron_run_from_request, legacy_cron_run_job
+from hermes_state_keys import (
+    CRON_PRINCIPAL_ID, CRON_REQUEST_PREFIX, cron_request_range, cron_run_from_request, legacy_cron_run_job,
+    prefix_range,
+)
 from hermes_state_local import POLICY_PREFIX
 from hermes_state_common import SCHEMA_SQL, _shape_preview, _sql_preview_raw, _sql_session_last_active
 from hermes_state_messages import _parse_tool_calls, _tool_calls_count
@@ -102,7 +105,7 @@ _CRON_POLICY_JOB_SQL = (
     f" FROM state_meta WHERE key = '{POLICY_PREFIX}' || target_session_id))")
 _CRON_ADMISSION_SQL = (
     f"principal_id = '{CRON_PRINCIPAL_ID}' AND request_id >= :request_lo AND request_id < :request_hi"
-    " AND ((instr(substr(request_id, 6), ':') = length(:request_lo) - 5"
+    f" AND ((instr(substr(request_id, {len(CRON_REQUEST_PREFIX) + 1}), ':') = length(:request_lo) - {len(CRON_REQUEST_PREFIX)}"
     " AND instr(substr(request_id, length(:request_lo) + 1), ':') = 0)"
     f" OR {_CRON_POLICY_JOB_SQL} = :job_id)")
 _CRON_RUN_IDS_SQL = (
@@ -256,9 +259,7 @@ class SessionPortabilityMixin:
         Limitation (legacy runs had it too): only a run's FIRST session is listed. A run that
         compressed continues in a child session this does not return, and the row's liveness
         (``scheduler_owned``) and detail identity follow that first session, not the tip."""
-        prefix = f"cron_{job_id}_"
-        # Half-open upper bound: bump the final byte so the range covers exactly the prefix.
-        prefix_hi = prefix[:-1] + chr(ord(prefix[-1]) + 1)
+        prefix, prefix_hi = prefix_range(f"cron_{job_id}_")
         request_lo, request_hi = cron_request_range(job_id)
         query = _rich_select(
             f"s.*, {_CRON_EXECUTION_SQL} AS cron_execution_id", f"s.source = 'cron' AND s.id IN ({_CRON_RUN_IDS_SQL})",

tests/fakes/cron_owner.py

@@ -0,0 +1,43 @@
+"""A real profile gateway authority that admits cron fires without running an agent turn.
+
+Shared by the cron run-history suites (tests/hermes_state, tests/hermes_cli): fires go through
+the real producer (``gateway/session_cron.py``) over SQLite, so run history reads exactly what a
+scheduler fire writes.
+"""
+from types import SimpleNamespace
+
+from cron import executions
+from gateway import run, session_cron
+from gateway.config import GatewayConfig
+from gateway.session import SessionStore
+from gateway.session_authority import SessionAuthority
+from hermes_state import SessionDB
+from hermes_state_runtime import begin_runtime_epoch
+
+
+def cron_owner(home, monkeypatch):
+    """The owner of the profile at *home*, over ``home/state.db``; the caller closes ``authority.db``."""
+    home.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setenv('HERMES_HOME', str(home))
+    (home / 'config.yaml').write_text('model: test-model\n', encoding='utf-8')
+    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {}, 'platform_toolsets': {'cli': []}})
+    db = SessionDB(home / 'state.db')
+    runner = SimpleNamespace(_draining=False, session_store=SessionStore(home / 'sessions', GatewayConfig()),
+                             adapters={})
+    authority = SessionAuthority(runner, profile_id=str(home), instance_id='test', db=db,
+                                 epoch=begin_runtime_epoch(db, instance_id='test'))
+    monkeypatch.setattr(authority, '_schedule', lambda ref: None)
+    runner.session_authority = authority
+    return authority
+
+
+async def admit_fire(authority, job_id, *, running=False):
+    """One scheduler fire: its ledger row (marked running when asked), then the owner's submit and
+    the run's first user message. Returns ``(session_id, execution_id)``."""
+    execution_id = executions.create_execution(job_id, source='test')['id']
+    if running:
+        executions.mark_execution_running(execution_id)
+    receipt = await session_cron.operation(
+        authority, 'submit', {'job_id': job_id, 'request_id': execution_id, 'extra_prompt': None})
+    authority.db.append_message(receipt['session_id'], role='user', content=f'run {job_id}')
+    return receipt['session_id'], execution_id

tests/hermes_cli/test_cron_run_history_sessions.py

@@ -8,7 +8,6 @@ session detail could not say whether the scheduler still owned it. These drive t
 through the real routes in a temp default profile.
 """
 from datetime import timedelta
-from types import SimpleNamespace
 from zoneinfo import ZoneInfo
 
 import pytest
@@ -17,12 +16,8 @@ import hermes_cli.web_routers.cron as cron_routes
 import hermes_cli.web_routers.sessions as session_routes
 import hermes_time
 from cron import executions, jobs
-from gateway import run, session_cron
-from gateway.config import GatewayConfig
-from gateway.session import SessionStore
-from gateway.session_authority import SessionAuthority
-from hermes_state import SessionDB
-from hermes_state_runtime import begin_runtime_epoch, claim_session_input, settle_session_input
+from hermes_state_runtime import claim_session_input, settle_session_input
+from tests.fakes.cron_owner import admit_fire, cron_owner
 
 
 @pytest.fixture()
@@ -33,25 +28,16 @@ def owner(tmp_path, monkeypatch):
 
     home = tmp_path / '.hermes'
     (home / 'cron').mkdir(parents=True)
-    (home / 'config.yaml').write_text('model: test-model\n', encoding='utf-8')
     monkeypatch.setattr(profiles, '_get_default_hermes_home', lambda: home)
     monkeypatch.setattr(profiles, '_get_profiles_root', lambda: home / 'profiles')
-    monkeypatch.setenv('HERMES_HOME', str(home))
-    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {}, 'platform_toolsets': {'cli': []}})
-    db = SessionDB(home / 'state.db')
-    runner = SimpleNamespace(_draining=False, session_store=SessionStore(home / 'sessions', GatewayConfig()),
-                             adapters={})
-    authority = SessionAuthority(runner, profile_id=str(home), instance_id='test', db=db,
-                                 epoch=begin_runtime_epoch(db, instance_id='test'))
-    monkeypatch.setattr(authority, '_schedule', lambda ref: None)
-    runner.session_authority = authority
+    authority = cron_owner(home, monkeypatch)
     # Pin the configured zone so the output writer and the run-history reader resolve the same
     # one, whatever the process's C-level zone (conftest sets TZ=UTC in the env without tzset).
     utc = ZoneInfo('UTC')
     monkeypatch.setattr(hermes_time, 'get_timezone', lambda: utc)
     monkeypatch.setattr(cron_routes, '_get_timezone', lambda: utc)
     yield authority
-    db.close()
+    authority.db.close()
 
 
 @pytest.fixture()
@@ -69,12 +55,7 @@ def clock(monkeypatch):
 
 async def _start_fire(authority, job_id):
     """A scheduler fire as it starts: its running ledger row, then the owner admits and claims it."""
-    execution_id = executions.create_execution(job_id, source='test')['id']
-    executions.mark_execution_running(execution_id)
-    receipt = await session_cron.operation(
-        authority, 'submit', {'job_id': job_id, 'request_id': execution_id, 'extra_prompt': None})
-    session_id = receipt['session_id']
-    authority.db.append_message(session_id, role='user', content=f'report for {job_id}')
+    session_id, execution_id = await admit_fire(authority, job_id, running=True)
     claim = claim_session_input(authority.db, epoch=authority.epoch, session_id=session_id)
     return session_id, execution_id, claim
 
@@ -174,9 +155,6 @@ async def test_a_same_second_output_doc_is_not_pinned_on_the_earlier_execution(
     while an earlier script-only fire that failed within the grace also spans that second. The
     guess must not decide the doc is that neighbour's: one row per execution, and the neighbour's
     failure keeps its own row."""
-    import hermes_time
-    from datetime import timedelta
-
     start = hermes_time.now().replace(microsecond=0) - timedelta(hours=1)
     wall = {'at': start}
     monkeypatch.setattr(executions, '_hermes_now', lambda: wall['at'])
@@ -212,3 +190,33 @@ async def test_a_same_second_output_doc_is_not_pinned_on_the_earlier_execution(
     assert (runs[0]['id'], runs[0]['cron_execution_id']) == (session_id, fire)
     assert runs[1]['title'].startswith('FAILED')
     assert ('script failed' if neighbour_doc_survives else 'script_failed') in runs[1]['preview']
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('overlap_finishes_after_doc', [False, True])
+async def test_an_overlapping_attempt_claimed_before_a_runs_doc_keeps_its_own_row(
+        owner, clock, overlap_finishes_after_doc):
+    """A manual ``hermes cron run`` overlapping a tick: attempt B is claimed while run A is still
+    going, before A writes its doc, and fails without creating a session. B is the doc's newest
+    candidate, but the doc is A's: it folds into A's session, and B keeps its own failed row."""
+    job = jobs.create_job(prompt='report', schedule='every 1h', deliver='local')['id']
+    session_id, fire, claim = await _start_fire(owner, job)
+    clock['seconds'] = 10
+    overlap = executions.create_execution(job, source='test')['id']
+    executions.mark_execution_running(overlap)
+    if not overlap_finishes_after_doc:
+        clock['seconds'] = 12
+        executions.finish_execution(overlap, success=False, error='owner_unavailable')
+    _close_run(owner, session_id, claim)
+    clock['seconds'] = 20
+    jobs.save_job_output(job, '# Cron Job: report\n\nreport done')
+    executions.finish_execution(fire, success=True)
+    if overlap_finishes_after_doc:
+        clock['seconds'] = 25
+        executions.finish_execution(overlap, success=False, error='owner_unavailable')
+
+    runs = _runs(job)
+
+    assert [r['source'] for r in runs] == ['cron_output', 'cron'], runs
+    assert runs[0]['title'].startswith('FAILED') and runs[0]['preview'] == 'owner_unavailable', runs
+    assert (runs[1]['id'], runs[1]['cron_execution_id']) == (session_id, fire)

tests/hermes_state/test_cron_run_sessions.py

@@ -6,43 +6,22 @@ carry ``cron_{job_id}_{timestamp}``. These drive that real producer over SQLite
 no agent turn) and read back through ``SessionDB.list_cron_job_runs``.
 """
 import sqlite3
-from types import SimpleNamespace
 
 import pytest
 
-from cron import executions, jobs
-from gateway import run, session_cron
-from gateway.config import GatewayConfig
-from gateway.session import SessionStore
-from gateway.session_authority import SessionAuthority
+from cron import jobs
+from gateway import session_cron
 from hermes_state import SessionDB
-from hermes_state_runtime import RuntimeStoreError, begin_runtime_epoch
+from hermes_state_runtime import RuntimeStoreError
+from tests.fakes.cron_owner import admit_fire, cron_owner
 
 
 @pytest.fixture()
 def owner(tmp_path, monkeypatch):
     """A real SessionAuthority over ``tmp_path/state.db`` that admits cron fires without running them."""
-    monkeypatch.setenv('HERMES_HOME', str(tmp_path))
-    (tmp_path / 'config.yaml').write_text('model: test-model\n', encoding='utf-8')
-    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {}, 'platform_toolsets': {'cli': []}})
-    db = SessionDB(tmp_path / 'state.db')
-    runner = SimpleNamespace(_draining=False, session_store=SessionStore(tmp_path / 'sessions', GatewayConfig()),
-                             adapters={})
-    authority = SessionAuthority(runner, profile_id=str(tmp_path), instance_id='test', db=db,
-                                 epoch=begin_runtime_epoch(db, instance_id='test'))
-    monkeypatch.setattr(authority, '_schedule', lambda ref: None)
-    runner.session_authority = authority
+    authority = cron_owner(tmp_path, monkeypatch)
     yield authority
-    db.close()
-
-
-async def _fire(authority, job_id):
-    """One scheduler fire through the real producer: its ledger row, then the owner's submit."""
-    execution_id = executions.create_execution(job_id, source='test')['id']
-    receipt = await session_cron.operation(
-        authority, 'submit', {'job_id': job_id, 'request_id': execution_id, 'extra_prompt': None})
-    authority.db.append_message(receipt['session_id'], role='user', content=f'run {job_id}')
-    return receipt['session_id'], execution_id
+    authority.db.close()
 
 
 @pytest.mark.asyncio
@@ -54,8 +33,8 @@ async def test_unified_runs_are_listed_with_legacy_runs_and_only_for_their_job(o
     db.create_session(legacy, source='cron')  # a run from before the unified runtime
     db._conn.execute('UPDATE sessions SET started_at = started_at - 86400 WHERE id = ?', (legacy,))
     db._conn.commit()
-    alpha_run, alpha_fire = await _fire(owner, alpha)
-    beta_run, beta_fire = await _fire(owner, beta)
+    alpha_run, alpha_fire = await admit_fire(owner, alpha)
+    beta_run, beta_fire = await admit_fire(owner, beta)
     assert alpha_run.startswith('local-') and beta_run.startswith('local-')
 
     alpha_runs = db.list_cron_job_runs(alpha)