diff --git a/api/models.py b/api/models.py
index 0322a97b..6bf1ee0a 100644
--- a/api/models.py
+++ b/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
@@ -4239,6 +4295,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).
 
diff --git a/api/session_recovery.py b/api/session_recovery.py
index d2d46cb2..d9b09479 100644
--- a/api/session_recovery.py
+++ b/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,
diff --git a/tests/test_save_count_without_full_parse.py b/tests/test_save_count_without_full_parse.py
new file mode 100644
index 00000000..543ace41
--- /dev/null
+++ b/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
diff --git a/tests/test_session_db_sidecar_reconciliation.py b/tests/test_session_db_sidecar_reconciliation.py
index 8b00892b..ba27e574 100644
--- a/tests/test_session_db_sidecar_reconciliation.py
+++ b/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)