Release v0.51.311 — Release KA (brick-wave: workspace Git RCE hardening #3769 + stale-snapshot sidebar visibility #3770) (#3776)

Brick-wave batch. #3769 (@Hinotoi-agent) hardens workspace Git config execution against repo-local RCE; #3770 (@ai-ag2026) keeps fuller pre-compression snapshots visible when _index.json is stale. Full suite 8176 passed, Codex SAFE, Opus SHIP IT. Co-authored-by: Hinotoi-agent; Co-authored-by: ai-ag2026

api/models.py

@@ -2550,13 +2550,19 @@ def _prefer_fuller_snapshots_for_sidebar(sessions: list[dict]) -> list[dict]:
 
         newest_visible_ts = max(_session_sort_timestamp(session) for session in visible)
         snapshot_ts = _session_sort_timestamp(best_snapshot)
-        # Keep the active continuation visible when it has newer activity than
-        # the archived snapshot. A fuller snapshot can still be older than a
-        # continuation that contains the latest turns after compression.
+        snapshot_id = str(best_snapshot.get('session_id') or '')
+        if not snapshot_id:
+            continue
+
+        snapshot_ids_to_show.add(snapshot_id)
+        # If the continuation is newer, keep it visible too. That means the
+        # lineage is split-brain-ish: the snapshot has more transcript rows, but
+        # the continuation may still contain the newest post-compression turn.
+        # Showing both is less tidy than hiding one, but it preserves every
+        # reachable message. Tidy and wrong is how users start doubting reality.
         if newest_visible_ts > snapshot_ts:
             continue
 
-        snapshot_ids_to_show.add(str(best_snapshot.get('session_id')))
         continuation_ids_to_hide.update(
             str(session.get('session_id'))
             for session in visible
@@ -2583,28 +2589,88 @@ def _strip_sidebar_internal_flags(sessions: list[dict]) -> None:
         session.pop('_show_pre_compression_snapshot', None)
 
 
-def _row_may_need_sidecar_metadata_refresh(session: dict) -> bool:
+def _row_may_need_sidecar_metadata_refresh(
+    session: dict,
+    *,
+    stale_snapshot_ids: set[str] | None = None,
+) -> bool:
     """Return True when a row needs canonical sidecar runtime/snapshot metadata.
 
     Compression lineage fields are enriched from state.db in one batched query
     later in all_sessions(). Loading hundreds of lineage sidecars on every
     /api/sessions poll turns the sidebar into molasses, so keep this refresh
-    limited to the few rows whose transient runtime or snapshot state is not
-    cheaply available from state.db.
+    limited to rows with transient runtime state, missing snapshot sidebar
+    metadata, or a stale snapshot candidate that can affect the visibility
+    decision for its lineage.
     """
     is_runtime_row = bool(
         session.get('active_stream_id')
         or session.get('has_pending_user_message')
         or session.get('pending_user_message')
     )
-    snapshot_missing_sidebar_metadata = bool(
-        session.get('pre_compression_snapshot')
-        and (
-            session.get('message_count') is None
-            or session.get('last_message_at') is None
-        )
-    )
-    return is_runtime_row or snapshot_missing_sidebar_metadata
+    if is_runtime_row:
+        return True
+    if not session.get('pre_compression_snapshot'):
+        return False
+    if session.get('message_count') is None or session.get('last_message_at') is None:
+        return True
+    sid = str(session.get('session_id') or '')
+    return bool(sid and stale_snapshot_ids and sid in stale_snapshot_ids)
+
+
+def _sidecar_mtime_after_index_timestamp(session: dict) -> bool:
+    sid = str(session.get('session_id') or '')
+    if not sid or not is_safe_session_id(sid):
+        return False
+    try:
+        sidecar_mtime = (SESSION_DIR / f'{sid}.json').stat().st_mtime
+    except OSError:
+        return False
+    indexed_ts = _session_sort_timestamp(session)
+    return sidecar_mtime > indexed_ts + 0.001
+
+
+def _stale_snapshot_metadata_refresh_ids(sessions: list[dict]) -> set[str]:
+    """Return pre-compression snapshots worth a sidecar metadata refresh.
+
+    Most snapshot rows can be decided from the index: either their indexed count
+    already beats the visible continuation, or they are normal older snapshots
+    that should remain hidden. Only stat candidate sidecars when a hidden
+    snapshot has a visible continuation in the same lineage and its indexed
+    metadata would otherwise fail to expose it.
+    """
+    sessions_by_id = {
+        str(session.get('session_id')): session
+        for session in sessions
+        if session.get('session_id')
+    }
+    groups: dict[str, list[dict]] = {}
+    for session in sessions:
+        sid = str(session.get('session_id') or '')
+        source = session.get('source_tag') or session.get('source')
+        if source == 'cron' or sid.startswith('cron_'):
+            continue
+        root = _sidebar_lineage_root_id(session, sessions_by_id)
+        groups.setdefault(root, []).append(session)
+
+    refresh_ids: set[str] = set()
+    for group in groups.values():
+        visible = [session for session in group if not session.get('pre_compression_snapshot')]
+        snapshots = [session for session in group if session.get('pre_compression_snapshot')]
+        if not visible or not snapshots:
+            continue
+        if any(_has_live_sidebar_state(session) for session in visible):
+            continue
+        best_visible_count = max(_sidebar_message_count(session) for session in visible)
+        for snapshot in snapshots:
+            sid = str(snapshot.get('session_id') or '')
+            if not sid:
+                continue
+            if _sidebar_message_count(snapshot) > best_visible_count:
+                continue
+            if _sidecar_mtime_after_index_timestamp(snapshot):
+                refresh_ids.add(sid)
+    return refresh_ids
 
 
 def _refresh_index_rows_from_sidecar_metadata(sessions: list[dict]) -> list[dict]:
@@ -2616,8 +2682,12 @@ def _refresh_index_rows_from_sidecar_metadata(sessions: list[dict]) -> list[dict
     historical transcript.
     """
     out: list[dict] = []
+    stale_snapshot_ids = _stale_snapshot_metadata_refresh_ids(sessions)
     for session in sessions:
-        if not _row_may_need_sidecar_metadata_refresh(session):
+        if not _row_may_need_sidecar_metadata_refresh(
+            session,
+            stale_snapshot_ids=stale_snapshot_ids,
+        ):
             out.append(session)
             continue
         sid = session.get('session_id')