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 0ebe6ab07e869..8248d75bde3fb 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 c958a39ad2dad..9012ec670a40d 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 5f002689f9c67..e89655c2aa2bb 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/cron.py b/hermes_cli/web_routers/cron.py index 7827b1e294dd0..30615ccce0b23 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_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py index d6a5cb947a148..5879f2b615d9c 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 1f74f4bff0387..c0dd4121ed44d 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"]) @@ -287,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 bbe1408cd512f..80af6e069a720 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,71 @@ 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 + + +@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 diff --git a/tests/hermes_state/test_cron_run_sessions.py b/tests/hermes_state/test_cron_run_sessions.py index 24bd2e260862b..757402ff7f87c 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