fix(sessions): gate the prefix count on a writer marker

A persisted message_count is only authoritative when the write that
produced the messages array also vouched for it. Sidecars written by an
older writer (recovery materialization copied the denormalized state.db
count over real rows) could carry a count stale against the array beside
it, and the prefix shortcut then reads a real shrink as a growth and
skips the #1558 backup.

save() now stamps _mc_v alongside message_count, and _prefix_message_count
refuses any count without the current marker: those files get the full
parse, exactly as on the legacy path. One save re-marks the file, so the
fast path resumes for steady state.

Regression covers the unmarked stale-count shape and the re-marking
transition; mutation-checked by removing the gate (the unmarked fixture
then skips the backup, as the review's reproduction predicted).

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('_')}
@@ -1562,8 +1580,11 @@ class Session:
                 #
                 # Every unknown falls through to the full read + parse below: a
                 # legacy (pre-#5854) sidecar whose count is not in the prefix, a
-                # corrupt or truncated prefix, a file with no top-level `messages`
-                # key at all, or metadata alone that overflows the budget.
+                # 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)
@@ -4274,6 +4295,9 @@ 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.
 
@@ -4290,11 +4314,20 @@ def _prefix_message_count(path):
     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 count: a legacy
-    pre-#5854 sidecar (scenes serialized before the count), 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.
+    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)
@@ -4308,6 +4341,8 @@ def _prefix_message_count(path):
         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'))
 
 

tests/test_save_count_without_full_parse.py

@@ -233,6 +233,72 @@ def test_file_changed_on_disk_without_a_count_still_backs_up_on_shrink(session_s
     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,
@@ -252,6 +318,7 @@ def test_collapse_self_heal_on_load_still_backs_up_the_pre_collapse_array(sessio
     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(),
@@ -290,7 +357,8 @@ def test_same_length_in_place_rewrite_inside_one_mtime_tick_still_backs_up(sessi
             "model": "glm", "created_at": 1.0, "updated_at": 2.0}
 
     def _render(n):
-        return json.dumps({**base, "message_count": n, "messages": _msgs(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)))