fix(regeneration): bounded tail read only when skipped state.db prefix is provably identical; compression-anchor coverage (#6826 r3)

api/session_ops.py

@@ -254,6 +254,70 @@ def _sidecar_regeneration_read_floor(session):
     return min(timestamps[-_REGENERATION_SIDECAR_ANCHOR_BUDGET:])
 
 
+def _bounded_tail_read_is_safe(session, read_floor) -> bool:
+    """Prove the skipped state.db prefix is represented identically in the
+    sidecar; otherwise the caller MUST fall back to the full read (#6826 r3).
+
+    The bounded ``since_timestamp`` read silently omits every state.db row
+    older than the floor. That is only safe when the skipped prefix is
+    provably identical in the already-loaded sidecar — same count AND same
+    ordered visible identity — and when a compression anchor (if any) is
+    covered by the floor. Any mismatch, missing database, or uncertainty
+    returns False (full-read fallback), so the #6611 regeneration authority
+    never operates on an unreconciled view.
+    """
+    sid = getattr(session, "session_id", None)
+    if not sid:
+        return False
+    profile = getattr(session, "profile", None)
+    from api.models import (
+        _session_message_visible_key,
+        get_state_db_session_message_keys_before_timestamp,
+        get_state_db_session_message_prefix_summary,
+    )
+
+    prefix = get_state_db_session_message_prefix_summary(sid, read_floor, profile=profile)
+    if prefix is None:
+        return False  # cannot prove the prefix → full read
+    # Compression-anchor coverage: auto-compressed sessions keep context rows
+    # around the anchor; if the anchor predates the floor the bounded read can
+    # drop compacted-tail context rows (display may still match). The anchor
+    # is a dict {"role","text","ts","attachments"} — compare its timestamp.
+    anchor = getattr(session, "compression_anchor_message_key", None)
+    if isinstance(anchor, dict):
+        try:
+            anchor_ts = float(anchor.get("ts"))
+        except (TypeError, ValueError):
+            anchor_ts = None
+        if anchor_ts is None or anchor_ts < read_floor:
+            return False
+    if prefix.get("count") == 0 and prefix.get("null_timestamp_count") == 0:
+        # Empty skipped prefix: the bounded read already covers every row.
+        return True
+    # Non-empty skipped prefix: prove identical ordered visible identity.
+    db_keys = get_state_db_session_message_keys_before_timestamp(
+        sid, read_floor, profile=profile
+    )
+    if db_keys is None:
+        return False
+    sidecar_keys = []
+    for message in getattr(session, "messages", None) or []:
+        if not isinstance(message, dict):
+            continue
+        try:
+            ts = float(message.get("timestamp"))
+        except (TypeError, ValueError):
+            ts = None
+        if ts is not None and ts < read_floor:
+            key = _session_message_visible_key(message)
+            if key is None:
+                return False
+            sidecar_keys.append(key)
+    if list(db_keys) != sidecar_keys:
+        return False  # mismatch → full read
+    return True
+
+
 def regeneration_state(session, *, use_sidecar=False):
     """Read one immutable state.db snapshot and reconcile both transcript views.
 
@@ -263,6 +327,11 @@ def regeneration_state(session, *, use_sidecar=False):
     through :func:`reconciled_state_db_messages_for_session`, so the #6611
     reconciliation authority (recovered display/context pair survives local
     and gateway apply) is preserved on the fast path.
+
+    The bounded tail is only trusted when
+    :func:`_bounded_tail_read_is_safe` proves the skipped state.db prefix is
+    identical in the sidecar (count + ordered visible identity + compression
+    anchor coverage); otherwise the read falls back to the full transcript.
     """
     from api.models import (
         get_state_db_session_messages,
@@ -272,7 +341,7 @@ def regeneration_state(session, *, use_sidecar=False):
     state_reader_kwargs = {}
     if use_sidecar:
         read_floor = _sidecar_regeneration_read_floor(session)
-        if read_floor is not None:
+        if read_floor is not None and _bounded_tail_read_is_safe(session, read_floor):
             state_reader_kwargs["since_timestamp"] = read_floor
     state_messages = get_state_db_session_messages(
         getattr(session, "session_id", None),

tests/test_issue6611_regeneration_authority.py

@@ -513,6 +513,24 @@ def test_regeneration_sidecar_path_avoids_full_transcript_refetch(monkeypatch):
         fake_get_state_db_session_messages,
     )
 
+    # The bounded guard must prove the skipped prefix (rows < floor) is
+    # identical in the sidecar: the fake state.db carries the same 300 prefix
+    # rows, so simulate the prefix summary + keys from the sidecar itself.
+    from api.models import _session_message_visible_key
+
+    floor = min(row["timestamp"] for row in session.messages[-200:])
+    prefix_rows = [r for r in session.messages if (r.get("timestamp") or 0) < floor]
+    monkeypatch.setattr(
+        models_api,
+        "get_state_db_session_message_prefix_summary",
+        lambda *_a, **_k: {"count": len(prefix_rows), "null_timestamp_count": 0},
+    )
+    monkeypatch.setattr(
+        models_api,
+        "get_state_db_session_message_keys_before_timestamp",
+        lambda *_a, **_k: [_session_message_visible_key(r) for r in prefix_rows],
+    )
+
     full_rows, full_context = regeneration_state(session)
     assert calls[-1].get("since_timestamp") is None  # full read
 
@@ -526,3 +544,119 @@ def test_regeneration_sidecar_path_avoids_full_transcript_refetch(monkeypatch):
     assert fast_rows == full_rows
     assert fast_context == full_context
     assert fast_rows[-1]["content"] == "gateway applied turn"
+
+
+# ── #6826 r3: bounded tail-read guard ────────────────────────────────────────
+
+def _sidecar_session(rows=3, *, anchor_key=None, anchor_ts=None):
+    """Session whose sidecar carries `rows` timestamped display messages."""
+    messages = [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}", "timestamp": 100.0 + i * 100.0}
+        for i in range(rows)
+    ]
+    s = Session(
+        session_id="sidecar-guard-6826",
+        messages=messages,
+        context_messages=[dict(m) for m in messages],
+    )
+    if anchor_key:
+        s.compression_anchor_message_key = anchor_key
+        if anchor_ts is not None:
+            s._anchor_ts = anchor_ts
+    return s
+
+
+def test_bounded_guard_empty_prefix_allows_tail_read(monkeypatch):
+    from api import session_ops
+
+    session = _sidecar_session(3)
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_prefix_summary",
+        lambda *_a, **_k: {"count": 0, "null_timestamp_count": 0},
+    )
+    seen = {}
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_keys_before_timestamp",
+        lambda *_a, **_k: [],
+    )
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_messages",
+        lambda *_a, **_k: _capture(seen, _k),
+    )
+    session_ops.regeneration_state(session, use_sidecar=True)
+    assert seen.get("since_timestamp") is not None, "empty skipped prefix must take the bounded tail read"
+
+
+def test_bounded_guard_unrepresented_prefix_row_falls_back(monkeypatch):
+    from api import session_ops
+
+    session = _sidecar_session(3)
+    # state.db has one row older than the floor that the sidecar does NOT carry
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_prefix_summary",
+        lambda *_a, **_k: {"count": 1, "null_timestamp_count": 0},
+    )
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_keys_before_timestamp",
+        lambda *_a, **_k: [("user", "recoverable-only-in-db")],
+    )
+    seen = {}
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_messages",
+        lambda *_a, **_k: _capture(seen, _k),
+    )
+    session_ops.regeneration_state(session, use_sidecar=True)
+    assert "since_timestamp" not in seen, "unrepresented prefix row must force the full read (no since_timestamp)"
+
+
+def test_bounded_guard_compression_anchor_below_floor_falls_back(monkeypatch):
+    from api import session_ops
+
+    session = _sidecar_session(3)
+    # anchor timestamp predates the floor → bounded read cannot cover it
+    session.compression_anchor_message_key = {"role": "user", "text": "m0", "ts": 100.0}
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_prefix_summary",
+        lambda *_a, **_k: {"count": 0, "null_timestamp_count": 0},
+    )
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_keys_before_timestamp",
+        lambda *_a, **_k: [],
+    )
+    # shrink the anchor budget window so the floor (300.0) sits above the anchor
+    monkeypatch.setattr(session_ops, "_REGENERATION_SIDECAR_ANCHOR_BUDGET", 1)
+    seen = {}
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_messages",
+        lambda *_a, **_k: _capture(seen, _k),
+    )
+    session_ops.regeneration_state(session, use_sidecar=True)
+    assert "since_timestamp" not in seen, "compression anchor below floor must force the full read"
+
+
+def test_bounded_guard_compression_anchor_covered_allows(monkeypatch):
+    from api import session_ops
+
+    session = _sidecar_session(3)
+    session.compression_anchor_message_key = {"role": "user", "text": "m0", "ts": 100.0}
+    # budget 200 covers all 3 rows → floor = min(all) = 100 → anchor at floor is covered
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_prefix_summary",
+        lambda *_a, **_k: {"count": 0, "null_timestamp_count": 0},
+    )
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_message_keys_before_timestamp",
+        lambda *_a, **_k: [],
+    )
+    seen = {}
+    monkeypatch.setattr(
+        "api.models.get_state_db_session_messages",
+        lambda *_a, **_k: _capture(seen, _k),
+    )
+    session_ops.regeneration_state(session, use_sidecar=True)
+    assert seen.get("since_timestamp") is not None, "anchor covered by floor must keep the bounded tail read"
+
+
+def _capture(seen, kwargs):
+    seen.update(kwargs)
+    return []