fix(models): add defensive row backstop to display-path state.db message read

Memory: get_state_db_session_messages had no SQL LIMIT — it full-scanned the
state.db messages table and fetchall()'d into a Python list. The display path
(GET /api/session) already has a careful since_timestamp optimization that
bounds the common uncompressed tail load, but it deliberately bails (returns
None) on compressed sessions (truncation_watermark/boundary) and msg_before
paging — in those cases the read used to full-scan with no cap. A
pathological/huge state.db could then materialize unbounded rows into memory.

Add an opt-in `limit` parameter to get_state_db_session_messages (applied as a
SQL LIMIT that keeps the NEWEST rows via a DESC subquery re-sorted ASC). It is
a BACKSTOP, not a semantic window: the display window counts visible rows
post-reconciliation, so a true window LIMIT here would corrupt the
sidecar/state.db merge (see _state_db_since_timestamp_for_limited_display,
which deliberately does NOT SQL-LIMIT raw rows for exactly that reason).

Thread a generous _STATE_DB_DISPLAY_ROW_BACKSTOP (50000) from the GET /api/session
display caller only. The full-history model-context callers (reconciliation at
models.py:3437, context reconstruction at :8576, new-turn context at
streaming.py:8443) do NOT pass a limit — their semantics require the full
history, unchanged.

Test: tests/test_state_db_read_backstop.py — no-limit returns full history
oldest-first (model-context contract unchanged); limit=N keeps the newest N
re-sorted oldest-first; limit > row count returns all; limit<=0 clamps to 1
(newest); invalid limit falls back to unbounded; limit composes with
since_timestamp. Existing state.db active-filter + reconciliation suites (54
tests) unchanged.

Contract Routing
- Task type: memory-bound fix (unbounded state.db row materialization on the
  display path's bail cases).
- Touched areas: state.db message reader (api/models.py) + GET /api/session
  display caller (api/routes.py). NOT the model-context callers.
- Relevant public docs: AGENTS.md, CONTRIBUTING.md, docs/CONTRACTS.md,
  docs/rfcs/webui-run-state-consistency-contract.md (Model context layer).
- State layer mutated: none (read-only). The model-context read path is
  UNCHANGED (no limit passed) — Invariant 1 (visible current turns enter model
  context) and the reconciliation contract are preserved. Only the display path
  gets the defensive cap, which does not affect what the agent receives.
- Scope boundaries: this is narrower than a naive "add LIMIT everywhere" —
  _state_db_since_timestamp_for_limited_display's docstring explains why a
  semantic window LIMIT would corrupt the merge; this cap is purely a
  pathological-input safety net above any legitimate session size.

Release note: GET /api/session's state.db read now has a defensive row backstop,
capping memory when the bounded since_timestamp optimization bails on
compressed/paged sessions.

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

api/models.py

@@ -7330,6 +7330,7 @@ def get_state_db_session_messages(
     profile=None,
     since_timestamp=None,
     include_inactive: bool = False,
+    limit=None,
 ) -> list:
     """Read messages for a Hermes session from state.db.
 
@@ -7346,6 +7347,14 @@ def get_state_db_session_messages(
     preserving the caller's normal merge/window logic.  Full-history callers must
     leave it unset.
 
+    ``limit`` is an optional defensive row cap (applied after ORDER BY as a SQL
+    LIMIT). It is a BACKSTOP against a pathological/huge state.db materializing
+    unbounded rows into a Python list, NOT a semantic window: the display path
+    counts visible rows post-reconciliation, so a true window LIMIT here would
+    corrupt the sidecar/state.db merge (see _state_db_since_timestamp_for_limited_display,
+    which deliberately does NOT SQL-LIMIT raw rows for that reason). Callers that
+    need the full history for model-context reconstruction leave this unset.
+
     When the messages table exposes an ``active`` column, inactive rows are
     compacted/archived history and are intentionally excluded by default. WebUI
     reconciliation feeds this reader straight into the next model context; pulling
@@ -7446,14 +7455,42 @@ def get_state_db_session_messages(
             active_clause = ""
             if 'active' in available and not include_inactive:
                 active_clause = " AND (active IS NULL OR active != 0)"
-            cur.execute(f"""
-                SELECT {', '.join(selected)}, session_id
-                FROM messages
-                WHERE session_id IN ({placeholders})
-                {since_clause}
-                {active_clause}
-                ORDER BY timestamp ASC, id ASC
-            """, params)
+            # Defensive row cap (backstop only — see docstring). Applied as a
+            # SQL LIMIT bound parameter (?) so the tail (newest) rows are
+            # retained and a pathological state.db can't materialize unbounded
+            # rows. None = unchanged full-history read for model-context callers.
+            limit_clause = ""
+            if limit is not None:
+                try:
+                    limit_int = max(1, int(limit))
+                except (TypeError, ValueError):
+                    limit_int = None
+                if limit_int is not None:
+                    # The query orders ASC (oldest first); to keep the NEWEST
+                    # rows under the cap, take a descending-ordered subquery and
+                    # re-sort ascending — a plain LIMIT would keep the oldest.
+                    limit_clause = " ORDER BY timestamp DESC, id DESC LIMIT ?"
+                    params.append(limit_int)
+            if limit_clause:
+                cur.execute(f"""
+                    SELECT * FROM (
+                        SELECT {', '.join(selected)}, session_id
+                        FROM messages
+                        WHERE session_id IN ({placeholders})
+                        {since_clause}
+                        {active_clause}
+                        {limit_clause}
+                    ) ORDER BY timestamp ASC, id ASC
+                """, params)
+            else:
+                cur.execute(f"""
+                    SELECT {', '.join(selected)}, session_id
+                    FROM messages
+                    WHERE session_id IN ({placeholders})
+                    {since_clause}
+                    {active_clause}
+                    ORDER BY timestamp ASC, id ASC
+                """, params)
             msgs = []
             for row in cur.fetchall():
                 msg = {

api/routes.py

@@ -8441,6 +8441,17 @@ def _message_window_for_display(messages, msg_limit=None, msg_before=None, expan
 
 
 _LIMITED_TOOL_CONTENT_MAX_CHARS = 4096
+# Defensive row backstop for the GET /api/session display path's state.db read.
+# This is NOT a semantic window (the display window counts visible rows
+# post-reconciliation via _message_window_for_display); it is a safety net so a
+# pathological/huge state.db cannot materialize unbounded rows into memory on the
+# display path. Legitimate sessions stay far below this; the compressed-session
+# case where _state_db_since_timestamp_for_limited_display bails (and would
+# otherwise full-scan) is the main beneficiary. Generous on purpose: no real
+# conversation approaches it, and the existing since_timestamp optimization
+# 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
 _LIMITED_TOOL_CONTENT_NOTICE = (
     "\n\n[Tool output truncated in paginated session response; "
     "load the full transcript to inspect the complete result.]"
@@ -12424,6 +12435,13 @@ 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
                 state_db_messages = get_state_db_session_messages(
                     sid,
                     **_state_db_reader_kwargs,

tests/test_state_db_read_backstop.py

@@ -0,0 +1,119 @@
+"""Tests: ``get_state_db_session_messages`` defensive ``limit`` backstop.
+
+The display path (``GET /api/session``) reads state.db rows that feed a
+reconciliation merge then a visible-row window. The existing
+``since_timestamp`` optimization bounds the common uncompressed tail load but
+deliberately bails on compressed sessions (``truncation_watermark`` /
+``truncation_boundary``) and ``msg_before`` paging — in those cases the read
+used to full-scan with no SQL ``LIMIT``. A pathological/huge state.db could
+then materialize unbounded rows into memory.
+
+``get_state_db_session_messages`` now accepts an opt-in ``limit`` (applied as a
+SQL ``LIMIT`` that keeps the NEWEST rows). It is a defensive backstop, NOT a
+semantic window — full-history model-context callers leave it unset. These
+tests verify the newest-row retention and that the default (no limit) is
+unchanged.
+"""
+import sqlite3
+
+import api.models as models
+from api.models import get_state_db_session_messages
+
+
+def _make_state_db(path, n_rows=100):
+    conn = sqlite3.connect(str(path))
+    conn.execute(
+        "CREATE TABLE sessions (id TEXT PRIMARY KEY)"
+    )
+    conn.execute(
+        """
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            role TEXT,
+            content TEXT,
+            timestamp REAL,
+            active INTEGER DEFAULT 1
+        )
+        """
+    )
+    conn.execute("INSERT INTO sessions (id) VALUES ('s1')")
+    conn.executemany(
+        "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
+        [("s1", "user" if i % 2 == 0 else "assistant", f"msg{i}", float(i)) for i in range(n_rows)],
+    )
+    conn.commit()
+    conn.close()
+
+
+def test_no_limit_returns_full_history_oldest_first(tmp_path, monkeypatch):
+    """Default (no limit) — the model-context caller contract — returns every
+    row, oldest-first. Unchanged by this fix."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, n_rows=100)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    msgs = get_state_db_session_messages("s1")
+    assert len(msgs) == 100
+    assert msgs[0]["content"] == "msg0"
+    assert msgs[-1]["content"] == "msg99"
+
+
+def test_limit_keeps_newest_rows_resorted_oldest_first(tmp_path, monkeypatch):
+    """limit=N keeps the NEWEST N rows (the query orders ASC for oldest-first,
+    so the cap takes a DESC subquery then re-sorts), so a tail window is
+    retained rather than the oldest rows."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, n_rows=100)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    msgs = get_state_db_session_messages("s1", limit=10)
+    assert len(msgs) == 10
+    # Newest 10 retained (msg90..msg99), re-sorted oldest-first.
+    assert [m["content"] for m in msgs] == [f"msg{i}" for i in range(90, 100)]
+
+
+def test_limit_above_row_count_returns_all(tmp_path, monkeypatch):
+    """A limit larger than the row count returns everything (no truncation)."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, n_rows=50)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    msgs = get_state_db_session_messages("s1", limit=10000)
+    assert len(msgs) == 50
+
+
+def test_limit_zero_or_negative_clamps_to_one_newest(tmp_path, monkeypatch):
+    """A non-positive limit clamps to 1 and returns the single newest row."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, n_rows=20)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    for bad in (0, -5):
+        msgs = get_state_db_session_messages("s1", limit=bad)
+        assert len(msgs) == 1
+        assert msgs[0]["content"] == "msg19"  # newest
+
+
+def test_limit_invalid_falls_back_to_unbounded(tmp_path, monkeypatch):
+    """A non-numeric limit is ignored (full read) rather than raising."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, n_rows=30)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    msgs = get_state_db_session_messages("s1", limit="not-a-number")
+    assert len(msgs) == 30
+
+
+def test_limit_composes_with_since_timestamp(tmp_path, monkeypatch):
+    """limit and since_timestamp compose: the floor filters rows, then the cap
+    keeps the newest of the remainder."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, n_rows=100)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db)
+
+    # since_timestamp=50.0 → rows with timestamp >= 50 (msg50..msg99 = 50 rows),
+    # then limit=10 keeps the newest 10 (msg90..msg99).
+    msgs = get_state_db_session_messages("s1", since_timestamp=50.0, limit=10)
+    assert len(msgs) == 10
+    assert [m["content"] for m in msgs] == [f"msg{i}" for i in range(90, 100)]