Merge master (exp-v0.52.362) into stage/7610-oob-writeback

# Conflicts:
#	CHANGELOG.md

CHANGELOG.md

@@ -33,6 +33,16 @@
   last tool result. After the turn settled, that wrapper stayed visible in the chat transcript. The
   settled-transcript writeback now scrubs the consumed wrapper from the rows it is built from.
   (#7610 by @webtecnica)
+- **Saving a very large session no longer reads and parses the whole file just to count messages.**
+  The #1558 backup safeguard in `Session.save()` needs the on-disk message count; it obtained it by
+  loading the entire sidecar. On a real 203 MB / 266,940-message session that made every save
+  expensive. The count is now taken without a full parse, and the shrink-backup behaviour is
+  unchanged. (#7578 by @rodrigogs)
+- **A cancelled or recovered turn no longer duplicates an answer that was already saved (#6366).**
+  When a completed assistant turn had been persisted and a later cancel/recovery path ran for the
+  same pending turn, recovery could append a duplicate user turn plus a `_partial` clone of the
+  journal. Recovery now stops once the transcript has already advanced past the pending turn.
+  (#7682 by @happy5318)
 
 - **The live stream reports which model actually served the turn.** A new additive `runtime_model`
   SSE event carries the model (and provider, when known) that the Agent reported while producing

api/gateway_chat.py

@@ -38,6 +38,7 @@ from api.config import (
 from api.helpers import _redact_text, redact_session_data
 from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only
 from api.run_journal import RunJournalWriter, bound_run_journal_snapshot_args
+from api.turn_journal import append_turn_journal_event_for_stream
 
 logger = logging.getLogger(__name__)
 
@@ -1404,6 +1405,21 @@ def _run_gateway_chat_streaming(
             if cancel_event.is_set():
                 _restore_cancelled_success_writeback()
                 return
+            # #6366 re-gate: record the durable same-stream completion
+            # event in the crash-safe turn journal. The run journal's
+            # terminal state is only reached on its own ``stream_end``
+            # write path, so a Gateway run whose terminal write is lost
+            # would otherwise leave no completion evidence at all and
+            # stale-cancel recovery would re-append a duplicate
+            # recovered row after the valid final answer.
+            try:
+                append_turn_journal_event_for_stream(
+                    session_id,
+                    stream_id,
+                    {"event": "completed", "created_at": time.time()},
+                )
+            except Exception:
+                logger.debug("Failed to append completed turn journal event", exc_info=True)
             success_writeback_committed = True
         try:
             from api.goals import evaluate_goal_after_turn, has_active_goal

api/models.py

@@ -1107,6 +1107,14 @@ _METADATA_PREFIX_MAX_BYTES = 1024 * 1024
 #: even when the stop key sits at 2 KB.
 _METADATA_PREFIX_FIRST_STAGE_BYTES = 64 * 1024
 
+# Marks a sidecar as written by the CURRENT writer contract, where the
+# persisted `message_count` equals len(messages) by construction (both keys
+# land in the same atomic write). save()'s bounded-prefix shrink check trusts
+# a persisted count ONLY when this marker sits next to it in the file; see
+# _prefix_message_count(). Bump the value whenever the writer contract changes
+# meaning, so counts from an older contract fall back to the full parse.
+_MESSAGE_COUNT_MARKER = 1
+
 
 def _read_metadata_json_prefix(path, max_prefix_bytes=_METADATA_PREFIX_MAX_BYTES):
     """Read only the metadata portion before the large arrays.
@@ -1505,6 +1513,16 @@ class Session:
         # legacy-format reader that stops at a scene key still finds the count.
         # The full anchor_activity_scenes bodies serialize AFTER messages.
         meta['message_count'] = len(self.messages or [])
+        # _mc_v marks this file as written by the current writer contract,
+        # where `message_count` equals len(messages) by construction and both
+        # keys land in the same atomic write. save()'s shrink guard takes the
+        # bounded-prefix shortcut ONLY for marked files; an unmarked count
+        # (a legacy pre-#5854 sidecar, a sidecar materialized by an older
+        # recovery writer, any foreign writer) gets the full parse, so a stale
+        # count from outside this contract can never read a real shrink as a
+        # growth and skip the #1558 backup. One save re-marks the file, so the
+        # fast path still covers steady state.
+        meta['_mc_v'] = _MESSAGE_COUNT_MARKER
         meta['anchor_scene_index'] = _anchor_scene_index_from_records(self.anchor_activity_scenes)
         # Keep the in-memory fingerprint aligned with what we just persisted, so a
         # later metadata-only reload of THIS object (or any fingerprint reader)
@@ -1517,7 +1535,7 @@ class Session:
         meta['anchor_activity_scenes'] = self.anchor_activity_scenes if isinstance(self.anchor_activity_scenes, dict) else {}
         # Fields not in METADATA_FIELDS (e.g. last_usage) go at the end. Exclude
         # the keys we placed explicitly above so they aren't emitted twice.
-        _placed = {'message_count', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'}
+        _placed = {'message_count', '_mc_v', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'}
         extra = {k: v for k, v in self.__dict__.items()
                  if k not in METADATA_FIELDS and k not in _placed
                  and not k.startswith('_')}
@@ -1537,12 +1555,46 @@ class Session:
         # their .bak get restored automatically.
         try:
             if self.path.exists():
-                existing_text = self.path.read_text(encoding='utf-8')
-                try:
-                    existing = json.loads(existing_text)
-                    existing_msg_count = len(existing.get('messages') or [])
-                except (json.JSONDecodeError, ValueError):
-                    existing_msg_count = -1  # corrupt → always back up
+                # The on-disk count, without reading the body.
+                #
+                # The decision below is a function of ONE integer -- how many
+                # messages the file on disk holds -- and save() already writes
+                # that integer into the metadata prefix, before `messages`, as
+                # `message_count` (see METADATA_FIELDS above; load_metadata_only
+                # and the sidebar freshness check read it the same way). So read
+                # THAT through a bounded 64 KiB prefix instead of the whole file.
+                # Measured before this: a 203,439,398-byte sidecar cost 20,377 ms
+                # (17,453 in read_text, 2,924 in json.loads) to yield one integer,
+                # on EVERY save -- including the grow-saves that never back
+                # anything up. The prefix read is O(64 KiB), and because the count
+                # is part of the bytes on disk it travels with any rewrite of them.
+                #
+                # An in-memory "I wrote this, stat says nothing changed" cache is
+                # NOT sufficient here, and was removed after review: (inode, size,
+                # mtime_ns) is not a content identity. A same-length in-place
+                # rewrite inside one mtime tick keeps all three fields -- ext4
+                # stamps mtime from a coarse clock, so two writes in the same tick
+                # share one mtime_ns -- and a stale cached count then reads a real
+                # shrink as a growth and skips the #1558 backup. The prefix count
+                # cannot be fooled that way.
+                #
+                # Every unknown falls through to the full read + parse below: a
+                # legacy (pre-#5854) sidecar whose count is not in the prefix, a
+                # count written without the current writer's _mc_v marker (an
+                # older writer's count can be stale relative to the messages
+                # array next to it), a corrupt or truncated prefix, a file with
+                # no top-level `messages` key at all, or metadata alone that
+                # overflows the budget.
+                # Fail-open is the contract -- never "assume no shrink".
+                existing_text = None
+                existing_msg_count = _prefix_message_count(self.path)
+                if existing_msg_count is None:
+                    existing_text = self.path.read_text(encoding='utf-8')
+                    try:
+                        existing = json.loads(existing_text)
+                        existing_msg_count = len(existing.get('messages') or [])
+                    except (json.JSONDecodeError, ValueError):
+                        existing_msg_count = -1  # corrupt → always back up
                 incoming_msg_count = len(self.messages or [])
                 if (
                     existing_msg_count > 0
@@ -1560,6 +1612,10 @@ class Session:
                     return
                 if existing_msg_count > incoming_msg_count:
                     bak_path = self.path.with_suffix('.json.bak')
+                    if existing_text is None:
+                        # The .bak body is the one thing that needs the full text,
+                        # and a shrink is the one time it is needed.
+                        existing_text = self.path.read_text(encoding='utf-8')
                     # SHOULD-FIX #2 (Opus): atomic write via tmp+replace,
                     # mirroring the main save() pattern below. Prevents a
                     # torn .bak from a crash mid-write or a concurrent
@@ -2442,6 +2498,191 @@ def _latest_user_matches_pending_text(messages, pending_text):
     return False
 
 
+def _pending_active_turn_token(session):
+    """Return the exact WebUI active-turn token for the pending turn.
+
+    The token is ``"<stream_id>:<pending_started_at>"`` — the same value the
+    eager user-message checkpoint and the agent-result merge stamp onto the
+    materialized user row (``stamp_message_source``). It is unforgeable
+    per-turn identity: two distinct streams that submit the same prompt
+    inside the same second produce different tokens.
+    """
+    from api.process_event_utils import build_active_turn_token
+
+    return build_active_turn_token(
+        getattr(session, 'active_stream_id', None),
+        getattr(session, 'pending_started_at', None),
+    )
+
+
+def _transcript_user_row_is_pending_turn(message, session, pending_token) -> bool:
+    """Return True only when a transcript user row provably IS the pending turn.
+
+    Identity is bound to the active-turn token, never to integer-second
+    timestamp equality: two different streams that send the same prompt
+    within one second truncate to the same ``int(timestamp)``, so the
+    checkpoint matcher alone would accept the *other* stream's row as the
+    pending turn. A token mismatch therefore always loses, and a row that
+    carries no token at all cannot be proven to be this turn — which means
+    the caller must recover, not suppress.
+    """
+    if not isinstance(message, dict) or message.get('role') != 'user':
+        return False
+    row_token = message.get('_active_turn_token')
+    if pending_token:
+        if not row_token:
+            # Ambiguous legacy identity: the pending turn has a resolvable
+            # token but this row predates token stamping. We cannot prove it
+            # is the same turn, so recovery must run.
+            return False
+        return row_token == pending_token
+    if row_token:
+        # The row belongs to a different, token-bearing turn. Never let a
+        # same-second timestamp collision override that mismatch.
+        return False
+    # Neither side carries a token (legacy/dead-stream pending state): fall
+    # back to the strict content + timestamp + source + attachments
+    # checkpoint, which is the only identity signal available.
+    return _message_matches_pending_checkpoint(
+        message,
+        getattr(session, 'pending_user_message', None),
+        getattr(session, 'pending_started_at', None),
+        getattr(session, 'pending_user_source', None),
+        getattr(session, 'pending_attachments', None),
+    )
+
+
+def _pending_turn_has_final_assistant_answer(messages, start_idx: int) -> bool:
+    """Return True when the matched user turn ends in a genuine final answer.
+
+    Reuses the established final-answer semantics
+    (``_assistant_message_has_final_visible_text``) instead of "an assistant
+    row exists". An assistant row only settles the turn when it carries real
+    visible answer text: empty rows, tool-call-only rows, interim
+    ``_partial`` progress rows, rows recovered from the run journal and
+    compaction reference cards are all rejected, so a turn that was
+    interrupted mid-tool-execution is never mistaken for a completed one.
+
+    The scan stops at the next real user row — a final answer that belongs
+    to a *later* turn must not settle this pending turn.
+    """
+    from api.streaming import (
+        _assistant_message_has_final_visible_text,
+        _is_synthetic_control_message,
+    )
+
+    for later in messages[start_idx + 1:]:
+        if not isinstance(later, dict):
+            continue
+        if _is_synthetic_control_message(later):
+            continue
+        role = later.get('role')
+        if role == 'user':
+            if is_context_compression_marker(later):
+                # Synthetic compaction cards are not a user-turn boundary.
+                continue
+            return False
+        if role != 'assistant':
+            # Tool rows never prove a final answer.
+            continue
+        if (
+            later.get('_partial')
+            or later.get('_error')
+            or later.get('_recovered')
+            or later.get('_recovered_from_run_journal')
+        ):
+            continue
+        if is_context_compression_marker(later):
+            continue
+        if _assistant_message_has_final_visible_text(later):
+            return True
+    return False
+
+
+def _transcript_already_advanced_past_pending(session) -> bool:
+    """#6366: a stale pending user message whose own user row already appears
+    in the durable transcript, followed by a genuine final assistant answer
+    inside that same user-turn boundary, means the transcript has already
+    advanced past this pending turn. Returning True here lets the recovery
+    path skip the recovered user / ``_partial`` clone / journal replay /
+    generic no-response error that would otherwise append after the valid
+    final answer. The tail-only check at line ~3485 (``session.messages[-1]``)
+    misses this case: when the pending turn's user row is older than
+    the transcript tail, the tail is a *newer* assistant row that
+    can never match the pending **user** checkpoint, and the recovery
+    path falls through into the append branch.
+
+    Identity is bound to the pending stream's exact active-turn token
+    (``_transcript_user_row_is_pending_turn``), so a repeated prompt from a
+    different stream — even one submitted in the same integer second —
+    can never be matched against this pending turn and silently dropped.
+
+    Completion requires a genuine final visible assistant answer
+    (``_pending_turn_has_final_assistant_answer``): empty, tool-call-only,
+    interim ``_partial``, journal-recovered and compaction tails all keep
+    recovery active, because that is precisely the interrupted-turn shape
+    the recovery path exists for.
+
+    The transcript heuristic alone is not durable: an unflagged interim
+    prose row ("Let me check the logs first.") followed by tool rows that
+    were never followed by a final answer in the transcript passes the
+    predicate as a "completed" turn, but the stream may have died mid-
+    tool and the run journal was never marked done. 9/23 re-gate: also
+    require durable same-stream terminal evidence. The run journal
+    cannot supply that evidence on this path — the caller in
+    ``_apply_core_sync_or_error_marker`` already returns early whenever
+    the run journal reports ``completed`` for the stream, so the branch
+    that reaches this predicate can never observe that state (which is
+    why the previous run-journal gate was unreachable dead code). The
+    evidence therefore has to come from the turn journal, which records
+    the exact-stream ``completed`` event for the pending turn
+    (``_turn_journal_records_completion``). Without it, fall through to
+    the normal journal-recovery path so the partial output is replayed
+    and the interruption is marked.
+
+    The suppression is therefore deliberately conservative: whenever the
+    turn's identity, its completion, or its durable terminal evidence
+    cannot be positively proven, this helper returns False and the
+    prompt is recovered instead of discarded. A false negative leaves a
+    visible cosmetic duplicate; a false positive silently destroys a
+    prompt or a response.
+
+    Idempotent: this is a pure read that does not mutate the session.
+    Running recovery twice therefore produces the same outcome.
+    """
+    pending_text = getattr(session, 'pending_user_message', None)
+    if not pending_text:
+        return False
+    messages = getattr(session, 'messages', None)
+    if not isinstance(messages, list):
+        return False
+    pending_token = _pending_active_turn_token(session)
+    for idx, message in enumerate(messages):
+        if not _transcript_user_row_is_pending_turn(message, session, pending_token):
+            continue
+        # Found the pending turn's own user row at ``idx``. A genuine
+        # final answer inside that turn's boundary is necessary but not
+        # sufficient: durable same-stream terminal evidence must also
+        # exist, otherwise unflagged interim prose
+        # ("Let me check the logs first.") followed by tool rows can be
+        # mistaken for a finished turn.
+        if not _pending_turn_has_final_assistant_answer(messages, idx):
+            continue
+        stream_id = getattr(session, 'active_stream_id', None)
+        # The turn journal — not the run journal — carries the
+        # completion evidence that is still observable here: the caller
+        # (``_apply_core_sync_or_error_marker``) already consumed the
+        # run-journal ``completed`` state on its own early return, so a
+        # run-journal gate at this depth can never fire.
+        if not _turn_journal_records_completion(session, stream_id):
+            # No durable same-stream terminal evidence — the turn may
+            # have died mid-tool. Fall through to the journal-recovery
+            # path instead of suppressing it.
+            return False
+        return True
+    return False
+
+
 def _partial_message_signature(message: dict) -> tuple:
     """Return a stable identity for partial assistant markers recovered on load."""
     if not isinstance(message, dict):
@@ -2643,6 +2884,54 @@ def _run_journal_terminal_state(session, stream_id: str | None) -> str | None:
     return str(terminal.get('terminal_state') or '') or None
 
 
+def _turn_journal_records_completion(session, stream_id: str | None) -> bool:
+    """Return True when the crash-safe turn journal proves the turn for
+    ``stream_id`` reached its terminal ``completed`` state.
+
+    The turn journal is written per stream/turn id by the exact-stream
+    workers and by :mod:`api.routes` on submission, so a ``completed``
+    event for the *pending stream id* is completion evidence that
+    survives a run-journal write failure: the run journal only reaches
+    ``done`` / ``stream_end`` on its own write path, and a process that
+    dies between the turn journal write and the run-journal terminal
+    append leaves the run journal without terminal state while the turn
+    journal still records that the turn actually finished.
+
+    Bound strictly to ``stream_id`` — a ``completed`` turn belonging to
+    another stream of the same session never settles this one — and to
+    the latest turn recorded for that stream, so a turn that was
+    cancelled or crashed (``interrupted``) keeps recovery active even
+    when an earlier turn of the same stream completed normally.
+    """
+    if not stream_id:
+        return False
+    try:
+        from api.turn_journal import (
+            derive_turn_journal_states,
+            read_turn_journal,
+        )
+        journal = read_turn_journal(session.session_id)
+        states, _ = derive_turn_journal_states(journal.get('events') or [])
+    except Exception:
+        return False
+    latest: tuple[float, str, dict] | None = None
+    for turn_id, event in states.items():
+        if str(event.get('stream_id') or '') != str(stream_id):
+            continue
+        try:
+            created_at = float(event.get('created_at') or 0)
+        except (TypeError, ValueError):
+            created_at = 0.0
+        # ``>=`` on both elements keeps the winner deterministic when two
+        # turns of the same stream share a ``created_at``.
+        candidate = (created_at, turn_id, event)
+        if latest is None or candidate[:2] >= latest[:2]:
+            latest = candidate
+    if latest is None:
+        return False
+    return latest[2].get('event') == 'completed'
+
+
 def _recoverable_unsaved_gateway_terminal_error(
     session,
     stream_id: str | None,
@@ -3519,6 +3808,29 @@ def _apply_core_sync_or_error_marker(
             )
             return True
         if not _tail_user_already_checkpointed:
+            # #6366 re-gate: when the durable transcript has already
+            # advanced past this pending turn into a newer settled
+            # user/assistant boundary, the recovery path must NOT
+            # append a recovered user row + ``_partial`` clone +
+            # journal replay + generic no-response error after the
+            # valid final answer. The tail-only check above misses
+            # that case (the tail is a newer assistant row that can
+            # never match the pending user checkpoint). Clear only
+            # the stale pending fields and return; the transcript is
+            # already correct and durable.
+            if _transcript_already_advanced_past_pending(session):
+                session.active_stream_id = None
+                session.pending_user_message = None
+                session.pending_attachments = []
+                session.pending_started_at = None
+                session.pending_user_source = None
+                session.save(touch_updated_at=touch_updated_at)
+                logger.info(
+                    "Session %s: cleared stale pending state for stream %s — transcript already advanced past this turn",
+                    sid,
+                    _stream_id,
+                )
+                return True
             _append_recovered_pending_turn(session, timestamp=_recovered_ts)
         else:
             recovered = {
@@ -4239,6 +4551,57 @@ def _sidecar_stat_signature(path):
             int(st.st_size), int(getattr(st, 'st_ctime_ns', int(st.st_ctime * 1_000_000_000))))
 
 
+_MESSAGE_COUNT_MARKER = 1
+
+
+def _prefix_message_count(path):
+    """The sidecar's own persisted ``message_count``, from a bounded prefix.
+
+    The #1558 shrink check in save() needs exactly one integer -- how many
+    messages the file on disk holds -- and the writer puts that integer in the
+    metadata prefix, before ``messages``, precisely so readers can have it
+    without parsing the body (see METADATA_FIELDS). Reading the prefix costs
+    O(64 KiB) against O(file size) for ``read_text`` + ``json.loads`` (measured
+    2026-09-15: 20,377 ms for one integer on a 203,439,398-byte sidecar).
+
+    Content-derived on purpose. A stat identity cannot stand in for it: an
+    in-place rewrite of the same length inside one mtime tick keeps inode, size
+    *and* mtime_ns, so a cached count would read a real shrink as a growth and
+    skip the backup. The count is part of the bytes, so any rewrite of them
+    rewrites it too.
+
+    Gated on the writer marker (``_mc_v``, review 2026-09-22): a count is only
+    trusted when the SAME write that produced the messages array also vouched
+    for the count. Anything unmarked -- a legacy pre-#5854 sidecar, a sidecar
+    materialized by an older recovery writer whose denormalized count could be
+    stale against its rows, any foreign writer -- returns None and the caller
+    falls back to the full read + parse. After one save by the current writer
+    the file carries the marker and the fast path resumes.
+
+    Returns None -- meaning "the caller must fall back to the full read +
+    parse" -- for every shape that does not carry a usable MARKED count: an
+    unmarked or wrong-marker file, a legacy sidecar whose count is not in the
+    prefix, a file with no top-level ``messages`` key at all, a corrupt or
+    truncated prefix, an unreadable path, or metadata alone that overflows the
+    budget.
+    """
+    try:
+        raw = _read_metadata_json_prefix(path)
+    except (OSError, ValueError):
+        return None
+    if not raw:
+        return None
+    try:
+        parsed = json.loads(raw)
+    except (json.JSONDecodeError, ValueError):
+        return None
+    if not isinstance(parsed, dict):
+        return None
+    if parsed.get('_mc_v') != _MESSAGE_COUNT_MARKER:
+        return None
+    return _parse_nonnegative_int(parsed.get('message_count'))
+
+
 def _legacy_sidecar_facts_get(sid):
     """Return cached authoritative facts for a LEGACY sidecar, or None (#5854).
 

api/session_recovery.py

@@ -601,7 +601,14 @@ def _state_db_row_to_sidecar(row: dict) -> dict:
         'session_id': row.get('id'),
         'title': row.get('title') or 'Recovered WebUI Session',
         'workspace': workspace_value if isinstance(workspace_value, str) else '',
-        'message_count': row.get('message_count') if isinstance(row.get('message_count'), int) else len(messages),
+        # state.db is canonical for the recovered message rows.  Its denormalized
+        # sessions.message_count is absent from older schemas (read above as 0)
+        # and can be stale after recovery, so never copy it into a sidecar whose
+        # `messages` array we just materialized.  Session.save() uses this field
+        # to decide whether an overwrite shrinks the transcript; it must match
+        # the payload exactly or the #1558 backup / empty-snapshot guards can be
+        # bypassed.
+        'message_count': len(messages if isinstance(messages, list) else []),
         'worktree_path': row.get('worktree_path') or None,
         'worktree_branch': row.get('worktree_branch') or None,
         'worktree_repo_root': row.get('worktree_repo_root') or None,

tests/test_save_count_without_full_parse.py

@@ -0,0 +1,386 @@
+"""A save must not read and parse the whole sidecar just to count its messages.
+
+The #1558 backup safeguard in `Session.save()` compares the on-disk message
+count with the incoming one, and writes `<sid>.json.bak` only when the save
+would SHRINK the array. Sound. But it obtained the on-disk count by
+`json.loads(self.path.read_text())` -- the entire file -- on EVERY save,
+including the 99.9% of saves that grow the conversation and never back
+anything up. `session_recovery._msg_count()` did the same on every boot, for
+every live file that has a `.bak`.
+
+Measured in production on 2026-09-15, in the running container:
+
+    sessions/fa3bca34a0c6.json   203,439,398 bytes / 266,940 messages
+      read_text     17,453 ms
+      json.loads     2,924 ms
+      TOTAL to obtain one integer: 20,377 ms   <- paid on every save
+
+That was the 39,357 ms `/api/session` request stuck at stage
+`t6_after_json_write`, after the read-side re-parse (#4633 recurrence) had
+already been fixed. The cost scales linearly with the file, and the file grows
+for as long as the conversation does.
+
+What the check needs is ONE integer -- how many messages the file on disk
+holds -- and `save()` already writes that integer into the metadata prefix,
+before `messages`, as `message_count`, exactly so readers can have it without
+parsing the body (that is how `load_metadata_only()` and the sidebar freshness
+check work). So `save()` reads that, through the same bounded 64 KiB prefix
+reader, and falls back to the full read + parse whenever the prefix does not
+carry a usable count. The `.bak` body is the one thing that still needs the
+full text, and a shrink is the one time it is needed.
+
+The count is read from the FILE, not cached in the object, and that is the
+point. An earlier revision of this PR trusted an in-memory (inode, size,
+mtime_ns) signature: "this object wrote the file and stat says nothing changed,
+so its length is already known". Review (nesquena-hermes, 2026-09-21) showed
+that is not a content identity -- a same-length in-place rewrite inside one
+mtime tick keeps all three fields, because ext4 stamps mtime from a coarse
+clock, so a stale cached count reads a real shrink as a growth and skips the
+#1558 backup. That is the test at the bottom of this file. A count read from
+the bytes cannot be fooled that way: any rewrite of the bytes rewrites it.
+
+Fail-open is the contract these tests pin -- whenever the prefix has no usable
+count (a legacy pre-#5854 sidecar, a file with no top-level `messages` key, a
+corrupt or truncated prefix, metadata that overflows the budget), the behaviour
+must be exactly today's full parse, never "assume no shrink".
+
+`session_recovery._msg_count()` pays the same cost at boot and is left alone
+on purpose: its "torn file -> -1" contract is what makes recovery restore a
+`.bak` over a truncated live file, and no bounded read can prove a file is
+not truncated.
+"""
+import json
+import os
+import pathlib
+from collections import OrderedDict
+
+import pytest
+
+import api.models as M
+
+
+@pytest.fixture
+def session_store(tmp_path, monkeypatch):
+    sdir = tmp_path / "sessions"
+    sdir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(M, "SESSION_DIR", sdir)
+    monkeypatch.setattr(M, "SESSIONS", OrderedDict())
+    return sdir
+
+
+def _msgs(n):
+    return [{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(n)]
+
+
+def _make(session_store, sid, n):
+    s = M.Session(session_id=sid, title="T", workspace=str(session_store.parent),
+                  model="glm", messages=_msgs(n))
+    s.save()
+    return s
+
+
+def _spy_full_reads(monkeypatch, target):
+    """Count `Path.read_text()` calls on ONE path.
+
+    `read_text` is the whole-file read; the metadata-prefix reader uses `open()`
+    with a byte budget and never goes through it. So this counter is exactly
+    "how many times was this file read in full", which is the cost being removed.
+    """
+    calls = {"n": 0}
+    real = pathlib.Path.read_text
+    target = pathlib.Path(target)
+
+    def counting(self, *a, **k):
+        if pathlib.Path(self) == target:
+            calls["n"] += 1
+        return real(self, *a, **k)
+
+    monkeypatch.setattr(type(pathlib.Path()), "read_text", counting)
+    return calls
+
+
+# ── save(): the grow path must be free of full reads ────────────────────────
+
+def test_grow_save_does_not_read_the_existing_file_in_full(session_store, monkeypatch):
+    """The common case. Appending messages must not read the previous file
+    in full -- this object wrote that file and it is untouched, so the on-disk
+    count is already known."""
+    s = _make(session_store, "g1", 5)
+    calls = _spy_full_reads(monkeypatch, s.path)
+
+    s.messages = _msgs(7)
+    s.save()
+
+    assert calls["n"] == 0, f"grow-save read the existing sidecar in full {calls['n']}x"
+    assert not s.path.with_suffix(".json.bak").exists(), "a grow-save must not produce a backup"
+    assert len(M.Session.load("g1").messages) == 7
+
+
+def test_same_size_save_does_not_read_the_existing_file_in_full(session_store, monkeypatch):
+    """Metadata-only saves (title, flags, stream state) keep the array as-is.
+    They are the most frequent save of all and must be just as cheap."""
+    s = _make(session_store, "g2", 5)
+    calls = _spy_full_reads(monkeypatch, s.path)
+
+    s.title = "renamed"
+    s.save()
+
+    assert calls["n"] == 0
+    assert not s.path.with_suffix(".json.bak").exists()
+
+
+# ── save(): the safeguards must be untouched ────────────────────────────────
+
+def test_shrink_save_still_writes_bak_with_the_pre_shrink_content(session_store, monkeypatch):
+    """#1558 must keep working. A shrink still produces a `.bak` holding the
+    pre-shrink array -- and reads the file in full exactly once, for that body.
+    Not twice: knowing the count must not be followed by a redundant parse."""
+    s = _make(session_store, "s1", 5)
+    calls = _spy_full_reads(monkeypatch, s.path)
+
+    s.messages = _msgs(3)
+    s.save()
+
+    reads_by_save = calls["n"]  # taken before load() below, which reads the file too
+    bak = s.path.with_suffix(".json.bak")
+    assert bak.exists(), "a shrinking save must leave a recoverable backup"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5
+    assert len(M.Session.load("s1").messages) == 3
+    assert reads_by_save == 1, f"shrink-save read the sidecar in full {reads_by_save}x; the .bak body needs exactly one"
+
+
+def test_empty_active_snapshot_is_still_refused(session_store, monkeypatch):
+    """The other guard in the same block: an empty array with a live stream or
+    pending prompt must NOT overwrite a populated file (the #1558 data-loss
+    shape). This object was neither loaded nor saved by this process, so it has
+    no signature to trust and refusing may cost today's single full read --
+    never more, and it must still refuse."""
+    _make(session_store, "r1", 5)
+    empty = M.Session(session_id="r1", title="T", workspace=str(session_store.parent),
+                      model="glm", messages=[], active_stream_id="a" * 32,
+                      pending_user_message="still typing")
+    calls = _spy_full_reads(monkeypatch, empty.path)
+
+    empty.save()
+
+    reads_by_save = calls["n"]  # taken before load() below, which reads the file too
+    assert len(M.Session.load("r1").messages) == 5, "the populated file must survive"
+    assert reads_by_save <= 1, f"refusing read the file {reads_by_save}x; an unknown identity costs at most one"
+
+
+# ── save(): fail-open whenever the prefix carries no usable count ───────────
+
+def _write_legacy_without_count(session_store, sid, n):
+    """A pre-#5854 sidecar: no persisted `message_count` at all."""
+    doc = {"session_id": sid, "title": "Legacy", "workspace": str(session_store.parent),
+           "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+           "messages": _msgs(n)}
+    p = session_store / f"{sid}.json"
+    p.write_text(json.dumps(doc, indent=2), encoding="utf-8")
+    return p
+
+
+def test_legacy_sidecar_without_message_count_still_backs_up_on_shrink(session_store):
+    """The prefix has no count to give, so this must fall back to the full
+    parse and keep the safeguard. Silently treating "no count" as "no shrink"
+    would reopen #1558."""
+    _write_legacy_without_count(session_store, "l1", 5)
+    s = M.Session(session_id="l1", title="Legacy", workspace=str(session_store.parent),
+                  model="glm", messages=_msgs(3))
+
+    s.save()
+
+    bak = s.path.with_suffix(".json.bak")
+    assert bak.exists(), "fail-open: a legacy file must still be backed up on shrink"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5
+
+
+def test_file_changed_on_disk_since_last_save_falls_back_and_still_backs_up(session_store):
+    """An externally grown file. This process saved 3 messages; something else
+    then rewrote the file with 6 (an external appender, another process, a
+    restore). The next save with 4 is a real shrink relative to DISK, and the
+    count that governs it is the one the file now carries (6), not the 3 this
+    process remembers: the `.bak` must hold the 6-message array."""
+    s = _make(session_store, "x1", 3)
+    external = {"session_id": "x1", "title": "T", "workspace": str(session_store.parent),
+                "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+                "message_count": 6, "messages": _msgs(6)}
+    s.path.write_text(json.dumps(external, indent=2), encoding="utf-8")
+
+    s.messages = _msgs(4)
+    s.save()
+
+    bak = s.path.with_suffix(".json.bak")
+    assert bak.exists(), "an externally grown file must still be backed up on shrink"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6
+
+
+def test_file_changed_on_disk_without_a_count_still_backs_up_on_shrink(session_store):
+    """The same external rewrite, but the replacement carries no
+    `message_count` (an older writer, a hand-edited file). No prefix count ->
+    the full parse runs and the 6-message array is still backed up."""
+    s = _make(session_store, "x2", 3)
+    external = {"session_id": "x2", "title": "T", "workspace": str(session_store.parent),
+                "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+                "messages": _msgs(6)}
+    s.path.write_text(json.dumps(external, indent=2), encoding="utf-8")
+
+    s.messages = _msgs(4)
+    s.save()
+
+    bak = s.path.with_suffix(".json.bak")
+    assert bak.exists(), "fail-open: a count-less external rewrite must still be backed up"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6
+
+
+def test_unmarked_stale_count_from_a_foreign_writer_still_backs_up_on_shrink(session_store):
+    """A persisted count with no `_mc_v` marker is not trusted.
+
+    The exact shape review (nesquena-hermes, 2026-09-22) pinned: a sidecar
+    materialized by an OLDER writer can carry a count stale against the
+    messages array beside it (the recovery writer used to copy a denormalized
+    state.db count over real rows). Here the foreign file says 2 but really
+    holds 6; a save of 4 must not read that 2 as "the save grows the file" and
+    skip the #1558 backup. The marker gate makes the prefix reader refuse the
+    unmarked count, the full parse sees the real array, and the 6 messages
+    land in the `.bak`. Remove the gate in `_prefix_message_count` and this
+    test fails with no backup written.
+    """
+    s = _make(session_store, "x3", 3)
+    foreign = {"session_id": "x3", "title": "T", "workspace": str(session_store.parent),
+               "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+               "message_count": 2, "messages": _msgs(6)}  # stale count, NO _mc_v
+    s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8")
+
+    s.messages = _msgs(4)
+    s.save()
+
+    bak = s.path.with_suffix(".json.bak")
+    assert bak.exists(), "an unmarked stale count must not hide the shrink"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6
+    assert len(M.Session.load("x3").messages) == 4
+
+
+def test_wrong_marker_version_still_backs_up_on_shrink(session_store):
+    """A count marked by a DIFFERENT writer contract is equally untrusted."""
+    s = _make(session_store, "x4", 3)
+    foreign = {"session_id": "x4", "title": "T", "workspace": str(session_store.parent),
+               "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+               "_mc_v": M._MESSAGE_COUNT_MARKER + 1,
+               "message_count": 2, "messages": _msgs(6)}
+    s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8")
+
+    s.messages = _msgs(4)
+    s.save()
+
+    bak = s.path.with_suffix(".json.bak")
+    assert bak.exists(), "a count from another writer version must not hide the shrink"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6
+
+
+def test_first_save_remarks_the_file_and_the_fast_path_resumes(session_store, monkeypatch):
+    """The gate costs one full parse, once. After a save by the current writer
+    the file carries `_mc_v`, and the cheap prefix path is back in service."""
+    s = _make(session_store, "x5", 3)
+    foreign = {"session_id": "x5", "title": "T", "workspace": str(session_store.parent),
+               "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+               "message_count": 2, "messages": _msgs(6)}  # unmarked foreign file
+    s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8")
+
+    s.messages = _msgs(6)
+    s.save()  # equal-count save; unmarked prefix count ignored, full parse, no shrink
+
+    assert json.loads(s.path.read_text(encoding="utf-8"))["_mc_v"] == M._MESSAGE_COUNT_MARKER
+
+    calls = _spy_full_reads(monkeypatch, s.path)
+    s.messages = _msgs(8)
+    s.save()
+    assert calls["n"] == 0, "a marked file must take the bounded-prefix path again"
+    assert not s.path.with_suffix(".json.bak").exists()
+
+
+def _tool_partial(ts=123):
+    """The exact shape the #2592 collapse recognises (copied from its test)."""
+    return {"role": "assistant", "content": "", "_partial": True, "timestamp": ts,
+            "reasoning": "same reasoning",
+            "_partial_tool_calls": [{"name": "execute_code",
+                                     "args": {"code": "raise RuntimeError('boom')"},
+                                     "done": True, "is_error": True, "duration": 3.87}]}
+
+
+def test_collapse_self_heal_on_load_still_backs_up_the_pre_collapse_array(session_store):
+    """#2592: load() de-duplicates adjacent partials and immediately saves the
+    shorter transcript, and that save must produce a `.bak` because it shrinks
+    the array on purpose. The count that decides it comes from the file's own
+    prefix, which the writer stamped with the length it wrote (5) -- the
+    ON-DISK length, not the post-collapse length of the object (3). Read the
+    object's own length instead and this backup silently disappears, so the
+    fixture carries the count a real save() would have written."""
+    doc = {"session_id": "h1", "title": "T", "workspace": str(session_store.parent),
+           "model": "glm", "created_at": 1.0, "updated_at": 2.0,
+           "_mc_v": M._MESSAGE_COUNT_MARKER,
+           "message_count": 5,
+           "messages": [{"role": "user", "content": "run this"},
+                        _tool_partial(), _tool_partial(), _tool_partial(),
+                        {"role": "assistant", "content": "**Task cancelled.**", "_error": True}]}
+    (session_store / "h1.json").write_text(json.dumps(doc), encoding="utf-8")
+
+    loaded = M.Session.load("h1")
+
+    assert sum(1 for m in loaded.messages if m.get("_partial")) == 1, "the collapse must have fired"
+    assert len(loaded.messages) == 3
+    bak = (session_store / "h1.json").with_suffix(".json.bak")
+    assert bak.exists(), "the self-heal shrink must leave the pre-collapse array recoverable"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5
+
+
+# ── the collision a stat identity cannot see (review, 2026-09-21) ───────────
+
+def test_same_length_in_place_rewrite_inside_one_mtime_tick_still_backs_up(session_store):
+    """A same-length, same-mtime, same-inode rewrite must not hide a shrink.
+
+    The removed revision of this PR trusted an in-memory (inode, size,
+    mtime_ns) identity: "this object wrote the file, stat says nothing changed,
+    so the on-disk length is the one I remember". A stat tuple is not a content
+    identity. Another writer replaces the sidecar IN PLACE with a longer
+    transcript that occupies the same number of bytes, inside one mtime tick,
+    so all three fields still match -- and the next shrinking save then reads
+    its stale count as "growing", overwrites the longer transcript and writes
+    no `.bak`. That is the #1558 data-loss shape the safeguard exists to
+    prevent, reachable on the branch, which is why the identity cache is gone.
+
+    The count now comes from the file's own metadata prefix, which any rewrite
+    of the bytes has to carry, so the shrink is seen for what it is.
+    """
+    p = session_store / "i1.json"
+    base = {"session_id": "i1", "title": "T", "workspace": str(session_store.parent),
+            "model": "glm", "created_at": 1.0, "updated_at": 2.0}
+
+    def _render(n):
+        return json.dumps({**base, "_mc_v": M._MESSAGE_COUNT_MARKER,
+                           "message_count": n, "messages": _msgs(n)},
+                          indent=2, ensure_ascii=False)
+
+    width = max(len(_render(2)), len(_render(5)))
+    short, grown = _render(2).ljust(width), _render(5).ljust(width)
+    assert len(short) == len(grown) == width
+    p.write_text(short, encoding="utf-8")
+
+    loaded = M.Session.load("i1")          # the only version this object knows
+    before = p.stat()
+
+    p.write_text(grown, encoding="utf-8")  # same inode, same byte length...
+    os.utime(p, ns=(before.st_atime_ns, before.st_mtime_ns))
+    after = p.stat()
+    assert (after.st_ino, after.st_size, after.st_mtime_ns) == \
+        (before.st_ino, before.st_size, before.st_mtime_ns), \
+        "premise: the rewrite must be invisible to (inode, size, mtime_ns)"
+    assert len(loaded.messages) == 2, "premise: this object remembers 2 messages"
+
+    loaded.messages = _msgs(3)
+    loaded.save()
+
+    bak = p.with_suffix(".json.bak")
+    assert bak.exists(), "the 5-message transcript must be recoverable"
+    assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5
+    assert len(M.Session.load("i1").messages) == 3

tests/test_session_db_sidecar_reconciliation.py

@@ -45,9 +45,54 @@ def test_recover_missing_sidecars_from_state_db_materializes_webui_row(tmp_path)
     assert data["parent_session_id"] == "parent-1"
     assert data["source_tag"] == "webui"
     assert data["session_source"] == "webui"
+    assert data["message_count"] == 2
     assert [m["content"] for m in data["messages"]] == ["message 1", "message 2"]
 
 
+def test_recovered_sidecar_derives_message_count_from_materialized_rows(tmp_path, monkeypatch):
+    """Recovery must not copy a stale state.db summary into a live sidecar.
+
+    `Session.save()` relies on sidecar `message_count` for both #1558 guards.
+    A recovery row whose denormalized DB count says zero but whose canonical
+    message rows contain five messages must therefore refuse an empty active
+    snapshot and preserve a five-message backup on a later shrink.
+    """
+    import api.models as models
+
+    sid = _make_state_db(tmp_path / "state.db", sid="recovered_stale_count", messages=5)
+    with sqlite3.connect(tmp_path / "state.db") as conn:
+        conn.execute("UPDATE sessions SET message_count = 0 WHERE id = ?", (sid,))
+
+    assert recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")["materialized"] == 1
+    sidecar = tmp_path / f"{sid}.json"
+    assert json.loads(sidecar.read_text(encoding="utf-8"))["message_count"] == 5
+
+    monkeypatch.setattr(models, "SESSION_DIR", tmp_path)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", tmp_path / "_index.json")
+    active_empty = models.Session(
+        session_id=sid,
+        title="Recovered from DB",
+        workspace=str(tmp_path),
+        model="openai/gpt-5",
+        messages=[],
+        active_stream_id="a" * 32,
+        pending_user_message="still typing",
+    )
+    active_empty.save()
+    assert len(json.loads(sidecar.read_text(encoding="utf-8"))["messages"]) == 5
+
+    shrinking = models.Session(
+        session_id=sid,
+        title="Recovered from DB",
+        workspace=str(tmp_path),
+        model="openai/gpt-5",
+        messages=[{"role": "user", "content": f"replacement {i}"} for i in range(3)],
+    )
+    shrinking.save()
+    backup = sidecar.with_suffix(".json.bak")
+    assert len(json.loads(backup.read_text(encoding="utf-8"))["messages"]) == 5
+
+
 def test_recover_missing_sidecars_from_state_db_skips_deleted_webui_tombstone(tmp_path, monkeypatch):
     import api.models as _m
     monkeypatch.setattr(_m, "SESSION_DIR", tmp_path)

tests/test_webui_gateway_chat_backend.py

@@ -25,6 +25,7 @@ from api.gateway_chat import (
     webui_chat_backend_mode,
     webui_gateway_chat_enabled,
 )
+from api.turn_journal import derive_turn_journal_states, read_turn_journal
 
 
 def test_gateway_chat_backend_is_default_off_for_truthy_values():
@@ -397,6 +398,73 @@ def test_gateway_chat_worker_translates_sse_and_persists_session(tmp_path, monke
     assert all(len(item) == 3 and item[2] for item in events)
 
 
+def test_gateway_chat_worker_records_turn_journal_completion(tmp_path, monkeypatch):
+    """#6366 re-gate: a successful Gateway run must record durable
+    same-stream completion evidence in the crash-safe turn journal.
+
+    The stale-cancel recovery predicate derives its completion evidence
+    from the turn journal, so a Gateway run that persisted its final
+    answer but lost the run journal's terminal write would otherwise
+    have no completion evidence at all — and recovery would re-append a
+    duplicate recovered row after the valid final answer.
+    """
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    monkeypatch.setattr(models, "SESSIONS", OrderedDict())
+
+    class FakeResponse:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def __iter__(self):
+            yield b'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n'
+            yield b"data: [DONE]\n\n"
+
+    monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local")
+    monkeypatch.setenv("HERMES_WEBUI_GATEWAY_API_KEY", "secret-token")
+    monkeypatch.setattr(
+        gateway_chat.urllib.request,
+        "urlopen",
+        lambda req, timeout=0: FakeResponse(),
+    )
+
+    s = new_session()
+    stream_id = "stream-gateway-turn-journal-completed"
+    s.active_stream_id = stream_id
+    s.pending_user_message = "Say hello"
+    s.pending_attachments = []
+    s.pending_started_at = 456
+    s.save()
+    channel = create_stream_channel()
+    STREAMS[stream_id] = channel
+
+    gateway_chat._run_gateway_chat_streaming(
+        s.session_id,
+        "Say hello",
+        "test-model",
+        str(tmp_path),
+        stream_id,
+        [],
+    )
+
+    journal = read_turn_journal(s.session_id, session_dir=session_dir)
+    states, _ = derive_turn_journal_states(journal.get("events") or [])
+    stream_events = [
+        event
+        for event in states.values()
+        if str(event.get("stream_id") or "") == stream_id
+    ]
+    assert stream_events, "expected turn journal events for the gateway stream"
+    assert any(
+        event.get("event") == "completed" for event in stream_events
+    ), "the gateway success writeback must record a completed turn journal event"
+
+
 def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp_path, monkeypatch):
     """Gateway terminal errors must survive an empty assistant stream."""
     from unittest.mock import MagicMock