fix(routes): gate state.db row backstop on absence of truncation_boundary prefix

Reviewer regression (reproduced): _STATE_DB_DISPLAY_ROW_BACKSTOP (newest-50k SQL
cap) was applied UNCONDITIONALLY to the GET /api/session display path's state.db
read. But merge_session_messages_append_only needs the rows at/around the
session's truncation_boundary (the preserved prefix) to reconcile correctly. A
newest-N-only cap drops those boundary rows for a >N-row session and corrupts
the merge (silently losing the preserved prefix), and also prevents older-page
(msg_before) requests from ever reaching them. So a pathological-but-real >50k-
row compressed session got a corrupted merge / unreachable older history rather
than a clean bound.

Fix: gate the backstop on the SAME conditions
_state_db_since_timestamp_for_limited_display uses to decide a read is
boundary-free — not msg_before paging, and no truncation_watermark /
truncation_boundary. Those cases stay on the full (uncapped) read; the backstop
is now a defensive net for the common uncompressed initial-tail path only.
Extracted into _state_db_backstop_limit_for_display for direct test coverage.

The get_state_db_session_messages(limit=) signature + newest-rows subquery are
unchanged — only the display-path APPLICATION of the cap is gated.

Test: tests/test_state_db_backstop_boundary_gate.py — backstop applies with no
boundary prefix; SKIPPED when truncation_boundary / truncation_watermark set or
msg_before paging; empty-string markers treated as unset (matching the existing
helper). Existing backstop + reconciliation suites (53 tests) unchanged.

Model Used: builtin:zai-coding-plan/GLM-5.2 (ZCode agent).

api/routes.py

@@ -8452,6 +8452,31 @@ _LIMITED_TOOL_CONTENT_MAX_CHARS = 4096
 # already handles the common tail-load case. The full-history model-context
 # callers (reconciliation, new-turn context) do NOT use this cap.
 _STATE_DB_DISPLAY_ROW_BACKSTOP = 50000
+
+
+def _state_db_backstop_limit_for_display(session, msg_before) -> int | None:
+    """Return the row backstop to apply to the display path's state.db read, or
+    ``None`` for an uncapped (full-history) read.
+
+    The backstop is a defensive net against a pathological/huge state.db, NOT a
+    semantic window. It is applied ONLY on provably-safe reads where no
+    ``truncation_boundary`` prefix is required for the merge:
+    ``merge_session_messages_append_only`` needs the rows at/around the session's
+    ``truncation_boundary`` to reconcile correctly, and a newest-N-only SQL cap
+    would drop those boundary rows for a >N-row session and corrupt the merge
+    (silently losing the preserved prefix). So this mirrors the same conditions
+    ``_state_db_since_timestamp_for_limited_display`` uses to decide a read is
+    boundary-free: not ``msg_before`` paging, and no ``truncation_watermark`` /
+    ``truncation_boundary``. Extracted for direct test coverage.
+    """
+    has_boundary_prefix = (
+        msg_before is not None
+        or getattr(session, "truncation_watermark", None) not in (None, "")
+        or getattr(session, "truncation_boundary", None) not in (None, "")
+    )
+    return None if has_boundary_prefix else _STATE_DB_DISPLAY_ROW_BACKSTOP
+
+
 _LIMITED_TOOL_CONTENT_NOTICE = (
     "\n\n[Tool output truncated in paginated session response; "
     "load the full transcript to inspect the complete result.]"
@@ -12435,13 +12460,14 @@ def handle_get(handler, parsed) -> bool:
                 _state_db_reader_kwargs = {"profile": _session_profile}
                 if state_db_since_timestamp is not None:
                     _state_db_reader_kwargs["since_timestamp"] = state_db_since_timestamp
-                # Apply the display-path row backstop (see
-                # _STATE_DB_DISPLAY_ROW_BACKSTOP). This is a defensive cap, not a
-                # semantic window — it only bounds a pathological state.db. The
-                # since_timestamp optimization above already bounds the common
-                # uncompressed tail load; this covers the compressed-session bail
-                # case where that optimization returns None.
-                _state_db_reader_kwargs["limit"] = _STATE_DB_DISPLAY_ROW_BACKSTOP
+                # Apply the display-path row backstop ONLY on provably-safe
+                # reads where no truncation_boundary prefix is required for the
+                # merge — see _state_db_backstop_limit_for_display. Compressed
+                # sessions and msg_before paging need their full prefix rows for
+                # correct reconciliation, so those stay uncapped.
+                _backstop = _state_db_backstop_limit_for_display(s, msg_before)
+                if _backstop is not None:
+                    _state_db_reader_kwargs["limit"] = _backstop
                 state_db_messages = get_state_db_session_messages(
                     sid,
                     **_state_db_reader_kwargs,

tests/test_state_db_backstop_boundary_gate.py

@@ -0,0 +1,68 @@
+"""Tests: the display-path state.db row backstop is GATED on the absence of a
+truncation_boundary prefix.
+
+Regression (reviewer): ``_STATE_DB_DISPLAY_ROW_BACKSTOP`` was applied
+unconditionally to the ``GET /api/session`` display path's state.db read. But
+``merge_session_messages_append_only`` needs the rows at/around the session's
+``truncation_boundary`` (the preserved prefix) to reconcile correctly. A
+newest-N-only SQL cap drops those boundary rows for a >N-row session and
+corrupts the merge (silently losing the preserved prefix), and also prevents
+older-page (``msg_before``) requests from reaching them.
+
+The fix gates the backstop via ``_state_db_backstop_limit_for_display`` on the
+same conditions ``_state_db_since_timestamp_for_limited_display`` uses to decide
+a read is boundary-free: NOT ``msg_before`` paging, and NO
+``truncation_watermark`` / ``truncation_boundary``. Those cases stay on the full
+(uncapped) read. The helper is the unit under test (driving the full handler
+end-to-end is brittle across its many dependencies).
+"""
+from __future__ import annotations
+
+from api.routes import _STATE_DB_DISPLAY_ROW_BACKSTOP, _state_db_backstop_limit_for_display
+
+
+class _StubSession:
+    """Minimal object exposing the boundary attributes the gate checks."""
+    def __init__(self, **attrs):
+        self.truncation_watermark = attrs.get("truncation_watermark", None)
+        self.truncation_boundary = attrs.get("truncation_boundary", None)
+
+
+def test_backstop_applied_when_no_boundary_prefix():
+    """An uncompressed, initial-tail session (no truncation_watermark /
+    truncation_boundary, no msg_before) gets the defensive row backstop."""
+    assert _state_db_backstop_limit_for_display(_StubSession(), msg_before=None) == _STATE_DB_DISPLAY_ROW_BACKSTOP
+
+
+def test_backstop_skipped_when_truncation_boundary_set():
+    """Regression: a compressed session (truncation_boundary set) needs its
+    preserved-prefix rows for the merge, so the backstop must NOT cap the read."""
+    stub = _StubSession(truncation_boundary="some-boundary-marker")
+    assert _state_db_backstop_limit_for_display(stub, msg_before=None) is None
+
+
+def test_backstop_skipped_when_truncation_watermark_set():
+    """Same gate for truncation_watermark — the merge needs the prefix rows."""
+    stub = _StubSession(truncation_watermark=12345)
+    assert _state_db_backstop_limit_for_display(stub, msg_before=None) is None
+
+
+def test_backstop_skipped_when_msg_before_paging():
+    """Older-page (msg_before) requests need to reach the boundary rows, so the
+    backstop must not cap them either."""
+    stub = _StubSession()
+    assert _state_db_backstop_limit_for_display(stub, msg_before=500) is None
+
+
+def test_backstop_boundary_check_handles_empty_string_as_unset_or_set():
+    """The gate treats empty-string boundary markers as 'not set' (matches the
+    existing since_timestamp helper's `in (None, "")` check). A real marker
+    (non-empty) triggers the skip."""
+    # Empty string = treated as unset → backstop applies.
+    assert _state_db_backstop_limit_for_display(
+        _StubSession(truncation_boundary=""), msg_before=None
+    ) == _STATE_DB_DISPLAY_ROW_BACKSTOP
+    # Non-empty string = boundary present → backstop skipped.
+    assert _state_db_backstop_limit_for_display(
+        _StubSession(truncation_watermark="wm"), msg_before=None
+    ) is None