fix(desktop): gate a cron tab on its run's tip, not an older segment

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.

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<SessionInfo, 'ended_at' | 'last_active' | 'scheduler_owned'> &
-  Partial<Pick<SessionInfo, 'is_active'>>
+  Partial<Pick<SessionInfo, 'cron_run_tip' | 'end_reason' | 'is_active'>>
 
-type CronRunRow = CronRunLiveness & Pick<SessionInfo, 'id'>
+type CronRunRow = CronRunLiveness &
+  Pick<SessionInfo, 'id'> &
+  Partial<Pick<SessionInfo, '_lineage_ids' | '_lineage_root_id'>>
 
 // 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 extends CronRunRow>(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))
+      }
     }
   }
 }

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

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

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)

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:<job>:<execution>``: 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,25 +282,39 @@ 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"])
             if run is None and isinstance(rows[0]["policy_job"], str):
                 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: