From 668aaa94a3ae9e7b015f15d6f3180a5c772e5e05 Mon Sep 17 00:00:00 2001 From: David Dudok de Wit <5354424+dokterdok@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:50:35 +0200 Subject: [PATCH 1/2] fix(cron): follow a compressed run to its tip in run history and detail A cron run whose context compresses carries on in a continuation session, but only its first segment carries the run's identity (its legacy id or its scheduler admission), so run history and session detail kept describing that segment. Compression stamps its ended_at: a live run read as closed (writable on Desktop), a run the scheduler lost after compressing read as closed instead of a zombie, and detail of the continuation named no run. list_cron_job_runs now projects each row onto its live tip with the sessions list's _project_compression_tips (the tip's id, ended_at, title and transcript fields; the run's started_at and identity kept). cron_run_identity resolves a segment through its compression root, walked up the continuation edges with the forward walk's bound, so a chain deeper than the resume walk's 100 hops still names its run. Refs #134889 Refs #138812 --- hermes_cli/web_routers/cron.py | 14 +++-- hermes_state_portability.py | 38 +++++++++---- .../test_cron_run_history_sessions.py | 56 ++++++++++++++++--- tests/hermes_state/test_cron_run_sessions.py | 45 +++++++++++++++ 4 files changed, 129 insertions(+), 24 deletions(-) diff --git a/hermes_cli/web_routers/cron.py b/hermes_cli/web_routers/cron.py index 7827b1e294dd..30615ccce0b2 100644 --- a/hermes_cli/web_routers/cron.py +++ b/hermes_cli/web_routers/cron.py @@ -461,8 +461,9 @@ def cron_run_scheduler_owned(session: dict[str, Any], profile: Optional[str] = N The session-detail endpoint stamps this so a client re-checking a run it already has open (a restored tab, a send) gets the same answer as the runs list. A unified-runtime run's ``local-*`` id says nothing, so the caller stamps the - row's ``cron_job_id`` / ``cron_execution_id`` (``SessionDB.cron_run_identity``) - first; a legacy ``cron_{job_id}_{timestamp}`` id still names its job. + row's ``cron_job_id`` / ``cron_execution_id`` (``SessionDB.cron_run_identity``, + which also resolves a compression continuation to its run) first; a legacy + ``cron_{job_id}_{timestamp}`` id still names its job. """ if session.get("source") != "cron": return None @@ -481,9 +482,12 @@ def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit: Agent runs are ordinary ``source='cron'`` sessions: a legacy ``cron_{job_id}_{timestamp}`` id, or a unified-runtime ``local-*`` id bound to this job by its scheduler admission. Backed by ``SessionDB.list_cron_job_runs`` - — bounded index range scans, so cost scales with this job's runs, not total - cron history. Script-only (no_agent) jobs never write sessions; their - per-fire output docs (``_list_cron_output_runs``) fill the gaps. The two + — bounded index range scans, so cost scales with this job's runs (and the + chain depth of those that compressed), not total cron history. A run that + compressed is served at its live tip, so its ``ended_at`` and ownership are + the run's, not its first segment's. Script-only (no_agent) jobs never write + sessions; their per-fire output docs (``_list_cron_output_runs``) fill the + gaps. The two representations are reconciled PER EXECUTION, not by branch: a job can switch modes without changing its id (update_job supports no_agent on an existing record), so a surviving historical agent session must not hide diff --git a/hermes_state_portability.py b/hermes_state_portability.py index 1f74f4bff038..4f4ddbd16e07 100644 --- a/hermes_state_portability.py +++ b/hermes_state_portability.py @@ -21,7 +21,9 @@ 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_common import ( + SCHEMA_SQL, _non_continuation_child_sql, _shape_preview, _sql_preview_raw, _sql_session_last_active) +from hermes_state_compression import _CHAIN_CAP from hermes_state_messages import _parse_tool_calls, _tool_calls_count import itertools @@ -115,6 +117,15 @@ def _rich_select(select_cols: str, where: str, tail: str = "", prompt_select: Op _CRON_EXECUTION_SQL = ( "(SELECT substr(request_id, length(:request_lo) + 1) FROM session_admissions" f" WHERE target_session_id = s.id AND {_CRON_ADMISSION_SQL} LIMIT 1)") +# The first segment of the run a compression continuation belongs to: up the continuation edges +# (fork markers bound to the edge's parent, as prune and delete walk them), bounded like the +# forward walk so every segment the runs list can reach resolves back. +_COMPRESSION_ROOT_SQL = ( + "WITH RECURSIVE up(id, depth) AS (SELECT ?, 0 UNION ALL SELECT p.id, up.depth + 1 FROM up" + " JOIN sessions c ON c.id = up.id JOIN sessions p ON p.id = c.parent_session_id" + f" WHERE p.end_reason = 'compression'\n{_non_continuation_child_sql('c.', 'p.id')}" + f" AND up.depth < {_CHAIN_CAP})" + " SELECT id FROM up ORDER BY depth DESC LIMIT 1") def _export_timings(messages: list[dict[str, Any]], session_id: Optional[str] = None) -> dict[str, Any]: @@ -254,11 +265,14 @@ def list_cron_job_runs(self, job_id: str, limit: int = 20, offset: int = 0) -> l must not change (exact-retry recovery re-derives it), but the scheduler principal admitted each under ``cron::``: a range of ``idx_session_admissions_cron_request``. Both scans scale with this job's runs, never with total cron history or with every local - session's policy receipt, and skip ``list_sessions_rich``'s compression-chain CTE. - - 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.""" + session's policy receipt, and skip ``list_sessions_rich``'s CTE over every seed's chain. + + Only a run's first segment carries that id or admission, yet a run whose context compressed + goes on in a continuation, which the scheduler finalizes. Each row is projected onto its + live tip as a sessions-list row is (``_project_compression_tips``: the tip's id, + ``ended_at``, title and transcript fields; the run's ``started_at`` and identity kept; + ``_lineage_ids`` naming every segment), so liveness and navigation follow the run. Only + the rows of this page that compressed walk their chain.""" prefix, prefix_hi = prefix_range(f"cron_{job_id}_") request_lo, request_hi = cron_request_range(job_id) query = _rich_select( @@ -268,18 +282,22 @@ def list_cron_job_runs(self, job_id: str, limit: int = 20, offset: int = 0) -> l ) rows = self._read_rows(query, {"prefix": prefix, "prefix_hi": prefix_hi, "request_lo": request_lo, "request_hi": request_hi, "job_id": job_id, "limit": limit, "offset": offset}) - return [{**self._rich_row(row), "cron_job_id": job_id} for row in rows] + runs = [{**self._rich_row(row), "cron_job_id": job_id} for row in rows] + return self._project_compression_tips(runs, compact_rows=False) def cron_run_identity(self, session_id: str) -> Optional[dict[str, Any]]: """``{"cron_job_id", "cron_execution_id"}`` of a cron run session, else None: a legacy id names its job (no execution), a unified-runtime run is one probe of its scheduler admission - (and, when its request id is ambiguous, of the job its creation receipt's policy names).""" - legacy_job = legacy_cron_run_job(session_id) + (and, when its request id is ambiguous, of the job its creation receipt's policy names). + A compression continuation is the same run, so a segment resolves through its compression + root (``_COMPRESSION_ROOT_SQL``), never across a branch, delegate or reset edge.""" + run_session_id = self._read_rows(_COMPRESSION_ROOT_SQL, (session_id,))[0]["id"] + legacy_job = legacy_cron_run_job(run_session_id) if legacy_job is not None: return {"cron_job_id": legacy_job, "cron_execution_id": None} rows = self._read_rows( "SELECT request_id, " + _CRON_POLICY_JOB_SQL + " AS policy_job FROM session_admissions" - " WHERE principal_id = ? AND target_session_id = ? ORDER BY seq LIMIT 1", (CRON_PRINCIPAL_ID, session_id)) + " WHERE principal_id = ? AND target_session_id = ? ORDER BY seq LIMIT 1", (CRON_PRINCIPAL_ID, run_session_id)) run = None if rows: run = cron_run_from_request(rows[0]["request_id"]) diff --git a/tests/hermes_cli/test_cron_run_history_sessions.py b/tests/hermes_cli/test_cron_run_history_sessions.py index bbe1408cd512..8398daa0bff6 100644 --- a/tests/hermes_cli/test_cron_run_history_sessions.py +++ b/tests/hermes_cli/test_cron_run_history_sessions.py @@ -15,9 +15,11 @@ 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 cron import executions, jobs, scheduler +from hermes_state import SessionDB from hermes_state_runtime import claim_session_input, settle_session_input from tests.fakes.cron_owner import admit_fire, cron_owner +from tests.hermes_state.test_cron_run_sessions import compress @pytest.fixture() @@ -60,13 +62,14 @@ async def _start_fire(authority, job_id): return session_id, execution_id, claim -def _close_run(authority, session_id, claim): - """The owner's run finishes: the answer is written, the admission settled, the session ended.""" - authority.db.append_message(session_id, role='assistant', content='report done') +def _close_run(authority, job_id, session_id, claim): + """The owner's run finishes: the answer is written where the run now is, the admission settled, + and the scheduler's own finalizer (its store handle) titles and ends the run's last segment.""" + authority.db.append_message(authority.db.get_compression_tip(session_id), role='assistant', content='report done') settle_session_input(authority.db, epoch=authority.epoch, admission_id=claim['admission_id'], generation=claim['generation'], outcome='completed', result={'result': {'cron_result': [True, 'doc', 'report done', None]}, 'usage': {}}) - authority.db.end_session(session_id, 'cron_complete') + scheduler._finalize_cron_session(SessionDB(authority.db.db_path), None, job_id, 'report', session_id) def _runs(job_id): @@ -93,7 +96,7 @@ async def test_unified_run_is_one_owned_session_row_and_detail_agrees(owner, clo assert (detail.get('cron_job_id'), detail.get('cron_execution_id')) == (job, execution_id) # Done: the agent fire's output doc is the same execution as its session, not a second row. - _close_run(owner, session_id, claim) + _close_run(owner, job, session_id, claim) clock['seconds'] = 5 jobs.save_job_output(job, '# Cron Job: report\n\nreport done') executions.finish_execution(execution_id, success=True) @@ -124,7 +127,7 @@ async def test_a_fire_that_failed_beside_a_unified_run_keeps_its_own_row(owner, it created a session, is its own run, not swallowed by its neighbour's time window.""" job = jobs.create_job(prompt='report', schedule='every 1m', deliver='local')['id'] session_id, first, claim = await _start_fire(owner, job) - _close_run(owner, session_id, claim) + _close_run(owner, job, session_id, claim) clock['seconds'] = 20 jobs.save_job_output(job, '# Cron Job: report\n\nreport done') executions.finish_execution(first, success=True) @@ -176,7 +179,7 @@ def at(seconds): at(60.5) # the job switched to agent mode; its next fire is claimed mid-second session_id, fire, claim = await _start_fire(owner, job) - _close_run(owner, session_id, claim) + _close_run(owner, job, session_id, claim) at(60.75) jobs.save_job_output(job, '# Cron Job: report\n\nreport done') # filed as second 60 executions.finish_execution(fire, success=True) @@ -207,7 +210,7 @@ async def test_an_overlapping_attempt_claimed_before_a_runs_doc_keeps_its_own_ro if not overlap_finishes_after_doc: clock['seconds'] = 12 executions.finish_execution(overlap, success=False, error='owner_unavailable') - _close_run(owner, session_id, claim) + _close_run(owner, job, session_id, claim) clock['seconds'] = 20 jobs.save_job_output(job, '# Cron Job: report\n\nreport done') executions.finish_execution(fire, success=True) @@ -220,3 +223,38 @@ async def test_an_overlapping_attempt_claimed_before_a_runs_doc_keeps_its_own_ro 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) + + +@pytest.mark.asyncio +async def test_a_compressed_run_is_served_at_its_tip_whether_live_closed_or_lost(owner, clock): + """Compression stamps ``ended_at`` on a run's first segment while the run goes on in its + continuation. Read as the run's own, that stamp made a live run look closed (so writable) and a + run the scheduler lost look closed rather than a zombie, and detail of the continuation could + not name its run at all.""" + job = jobs.create_job(prompt='report', schedule='every 1h', deliver='local')['id'] + lost_job = jobs.create_job(prompt='lost report', schedule='every 1h', deliver='local')['id'] + session_id, execution_id, claim = await _start_fire(owner, job) + lost_id, lost_execution, _ = await _start_fire(owner, lost_job) + tip, lost_tip = compress(owner.db, session_id), compress(owner.db, lost_id) + + [running] = _runs(job) + assert (running['id'], running['ended_at'], running['scheduler_owned']) == (tip, None, True) + detail = await _detail(tip) + assert (detail.get('cron_job_id'), detail.get('cron_execution_id')) == (job, execution_id) + assert detail.get('scheduler_owned') is True + + # The scheduler loses one run after compression (its attempt fails, the run is never closed). + executions.finish_execution(lost_execution, success=False, error='watchdog: inactivity timeout') + [zombie] = _runs(lost_job) + assert (zombie['id'], zombie['ended_at'], zombie['scheduler_owned']) == (lost_tip, None, False) + assert (await _detail(lost_tip)).get('scheduler_owned') is False + + _close_run(owner, job, session_id, claim) + clock['seconds'] = 5 + jobs.save_job_output(job, '# Cron Job: report\n\nreport done') + executions.finish_execution(execution_id, success=True) + + [finished] = _runs(job) + assert (finished['id'], finished['end_reason'], finished['scheduler_owned']) == (tip, 'cron_complete', False) + assert finished['title'] and finished['title'] == owner.db.get_session(tip)['title'] + assert (await _detail(tip))['scheduler_owned'] is False diff --git a/tests/hermes_state/test_cron_run_sessions.py b/tests/hermes_state/test_cron_run_sessions.py index 24bd2e260862..757402ff7f87 100644 --- a/tests/hermes_state/test_cron_run_sessions.py +++ b/tests/hermes_state/test_cron_run_sessions.py @@ -12,6 +12,7 @@ from cron import jobs from gateway import session_cron from hermes_state import SessionDB +from hermes_state_ids import new_session_id from hermes_state_runtime import RuntimeStoreError from tests.fakes.cron_owner import admit_fire, cron_owner @@ -24,6 +25,19 @@ def owner(tmp_path, monkeypatch): authority.db.close() +def compress(db, session_id): + """One context-compression rotation as the agent publishes it (``agent/conversation_compression.py``): + under the segment's lease the segment closes ``compression`` and a continuation carries the run on.""" + holder = f'compressor:{session_id}' + assert db.try_acquire_compression_lock(session_id, holder) + child = new_session_id() + db.publish_compression_child( + parent_session_id=session_id, child_session_id=child, source=db.get_session(session_id)['source'], + messages=[{'role': 'user', 'content': f'summary of {session_id}'}], compression_lock_holder=holder) + db.release_compression_lock(session_id, holder) + return child + + @pytest.mark.asyncio async def test_unified_runs_are_listed_with_legacy_runs_and_only_for_their_job(owner): db = owner.db @@ -49,6 +63,37 @@ async def test_unified_runs_are_listed_with_legacy_runs_and_only_for_their_job(o assert [r['id'] for r in db.list_cron_job_runs(alpha, limit=1, offset=1)] == [legacy] +@pytest.mark.asyncio +async def test_a_compressed_run_is_one_row_at_its_tip_and_each_segment_names_the_run(owner): + """Only a run's first segment carries its admission (or legacy id); a run that compressed goes on + in continuations. History lists it once, at its live tip, keeping the run's start and identity, + and every segment resolves to that run. A /branch copy is the user's own conversation.""" + db = owner.db + job = jobs.create_job(prompt='report', schedule='every 1h', deliver='local')['id'] + old_job = jobs.create_job(prompt='old report', schedule='every 1h', deliver='local')['id'] + run_id, fire = await admit_fire(owner, job) + middle = compress(db, run_id) + tip = compress(db, middle) + branch = new_session_id() + db.create_session(branch, source='cron', parent_session_id=run_id, model_config={'_branched_from': run_id}) + legacy = f'cron_{old_job}_20261011_120000' + db.create_session(legacy, source='cron') + legacy_tip = legacy + for _ in range(101): # deeper than the resume walk's 100-hop bound; the list reaches its tip + legacy_tip = compress(db, legacy_tip) + + [run] = db.list_cron_job_runs(job) + [legacy_run] = db.list_cron_job_runs(old_job) + + assert (run['id'], run['ended_at'], run['started_at']) == (tip, None, db.get_session(run_id)['started_at']) + assert (run['cron_job_id'], run['cron_execution_id'], run['_lineage_ids']) == (job, fire, [run_id, middle, tip]) + assert (legacy_run['id'], legacy_run['cron_job_id'], legacy_run['cron_execution_id']) == (legacy_tip, old_job, None) + for segment in (run_id, middle, tip): + assert db.cron_run_identity(segment) == {'cron_job_id': job, 'cron_execution_id': fire} + assert db.cron_run_identity(legacy_tip) == {'cron_job_id': old_job, 'cron_execution_id': None} + assert db.cron_run_identity(branch) is None + + @pytest.mark.asyncio async def test_a_fresh_execution_id_with_a_colon_is_refused_before_it_can_misfile_a_run(owner): """``cron::`` splits at its last ``:``, so a fire ``x:y`` of job ``a`` would read From ee2f628dd9227b3236f28c31758025a2ab0260d5 Mon Sep 17 00:00:00 2001 From: David Dudok de Wit <5354424+dokterdok@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:31:56 +0200 Subject: [PATCH 2/2] fix(desktop): gate a cron tab on its run's tip, not an older segment A cron run that compresses goes on in a continuation, and compression stamps ended_at on the segment it ends. A Desktop tab opened on the run while that segment was the tip keeps holding it, and the write gate read the segment's own stamp as the run closing: a run the scheduler lost after compressing stayed writable instead of view-only. Polls did not correct it, because run rows now arrive under the new tip's id and verdicts were matched by exact id. Session detail of a cron segment ended by compression now carries the run at its live tip (cron_run_tip: its id, ended_at, end_reason and scheduler ownership), read through the runs list's own projection. The Desktop gate judges that instead of the segment's stamp, and a poll reaches every verdict of the run through the row's lineage. A finalized run stays writable; a backend without the field reads as before. Refs #134889 Refs #138812 --- .../src/app/cron/open-cron-run.test.ts | 47 +++++++++++++++++++ apps/desktop/src/app/cron/open-cron-run.ts | 37 ++++++++++++--- apps/desktop/src/types/hermes.ts | 18 +++++++ hermes_cli/web_routers/sessions.py | 11 ++++- hermes_state_portability.py | 10 ++++ .../test_cron_run_history_sessions.py | 33 +++++++++++++ 6 files changed, 148 insertions(+), 8 deletions(-) diff --git a/apps/desktop/src/app/cron/open-cron-run.test.ts b/apps/desktop/src/app/cron/open-cron-run.test.ts index 0ebe6ab07e86..8248d75bde3f 100644 --- a/apps/desktop/src/app/cron/open-cron-run.test.ts +++ b/apps/desktop/src/app/cron/open-cron-run.test.ts @@ -164,6 +164,53 @@ describe('refreshCronRunWriteGate — restored tab after a restart (no verdict y }) }) +describe('a tab held on a segment that compression ended after it was opened', () => { + // The tab opened the run at its tip; the run then compressed and goes on in a continuation. + // The held segment's own `ended_at` is the compression's, not a close, and run rows now arrive + // under the new tip's id, naming every segment in `_lineage_ids`. + const held = 'local-0a1b2c3d4e5f6a7b' + const tip = 'local-9f8e7d6c5b4a3f2e' + + // Session detail of the held segment: the backend reports its run at the live tip. + const heldDetail = (runTip?: { ended_at: null | number; scheduler_owned: boolean }) => ({ + ...(runTip && { + cron_run_tip: { end_reason: runTip.ended_at == null ? null : 'cron_complete', id: tip, ...runTip } + }), + end_reason: 'compression', + ended_at: NOW_S - 600, + last_active: NOW_S - 600, + scheduler_owned: false, + source: 'cron' + }) + + it.each([ + { readOnly: true, runTip: { ended_at: null, scheduler_owned: false }, state: 'lost after compressing' }, + { readOnly: false, runTip: { ended_at: null, scheduler_owned: true }, state: 'live' }, + { readOnly: false, runTip: { ended_at: NOW_S, scheduler_owned: false }, state: 'finalized' }, + // A backend without `cron_run_tip` reads the segment's own stamp, as before. + { readOnly: false, runTip: undefined, state: 'unknown to an older backend' } + ])('the pre-send gate judges the run at its tip: $state', async ({ readOnly, runTip }) => { + openCronRun(run({ id: held, scheduler_owned: true }), vi.fn()) + const fetchRow = vi.fn(async () => heldDetail(runTip)) + + expect(await refreshCronRunWriteGate(held, fetchRow)).toBe(readOnly) + expect(fetchRow).toHaveBeenCalledWith(held) + }) + + it('a poll reaches the held segment through its run row lineage, and no other run', () => { + openCronRun(run({ id: held, scheduler_owned: true }), vi.fn()) + + const lostRun = { ...run({ id: tip, scheduler_owned: false }), _lineage_ids: [held, tip] } + const otherLostRun = { ...run({ id: 'local-other-tip', scheduler_owned: false }), _lineage_ids: ['local-other'] } + + reconcileCronRunVerdicts([otherLostRun]) + expect(isStoredTranscriptReadOnly(held)).toBe(false) + + reconcileCronRunVerdicts([lostRun]) + expect(isStoredTranscriptReadOnly(held)).toBe(true) + }) +}) + describe('refreshCronRunWriteGate — a unified-runtime run is known by its row, not its id (#138812)', () => { // The unified gateway runtime keeps the run's runtime-minted `local-*` id; // only the row the backend served (`source: 'cron'`) says it is a cron run. diff --git a/apps/desktop/src/app/cron/open-cron-run.ts b/apps/desktop/src/app/cron/open-cron-run.ts index c958a39ad2da..9012ec670a40 100644 --- a/apps/desktop/src/app/cron/open-cron-run.ts +++ b/apps/desktop/src/app/cron/open-cron-run.ts @@ -1,11 +1,18 @@ -import { hasCronRunVerdict, isCronRunReadOnly, recordCronRunVerdict } from '@/store/read-only-transcript' +import { + $cronRunReadOnlyVerdicts, + hasCronRunVerdict, + isCronRunReadOnly, + recordCronRunVerdict +} from '@/store/read-only-transcript' import { ownerLookupSessionRows, sessionMatchesStoredId } from '@/store/session' import type { SessionInfo } from '@/types/hermes' type CronRunLiveness = Pick & - Partial> + Partial> -type CronRunRow = CronRunLiveness & Pick +type CronRunRow = CronRunLiveness & + Pick & + Partial> // Mirrors the backend's activity window (`hermes_cli/web_routers/cron.py`, // `sessions.py::_ACTIVE_WINDOW_S`). Used ONLY against an older backend that @@ -34,8 +41,20 @@ const LEGACY_CRON_RUN_SESSION_ID = /^cron_.+_\d{8}_\d{6}$/ * (the backend's `is_active`, or the same formula over `last_active` for rows * that carry no flag). That is the best such a backend can say, and the verdict * is re-evaluated on every refresh and send, so it never latches. + * + * Compression ending a segment is not a close: the run goes on in a + * continuation. Run rows are served at the run's live tip, and session detail + * of an older segment (a tab opened it when it was the tip) carries the run at + * that tip (`cron_run_tip`), which is judged instead. A backend without it + * leaves the segment's own stamp to read as a close, as it always has. */ export function isResumableCronRun(run: CronRunLiveness, nowMs = Date.now()): boolean { + const tip = run.end_reason === 'compression' ? run.cron_run_tip : undefined + + if (tip) { + return tip.ended_at != null || tip.scheduler_owned === true + } + if (run.ended_at != null) { return true } @@ -112,12 +131,18 @@ export function openCronRun(run: Run, open: (sessionId: * Fold a fresh page of run rows (a Cron surface poll) into the verdicts of runs * the desktop already evaluated, so a zombie-looking run that later ticks or * closes becomes writable again — and a live one that dies becomes view-only — - * without waiting for a send. Rows never opened are left alone. + * without waiting for a send. A run that compressed arrives under its new + * tip's id, so its verdicts are found through its lineage: a tab still holding + * an older segment follows the run. Rows never opened are left alone. */ export function reconcileCronRunVerdicts(runs: readonly CronRunRow[]): void { + const opened = [...$cronRunReadOnlyVerdicts.get().keys()] + for (const run of runs) { - if (hasCronRunVerdict(run.id)) { - recordRunVerdict(run) + for (const id of opened) { + if (sessionMatchesStoredId(run, id)) { + recordCronRunVerdict(id, !isResumableCronRun(run)) + } } } } diff --git a/apps/desktop/src/types/hermes.ts b/apps/desktop/src/types/hermes.ts index 5f002689f9c6..e89655c2aa2b 100644 --- a/apps/desktop/src/types/hermes.ts +++ b/apps/desktop/src/types/hermes.ts @@ -554,6 +554,14 @@ export interface SessionCreateResponse { stored_session_id?: string } +/** A cron run at its live tip (see {@link SessionInfo.cron_run_tip}). */ +export interface CronRunTip { + end_reason: null | string + ended_at: null | number + id: string + scheduler_owned: boolean | null +} + export interface SessionInfo { archived?: boolean cwd?: null | string @@ -598,6 +606,16 @@ export interface SessionInfo { * long tool call and is false for a zombie whose process died (#88443). * Undefined against older backends and for non-cron rows. */ scheduler_owned?: boolean + /** Why the session ended (`end_session`'s reason); null while open. + * `'compression'` ends a segment whose conversation goes on in a + * continuation, so it is not a close. */ + end_reason?: null | string + /** Session detail of a cron run segment that compression ended: the run it + * belongs to at its live tip, as the runs list serves it. A tab can hold + * such a segment (it was the tip when opened), so the cron write gate + * judges this instead of the segment's own `ended_at`. Undefined for any + * other row and against older backends. */ + cron_run_tip?: CronRunTip last_active: number message_count: number model: null | string diff --git a/hermes_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py index d6a5cb947a14..5879f2b615d9 100644 --- a/hermes_cli/web_routers/sessions.py +++ b/hermes_cli/web_routers/sessions.py @@ -573,11 +573,18 @@ def _detail(db): # unified-runtime run's local id names no job, so stamp its identity first. from hermes_cli.web_routers.cron import cron_run_scheduler_owned - if session.get("source") == "cron": - session.update(db.cron_run_identity(sid) or {}) + identity = (db.cron_run_identity(sid) or {}) if session.get("source") == "cron" else {} + session.update(identity) owned = cron_run_scheduler_owned(session, profile) if owned is not None: session["scheduler_owned"] = owned + # Compression ending this segment did not end its run. A client still holding the + # segment (it was the tip when opened) gates writes on the run at its live tip. + run = db.cron_run_tip(sid) if session.get("end_reason") == "compression" else None + if run: + session["cron_run_tip"] = { + "id": run["id"], "ended_at": run["ended_at"], "end_reason": run["end_reason"], + "scheduler_owned": cron_run_scheduler_owned({**run, **identity}, profile)} return session return await asyncio.to_thread(_with_db, profile, _detail, read_only=True) diff --git a/hermes_state_portability.py b/hermes_state_portability.py index 4f4ddbd16e07..c0dd4121ed44 100644 --- a/hermes_state_portability.py +++ b/hermes_state_portability.py @@ -305,6 +305,16 @@ def cron_run_identity(self, session_id: str) -> Optional[dict[str, Any]]: run = cron_run_from_request(rows[0]["request_id"], rows[0]["policy_job"]) return {"cron_job_id": run[0], "cron_execution_id": run[1]} if run else None + def cron_run_tip(self, session_id: str) -> Optional[dict[str, Any]]: + """The run *session_id* is a segment of, as ``list_cron_job_runs`` serves it: its first + segment projected onto its live tip. None when that tip is *session_id* itself. Compression + ending a segment is not its run closing, so a reader still holding an older segment (a + client opened it when it was the tip) judges the run by this row instead.""" + root = self._read_rows("SELECT id, source, started_at, ended_at, end_reason, title FROM sessions" + f" WHERE id = ({_COMPRESSION_ROOT_SQL})", (session_id,)) + [run] = self._project_compression_tips([dict(row) for row in root], compact_rows=True) + return run if run["id"] != session_id else None + def _get_session_rich_row(self, session_id: str, compact_rows: bool = False) -> Optional[dict[str, Any]]: """One session with the ``list_sessions_rich`` enriched columns, or None. ``compact_rows=True`` omits the ``system_prompt`` blob. Public alias: diff --git a/tests/hermes_cli/test_cron_run_history_sessions.py b/tests/hermes_cli/test_cron_run_history_sessions.py index 8398daa0bff6..80af6e069a72 100644 --- a/tests/hermes_cli/test_cron_run_history_sessions.py +++ b/tests/hermes_cli/test_cron_run_history_sessions.py @@ -258,3 +258,36 @@ async def test_a_compressed_run_is_served_at_its_tip_whether_live_closed_or_lost assert (finished['id'], finished['end_reason'], finished['scheduler_owned']) == (tip, 'cron_complete', False) assert finished['title'] and finished['title'] == owner.db.get_session(tip)['title'] assert (await _detail(tip))['scheduler_owned'] is False + + +@pytest.mark.asyncio +async def test_detail_of_an_older_segment_reports_its_run_at_the_tip(owner, clock): + """A Desktop tab can hold a segment that compression ended after the tab opened it as the tip. + That segment's own ``ended_at`` is the compression's, not the run closing, so its detail reports + the run at its live tip, as the runs list serves it: the send gate reads that to keep a lost + run view-only and a live or finalized one writable.""" + job = jobs.create_job(prompt='report', schedule='every 1h', deliver='local')['id'] + lost_job = jobs.create_job(prompt='lost report', schedule='every 1h', deliver='local')['id'] + session_id, execution_id, claim = await _start_fire(owner, job) + lost_id, lost_execution, _ = await _start_fire(owner, lost_job) + middle = compress(owner.db, session_id) + tip = compress(owner.db, middle) + lost_tip = compress(owner.db, compress(owner.db, lost_id)) + + for segment in (session_id, middle): + assert (await _detail(segment)).get('cron_run_tip') == { + 'id': tip, 'ended_at': None, 'end_reason': None, 'scheduler_owned': True} + assert 'cron_run_tip' not in await _detail(tip) + + executions.finish_execution(lost_execution, success=False, error='watchdog: inactivity timeout') + [zombie] = _runs(lost_job) + assert (await _detail(lost_id)).get('cron_run_tip') == { + 'id': lost_tip, 'ended_at': None, 'end_reason': None, 'scheduler_owned': zombie['scheduler_owned']} + assert zombie['scheduler_owned'] is False + + _close_run(owner, job, session_id, claim) + clock['seconds'] = 5 + executions.finish_execution(execution_id, success=True) + closed = (await _detail(middle))['cron_run_tip'] + assert (closed['id'], closed['end_reason'], closed['scheduler_owned']) == (tip, 'cron_complete', False) + assert closed['ended_at'] == owner.db.get_session(tip)['ended_at'] is not None