fix(cron): mark running cron sessions as active in sidebar (#6734)

* fix(cron): mark running cron sessions as active in sidebar

* fix(cron): include cron_running in effective-streaming predicate + test updates (#6734)

* fix(cron): longest-prefix-first cron job matching in session list cache (#6728)

* fix(cron): require exact run-timestamp suffix in cron_running prefix match

Gate re-review (Codex) found the longest-prefix-first match could misattribute:
cron_job_prefixes is built ONLY from RUNNING jobs, so when the true longer owner
(backup_full) is NOT running, a running shorter job (backup) swallowed a completed
cron_backup_full_YYYYMMDD_HHMMSS session (its prefix cron_backup_ still matched),
deferring that row's completion/unread transition. Now the text after a matched
cron_{jid}_ prefix must be EXACTLY a run timestamp (\d{8}_\d{6}), so a shorter job
id can't claim a longer job id's session regardless of which jobs are running.
Regression test added (fails without the guard).

Co-authored-by: webtecnica <webtecnica@users.noreply.github.com>

---------

Co-authored-by: webtecnica <webtecnica@users.noreply.github.com>
Co-authored-by: nesquena-hermes <nesquena-hermes@users.noreply.github.com>

api/route_session_list_cache.py

@@ -2,6 +2,7 @@
 
 import os
 import copy
+import re
 import threading
 import time
 from collections import OrderedDict
@@ -12,6 +13,14 @@ from api.models import _active_state_db_path, _active_stream_ids
 from api.profiles import _profiles_match
 
 
+# Cron session ids are ``cron_{job_id}_{run_timestamp}`` where the run
+# timestamp is ``YYYYMMDD_HHMMSS`` (e.g. cron_job6728_20260803_100000). Used to
+# validate that the text after a matched ``cron_{jid}_`` prefix is EXACTLY a run
+# timestamp, so a shorter job id (backup) cannot claim a longer job id's session
+# (backup_full) when the longer job is not itself running (#6728 gate fix).
+_CRON_RUN_TS_RE = re.compile(r"\d{8}_\d{6}")
+
+
 _SESSIONS_CACHE_TTL_SECONDS = 2.5
 # #4808: while a turn is actively streaming the frontend polls /api/sessions on a
 # fixed cadence (static/sessions.js `_streamingPollMs`). With the idle TTL of
@@ -98,6 +107,27 @@ def _session_list_cache_active_stream_ids():
     return _active_stream_ids()
 
 
+def _session_list_cache_running_cron_jobs() -> dict[str, float]:
+    """Return {job_id: start_epoch} for cron jobs currently tracked as running.
+
+    Cron liveness lives only in the in-memory ``_RUNNING_CRON_JOBS`` dict in
+    api.routes (#6728): the sidebar polls /api/sessions (not /api/crons/status),
+    so without this overlay a still-running cron job's session row looks
+    completed the moment it appends a message. Fail closed to an empty dict.
+    """
+    try:
+        import api.routes as _routes
+
+        jobs = getattr(_routes, "_RUNNING_CRON_JOBS", None)
+        lock = getattr(_routes, "_RUNNING_CRON_LOCK", None)
+        if jobs is None or lock is None:
+            return {}
+        with lock:
+            return dict(jobs)
+    except Exception:
+        return {}
+
+
 def _session_list_cache_resolved_source_stamp(key: tuple):
     try:
         import api.routes as _routes
@@ -426,6 +456,11 @@ def _session_list_cache_overlay_runtime_rows(rows: list[dict]) -> list[dict]:
         active_stream_ids = _session_list_cache_active_stream_ids()
     except Exception:
         active_stream_ids = set()
+    try:
+        running_cron_jobs = _session_list_cache_running_cron_jobs()
+    except Exception:
+        running_cron_jobs = {}
+    cron_job_prefixes = [(jid, f"cron_{jid}_", started_at) for jid, started_at in running_cron_jobs.items()]
     session_ids = [
         str(row.get("session_id") or "").strip()
         for row in rows
@@ -457,11 +492,50 @@ def _session_list_cache_overlay_runtime_rows(rows: list[dict]) -> list[dict]:
                     item[key] = raw_live_value
         stream_id = item.get("active_stream_id")
         item["is_streaming"] = bool(stream_id and stream_id in active_stream_ids)
+        # #6728: a still-running cron job's session row must not look completed
+        # in the sidebar. Cron liveness is only exposed via /api/crons/status,
+        # which the sidebar never polls — stamp the flag here so the client can
+        # defer its completion/unread transition until the job actually ends.
+        # Session ids are cron_{job_id}_{run_timestamp}: only the run started at
+        # (or after) the tracked start belongs to the live execution — older runs
+        # of the same job stay completed.
+        item["cron_running"] = _session_list_row_cron_running(
+            sid, item, cron_job_prefixes
+        )
         overlaid.append(item)
     overlaid.sort(key=_session_list_runtime_sort_key, reverse=True)
     return overlaid
 
 
+def _session_list_row_cron_running(
+    sid: str, row: dict, cron_job_prefixes: list[tuple[str, str, float]]
+) -> bool:
+    if not cron_job_prefixes or not sid:
+        return False
+    created_at = _session_list_row_numeric_value(row.get("created_at"))
+    # Longest prefix first, no fall-through: job ids may nest (backup vs
+    # backup_full), and the shorter prefix is a valid prefix of the longer one.
+    # A session belongs to the longest matching job id — first-match in
+    # insertion order, or falling through to a shorter prefix after a time-miss,
+    # would let a running shorter-prefix job claim a completed longer-prefix
+    # session. Mirrors the max(matches, key=len) convention in
+    # api.routes._latest_cron_session_info_for_jobs.
+    #
+    # #6728 (gate fix): prefixes are built ONLY from RUNNING jobs, so if the
+    # true longer owner (backup_full) is not running, longest-prefix sorting
+    # never sees it and a running `backup` would otherwise swallow a
+    # `cron_backup_full_YYYYMMDD_HHMMSS` session (the leftover `full_...` still
+    # starts with nothing it should match). Require the text AFTER the prefix to
+    # be exactly a run-timestamp (YYYYMMDD_HHMMSS) so a shorter job id cannot
+    # claim a longer job id's session regardless of which jobs are running.
+    for _jid, prefix, started_at in sorted(
+        cron_job_prefixes, key=lambda item: len(item[1]), reverse=True
+    ):
+        if sid.startswith(prefix) and _CRON_RUN_TS_RE.fullmatch(sid[len(prefix):]):
+            return created_at >= started_at
+    return False
+
+
 def _session_list_row_numeric_value(value) -> float:
     try:
         numeric = float(value or 0)

api/routes.py

@@ -10026,6 +10026,7 @@ _SIDEBAR_SESSION_RESPONSE_FIELDS = {
     "is_cli_session",
     "is_messaging_session",
     "is_streaming",
+    "cron_running",
     "active_stream_id",
     "has_pending_user_message",
     "pending_started_at",

static/sessions.js

@@ -814,6 +814,7 @@ function _isSessionLocallyStreaming(s) {
 function _isSessionEffectivelyStreaming(s) {
   return Boolean(s && (
     s.is_streaming ||
+    s.cron_running ||
     _hasPendingUserMessageSignal(s) ||
     _isSessionLocallyStreaming(s)
   ));
@@ -1214,8 +1215,12 @@ function _markPollingCompletionUnreadTransitions(sessions) {
     const lastMessageAt = Number(s.last_message_at || 0);
     const hasServerRunSignal=Boolean(s.is_streaming||_hasPendingUserMessageSignal(s));
     const canMarkCompletedStream=Boolean(hasServerRunSignal||previousSnapshot||observedStreaming);
-    const completedObservedStream = canMarkCompletedStream&&wasStreaming === true && !isStreaming;
+    // #6728: cron liveness is server-side (only /api/crons/status exposes it);
-    const completedWithNewMessages = Boolean(
+    // the sidebar must defer its completion/unread transition while the job is
+    // still running, or a mid-run message makes the row look completed.
+    const cronRunning = Boolean(s.cron_running);
+    const completedObservedStream = !cronRunning && canMarkCompletedStream && wasStreaming === true && !isStreaming;
+    const completedWithNewMessages = !cronRunning && Boolean(
       (previousSnapshot || observedStreaming)
       && !isStreaming
       && (
@@ -1223,7 +1228,7 @@ function _markPollingCompletionUnreadTransitions(sessions) {
         || lastMessageAt > Number((previousSnapshot || observedStreaming).last_message_at || 0)
       )
     );
-    const completedPersistedObservedStream = Boolean(observedStreaming && !isStreaming);
+    const completedPersistedObservedStream = !cronRunning && Boolean(observedStreaming && !isStreaming);
     if (completedObservedStream || completedPersistedObservedStream || completedWithNewMessages) {
       if (!_isSessionActivelyViewedForList(sid)) {
         // Tag cron session-list markers with source+profile so profile-switch

tests/test_6728_cron_running_sidebar.py

@@ -0,0 +1,382 @@
+"""Regression tests for #6728: active cron sessions wrongly appear as completed.
+
+The sidebar polls /api/sessions (never /api/crons/status), so cron liveness
+must be stamped onto the session-list rows. A still-running cron job's row
+must carry ``cron_running=True`` so the client defers its completion/unread
+transition; a finished (or non-cron) row must not. Because cron session ids
+are ``cron_{job_id}_{run_timestamp}``, only the run whose ``created_at`` is at
+or after the tracked start belongs to the live execution — older runs of the
+same job must stay completed.
+"""
+
+import json
+import pathlib
+import shutil
+import subprocess
+import tempfile
+
+import pytest
+
+ROOT = pathlib.Path(__file__).parent.parent
+SESSIONS_JS_PATH = ROOT / "static" / "sessions.js"
+NODE = shutil.which("node")
+
+pytestmark = pytest.mark.skipif(NODE is None, reason="node not on PATH")
+
+
+def _run_node(source: str) -> str:
+    with tempfile.NamedTemporaryFile(
+        "w", suffix=".cjs", encoding="utf-8", dir=ROOT, delete=False
+    ) as script:
+        script.write(source)
+        script_path = pathlib.Path(script.name)
+    try:
+        result = subprocess.run(
+            [NODE, str(script_path)],
+            cwd=str(ROOT),
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+    finally:
+        script_path.unlink(missing_ok=True)
+    if result.returncode != 0:
+        raise RuntimeError(result.stderr)
+    return result.stdout.strip()
+
+
+def _extract_func_script(js: str) -> str:
+    # Brace-matches a function body while skipping braces inside string / template
+    # / regex literals and comments (same hardened extractor as the sibling #5744
+    # suite).
+    prelude = "const src = " + json.dumps(js) + ";\n"
+    body = r"""
+function extractFunc(name) {
+  const re = new RegExp('function\\s+' + name + '\\s*\\(');
+  const start = src.search(re);
+  if (start < 0) throw new Error(name + ' not found');
+  let i = src.indexOf('{', start);
+  let depth = 1; i++;
+  let str = null;
+  let inLine = false;
+  let inBlock = false;
+  let inRegex = false;
+  let prev = '';
+  while (depth > 0 && i < src.length) {
+    const c = src[i];
+    const n = src[i + 1];
+    if (inLine) { if (c === '\n') inLine = false; i++; continue; }
+    if (inBlock) { if (c === '*' && n === '/') { inBlock = false; i++; } i++; continue; }
+    if (str) {
+      if (c === '\\') { i += 2; continue; }
+      if (c === str) str = null;
+      i++; continue;
+    }
+    if (inRegex) {
+      if (c === '\\') { i += 2; continue; }
+      if (c === '/') inRegex = false;
+      i++; continue;
+    }
+    if (c === '/' && n === '/') { inLine = true; i += 2; continue; }
+    if (c === '/' && n === '*') { inBlock = true; i += 2; continue; }
+    if (c === '"' || c === "'" || c === '`') { str = c; i++; continue; }
+    if (c === '/' && !'})]0123456789'.includes(prev) && !/[A-Za-z_$]/.test(prev)) {
+      inRegex = true; i++; continue;
+    }
+    if (c === '{') depth++;
+    else if (c === '}') depth--;
+    if (c.trim()) prev = c;
+    i++;
+  }
+  return src.slice(start, i);
+}
+"""
+    return prelude + body
+
+
+def _js_prelude() -> str:
+    """Globals/stubs needed by the extracted sessions.js functions.
+
+    `_markPollingCompletionUnreadTransitions` reads/writes the module-level
+    `_sessionStreamingById` / `_sessionListSnapshotById` Maps directly, and
+    calls `_isSessionEffectivelyStreaming` (real), `_isSessionActivelyViewedForList`,
+    `_getSessionObservedStreaming`, `_markSessionCompletionUnread`,
+    `_setSessionViewedCount`, `_rememberObservedStreamingSession` and
+    `_forgetObservedStreamingSession`. All the `typeof`-guarded collaborators
+    (`_sessionListSourceById`, `_allSessionsScope`, `_rememberSessionListSource`,
+    `_cronCompletionUnreadMetaForSession`, `_showAllProfiles`,
+    `_cronMarkerProfileMatchesActive`) are intentionally left undefined so the
+    guards take their safe branches.
+    """
+    return r"""
+const _sessionStreamingById = new Map();
+const _sessionListSnapshotById = new Map();
+const _observedStreaming = {};
+let markCount = 0;
+const markedSids = [];
+function _getSessionObservedStreaming() { return _observedStreaming; }
+function _markSessionCompletionUnread(sid, messageCount, meta) { markCount += 1; markedSids.push(sid); }
+function _setSessionViewedCount(sid, messageCount) {}
+function _isSessionActivelyViewedForList(sid) { return false; }
+function _rememberObservedStreamingSession(s) {}
+function _forgetObservedStreamingSession(sid) {}
+let S = { session: null, busy: false };
+"""
+
+
+def _cron_row(sid, created_at, **overrides):
+    row = {
+        "session_id": sid,
+        "title": "Cron Session",
+        "profile": "default",
+        "created_at": created_at,
+        "updated_at": created_at,
+        "last_message_at": created_at,
+        "message_count": 1,
+        "user_message_count": 1,
+        "archived": False,
+        "project_id": "cron-project",
+        "source_tag": "cron",
+        "raw_source": "cron",
+        "session_source": "cron",
+        "source_label": "Cron",
+        "is_cli_session": False,
+    }
+    row.update(overrides)
+    return row
+
+
+def test_overlay_marks_current_running_cron_row(monkeypatch):
+    import api.routes as routes
+    import api.route_session_list_cache as slc
+
+    # job6728 started at epoch 1000; the current run's session was created at
+    # 1100 (>= start) and carries the matching cron_{job_id}_ prefix.
+    monkeypatch.setattr(routes, "_RUNNING_CRON_JOBS", {"job6728": 1000.0})
+    monkeypatch.setattr(slc, "_session_list_cache_active_stream_ids", lambda: set())
+
+    rows = slc._session_list_cache_overlay_runtime_rows(
+        [_cron_row("cron_job6728_20260803_100000", created_at=1100)]
+    )
+    assert len(rows) == 1
+    assert rows[0]["cron_running"] is True
+
+
+def test_overlay_does_not_mark_finished_or_other_cron_rows(monkeypatch):
+    import api.routes as routes
+    import api.route_session_list_cache as slc
+
+    # job6728 is running (started at 1000) but:
+    # - cron_job9999... belongs to a different job (not tracked)
+    # - cron_job6728_20260701... is an OLD run of the same job (created < start)
+    monkeypatch.setattr(routes, "_RUNNING_CRON_JOBS", {"job6728": 1000.0})
+    monkeypatch.setattr(slc, "_session_list_cache_active_stream_ids", lambda: set())
+
+    rows = slc._session_list_cache_overlay_runtime_rows(
+        [
+            _cron_row("cron_job9999_20260803_100000", created_at=1100),
+            _cron_row("cron_job6728_20260701_050000", created_at=500),
+            {"session_id": "regular-chat", "title": "Chat", "is_cli_session": False},
+        ]
+    )
+    by_sid = {row["session_id"]: row for row in rows}
+    assert by_sid["cron_job9999_20260803_100000"]["cron_running"] is False
+    assert by_sid["cron_job6728_20260701_050000"]["cron_running"] is False
+    assert by_sid["regular-chat"]["cron_running"] is False
+
+
+def test_overlay_fails_closed_when_routes_unavailable(monkeypatch):
+    import api.route_session_list_cache as slc
+
+    monkeypatch.setattr(slc, "_session_list_cache_running_cron_jobs", lambda: {})
+
+    rows = slc._session_list_cache_overlay_runtime_rows(
+        [_cron_row("cron_job6728_20260803_100000", created_at=1100)]
+    )
+    assert rows[0]["cron_running"] is False
+
+
+def test_overlay_nested_job_prefixes_do_not_cross_claim(monkeypatch):
+    """Overlapping cron job ids (backup vs backup_full) must not cross-claim.
+
+    ``cron_backup_`` is a valid prefix of ``cron_backup_full_...``, so a
+    first-match scan in insertion order would attribute a ``backup_full``
+    session to the running ``backup`` job, and a fall-through time-miss would
+    let ``backup`` claim an OLD completed ``backup_full`` run. The
+    longest-prefix-first scan must attribute each session to its own job with
+    no fall-through (mirrors api.routes._latest_cron_session_info_for_jobs).
+    """
+    import api.routes as routes
+    import api.route_session_list_cache as slc
+
+    # Both jobs running; `backup` is inserted FIRST so a first-match scan in
+    # insertion order would claim cron_backup_full_... sessions as `backup`.
+    monkeypatch.setattr(
+        routes, "_RUNNING_CRON_JOBS", {"backup": 1000.0, "backup_full": 2000.0}
+    )
+    monkeypatch.setattr(slc, "_session_list_cache_active_stream_ids", lambda: set())
+
+    rows = slc._session_list_cache_overlay_runtime_rows(
+        [
+            _cron_row("cron_backup_20260803_100000", created_at=1100),
+            _cron_row("cron_backup_full_20260803_210000", created_at=2100),
+            _cron_row("cron_backup_full_20260803_190000", created_at=1900),
+        ]
+    )
+    by_sid = {row["session_id"]: row for row in rows}
+    assert by_sid["cron_backup_20260803_100000"]["cron_running"] is True
+    assert by_sid["cron_backup_full_20260803_210000"]["cron_running"] is True
+    # Old backup_full run created after `backup` started: must NOT fall through
+    # to the shorter cron_backup_ prefix (it belongs to a completed run).
+    assert by_sid["cron_backup_full_20260803_190000"]["cron_running"] is False
+
+    # Single running job, non-overlapping id: unchanged behavior.
+    monkeypatch.setattr(routes, "_RUNNING_CRON_JOBS", {"job6728": 1000.0})
+    rows = slc._session_list_cache_overlay_runtime_rows(
+        [_cron_row("cron_job6728_20260803_100000", created_at=1100)]
+    )
+    assert rows[0]["cron_running"] is True
+
+
+def test_overlay_shorter_running_job_cannot_claim_longer_job_session(monkeypatch):
+    """A running `backup` must NOT claim a `backup_full` session when only
+
+    `backup` is running (#6728 gate finding). Prefixes are built ONLY from
+    RUNNING jobs, so if `backup_full` is not running, longest-prefix sorting
+    never sees the true longer owner and `cron_backup_` would otherwise swallow
+    `cron_backup_full_YYYYMMDD_HHMMSS`. The remainder after the matched prefix
+    must be EXACTLY a run timestamp, so `full_20260803_210000` is rejected.
+    """
+    import api.routes as routes
+    import api.route_session_list_cache as slc
+
+    # Only `backup` is running; `backup_full` is NOT tracked.
+    monkeypatch.setattr(routes, "_RUNNING_CRON_JOBS", {"backup": 1000.0})
+    monkeypatch.setattr(slc, "_session_list_cache_active_stream_ids", lambda: set())
+
+    rows = slc._session_list_cache_overlay_runtime_rows(
+        [
+            _cron_row("cron_backup_20260803_100000", created_at=1100),
+            _cron_row("cron_backup_full_20260803_210000", created_at=2100),
+        ]
+    )
+    by_sid = {row["session_id"]: row for row in rows}
+    # The genuine `backup` run is live.
+    assert by_sid["cron_backup_20260803_100000"]["cron_running"] is True
+    # The `backup_full` session must stay completed — `backup`'s prefix leaves
+    # `full_20260803_210000`, which is not a bare run timestamp.
+    assert by_sid["cron_backup_full_20260803_210000"]["cron_running"] is False
+
+
+def test_sidebar_response_preserves_cron_running(monkeypatch):
+    import api.routes as routes
+
+    row = _cron_row("cron_job6728_20260803_100000", created_at=1100)
+    row["cron_running"] = True
+    item = routes._sidebar_session_response_item(row)
+    assert item.get("cron_running") is True
+
+
+def test_payload_response_roundtrip_stamps_running_cron(monkeypatch):
+    """End-to-end: /api/sessions response rows carry cron_running for live jobs."""
+    import api.routes as routes
+
+    raw_cron_row = _cron_row("cron_job6728_20260803_100000", created_at=1100)
+    monkeypatch.setattr(routes, "all_sessions", lambda diag=None: [])
+    monkeypatch.setattr(
+        routes, "get_cli_sessions", lambda source_filter=None, all_profiles=False: [raw_cron_row]
+    )
+    monkeypatch.setattr(
+        routes, "_reconcile_stale_stream_state_for_session_rows", lambda _sessions: False
+    )
+    monkeypatch.setattr(routes, "_RUNNING_CRON_JOBS", {"job6728": 1000.0})
+
+    payload = routes._build_session_list_cache_payload(
+        active_profile="default",
+        all_profiles=False,
+        show_cli_sessions=True,
+        show_previous_messaging_sessions=False,
+        show_cron_sessions=True,
+    )
+    response = routes._session_list_payload_to_response(payload)
+    rows = response["sessions"]
+    matching = [row for row in rows if row["session_id"] == "cron_job6728_20260803_100000"]
+    assert len(matching) == 1
+    assert matching[0]["cron_running"] is True
+
+
+def test_running_cron_row_renders_effectively_streaming():
+    """A row with cron_running=true must render ACTIVE in the sidebar.
+
+    Rendering, sorting, polling and completion-transition tracking all key off
+    `_isSessionEffectivelyStreaming` (static/sessions.js). Before the fix the
+    predicate ignored `cron_running`, so a running cron row rendered visually
+    idle and could lose its eventual completion/unread transition.
+    """
+    js = SESSIONS_JS_PATH.read_text(encoding="utf-8")
+    source = _extract_func_script(js) + _js_prelude() + r"""
+eval(extractFunc('_hasPendingUserMessageSignal'));
+eval(extractFunc('_isSessionLocallyStreaming'));
+eval(extractFunc('_isSessionEffectivelyStreaming'));
+const running = { session_id: 'cron_job6728_20260803_100000', cron_running: true, is_streaming: false };
+const idle = { session_id: 'cron_job6728_20260701_050000', cron_running: false, is_streaming: false };
+console.log(JSON.stringify({
+  runningActive: _isSessionEffectivelyStreaming(running),
+  idleInactive: _isSessionEffectivelyStreaming(idle),
+}));
+"""
+    m = json.loads(_run_node(source))
+    assert m["runningActive"] is True
+    assert m["idleInactive"] is False
+
+
+def test_cron_running_defers_completion_unread_until_flag_clears():
+    """A cron_running row with advanced message_count must NOT be marked
+    completed-unread; exactly ONE true->false completion transition happens
+    when the flag clears.
+
+    The `!cronRunning` guards in `_markPollingCompletionUnreadTransitions`
+    defer all three completion signals while the job runs; the focused
+    regression proves the row is only marked once, on the poll where
+    cron_running flips to false.
+    """
+    js = SESSIONS_JS_PATH.read_text(encoding="utf-8")
+    source = _extract_func_script(js) + _js_prelude() + r"""
+eval(extractFunc('_hasPendingUserMessageSignal'));
+eval(extractFunc('_isSessionLocallyStreaming'));
+eval(extractFunc('_isSessionEffectivelyStreaming'));
+eval(extractFunc('_markPollingCompletionUnreadTransitions'));
+const sid = 'cron_job6728_20260803_100000';
+_sessionListSnapshotById.set(sid, { message_count: 1, last_message_at: 1000 });
+let row = {
+  session_id: sid,
+  cron_running: true,
+  is_streaming: false,
+  message_count: 2,
+  last_message_at: 1100,
+};
+// Poll 1: cron still running, message_count advanced -> no completion mark.
+_markPollingCompletionUnreadTransitions([row]);
+const marksWhileRunning = markCount;
+const streamWhileRunning = _sessionStreamingById.get(sid);
+// Poll 2: cron flag clears -> exactly one true->false completion transition.
+row.cron_running = false;
+_markPollingCompletionUnreadTransitions([row]);
+const marksAfterClear = markCount;
+const streamAfterClear = _sessionStreamingById.get(sid);
+console.log(JSON.stringify({
+  marksWhileRunning,
+  marksAfterClear,
+  streamWhileRunning,
+  streamAfterClear,
+  markedSids,
+}));
+"""
+    m = json.loads(_run_node(source))
+    assert m["marksWhileRunning"] == 0, (
+        "cron_running row with advanced message_count must not be marked completed-unread"
+    )
+    assert m["marksAfterClear"] == 1, "exactly one completion transition when the flag clears"
+    assert m["streamWhileRunning"] is True, "row must render active while cron_running"
+    assert m["streamAfterClear"] is False
+    assert m["markedSids"] == ["cron_job6728_20260803_100000"]

tests/test_issue856_background_completion_unread.py

@@ -362,11 +362,11 @@ def test_polling_transition_marks_completion_when_long_running_stream_snapshot_a
     assert "function _forgetObservedStreamingSession(" in SESSIONS_JS
     assert "const previousSnapshot = _sessionListSnapshotById.get(sid);" in transition_block
     assert "const observedStreaming = _getSessionObservedStreaming()[sid];" in transition_block
-    assert "const completedWithNewMessages = Boolean(" in transition_block
+    assert "const completedWithNewMessages = !cronRunning && Boolean(" in transition_block
     assert "(previousSnapshot || observedStreaming)" in transition_block
     assert "messageCount > Number((previousSnapshot || observedStreaming).message_count || 0)" in transition_block
     assert "lastMessageAt > Number((previousSnapshot || observedStreaming).last_message_at || 0)" in transition_block
-    assert "const completedPersistedObservedStream = Boolean(observedStreaming && !isStreaming);" in transition_block
+    assert "const completedPersistedObservedStream = !cronRunning && Boolean(observedStreaming && !isStreaming);" in transition_block
     assert "completedObservedStream || completedPersistedObservedStream || completedWithNewMessages" in transition_block
     assert "_sessionListSnapshotById.set(sid, {" in transition_block
     assert "_rememberRenderedSessionSnapshot(s);" in render_block, (
@@ -383,7 +383,7 @@ def test_polling_snapshot_fallback_does_not_mark_first_seen_historical_sessions(
     )
 
     prev_idx = transition_block.find("const previousSnapshot = _sessionListSnapshotById.get(sid);")
-    fallback_idx = transition_block.find("const completedWithNewMessages = Boolean(")
+    fallback_idx = transition_block.find("const completedWithNewMessages = !cronRunning && Boolean(")
     mark_idx = transition_block.find("_markSessionCompletionUnread(sid")
     snapshot_set_idx = transition_block.find("_sessionListSnapshotById.set(sid, {")