Use session index for compression continuations

This commit is contained in:
Frank Song authored and nesquena-hermes committed 2026-06-26 16:27:40 +00:00
1 parent 7763e92d09
commit 0f87757a47
2 files changed
+187 -45

No files matched your search

+94 -44
View File
@@ -6763,9 +6763,8 @@ def _pre_compression_continuation_session_id(session) -> str | None:
# snapshot's profile and reject any child that isn't profile-matched.
snapshot_profile = getattr(session, "profile", None)
def _child_rows() -> list:
def _child_rows_from_memory(seen_ids: set[str]) -> list:
rows = []
seen_ids = set()
try:
with LOCK:
memory_sessions = list(SESSIONS.values())
@@ -6777,6 +6776,35 @@ def _pre_compression_continuation_session_id(session) -> str | None:
rows.append(child)
except Exception:
pass
return rows
def _child_rows_from_index(seen_ids: set[str]) -> list | None:
if not SESSION_INDEX_FILE.exists():
return None
try:
entries = json.loads(SESSION_INDEX_FILE.read_text(encoding="utf-8"))
except Exception:
return None
if not isinstance(entries, list):
return None
rows = []
for entry in entries:
if not isinstance(entry, dict):
continue
child_sid = _safe_first(entry.get("session_id"))
if (
not child_sid
or child_sid in seen_ids
or not is_safe_session_id(child_sid)
or not _safe_first(entry.get("parent_session_id"))
):
continue
seen_ids.add(child_sid)
rows.append(entry)
return rows
def _child_rows_from_sidecars(seen_ids: set[str]) -> list:
rows = []
try:
for path in SESSION_DIR.glob("*.json"):
if path.name.startswith("_"):
@@ -6792,51 +6820,73 @@ def _pre_compression_continuation_session_id(session) -> str | None:
pass
return rows
children_by_parent: dict[str, list] = {}
for child in _child_rows():
parent_sid = _safe_first(getattr(child, "parent_session_id", None))
child_sid = _safe_first(getattr(child, "session_id", None))
if not parent_sid or not child_sid or child_sid == sid:
continue
# Cross-profile guard: only follow continuations within the snapshot's profile.
if not _profiles_match(getattr(child, "profile", None), snapshot_profile):
continue
children_by_parent.setdefault(parent_sid, []).append(child)
def _row_value(row, key, default=None):
return row.get(key, default) if isinstance(row, dict) else getattr(row, key, default)
candidates = []
frontier = [sid]
seen = {sid}
for _ in range(20):
if not frontier:
break
parent_sid = frontier.pop(0)
for child in children_by_parent.get(parent_sid, []):
child_sid = _safe_first(getattr(child, "session_id", None))
if not child_sid or child_sid in seen:
def _row_has_backing_state(row) -> bool:
child_sid = _safe_first(_row_value(row, "session_id"))
if not child_sid or not is_safe_session_id(child_sid):
return False
if not isinstance(row, dict):
return True
return (SESSION_DIR / f"{child_sid}.json").exists()
def _resolve_from_rows(rows: list) -> str | None:
children_by_parent: dict[str, list] = {}
for child in rows:
parent_sid = _safe_first(_row_value(child, "parent_session_id"))
child_sid = _safe_first(_row_value(child, "session_id"))
if not parent_sid or not child_sid or child_sid == sid:
continue
seen.add(child_sid)
if getattr(child, "pre_compression_snapshot", False):
frontier.append(child_sid)
else:
candidates.append(child)
# Cross-profile guard: only follow continuations within the snapshot's profile.
if not _profiles_match(_row_value(child, "profile"), snapshot_profile):
continue
children_by_parent.setdefault(parent_sid, []).append(child)
candidates = []
frontier = [sid]
seen = {sid}
for _ in range(20):
if not frontier:
break
parent_sid = frontier.pop(0)
for child in children_by_parent.get(parent_sid, []):
child_sid = _safe_first(_row_value(child, "session_id"))
if not child_sid or child_sid in seen or not _row_has_backing_state(child):
continue
seen.add(child_sid)
if _row_value(child, "pre_compression_snapshot", False):
frontier.append(child_sid)
else:
candidates.append(child)
if not candidates:
return None
latest = max(
candidates,
key=lambda child: float(
_safe_first(
_row_value(child, "updated_at"),
_row_value(child, "created_at"),
0,
) or 0
),
)
latest_sid = _row_value(latest, "session_id", None) or None
# Only hand the client a well-formed session id (it gets written to URL/localStorage).
if latest_sid and not is_safe_session_id(latest_sid):
return None
return latest_sid
seen_ids: set[str] = set()
rows = _child_rows_from_memory(seen_ids)
index_rows = _child_rows_from_index(seen_ids)
if index_rows is not None:
return _resolve_from_rows(rows + index_rows)
rows.extend(_child_rows_from_sidecars(seen_ids))
return _resolve_from_rows(rows)
if not candidates:
return None
latest = max(
candidates,
key=lambda child: float(
_safe_first(
getattr(child, "updated_at", None),
getattr(child, "created_at", None),
0,
) or 0
),
)
latest_sid = getattr(latest, "session_id", None) or None
# Only hand the client a well-formed session id (it gets written to URL/localStorage).
if latest_sid and not is_safe_session_id(latest_sid):
return None
return latest_sid
from api.workspace import (
load_workspaces,
save_workspaces,
@@ -1,5 +1,7 @@
"""Regression coverage for mobile reload recovery after compression session rotation."""
import collections
import json
from pathlib import Path
@@ -68,7 +70,6 @@ def test_continuation_lookup_is_profile_scoped(tmp_path, monkeypatch):
# Empty session dir so only in-memory SESSIONS are considered.
monkeypatch.setattr(config, "SESSION_DIR", tmp_path, raising=False)
monkeypatch.setattr(routes, "SESSION_DIR", tmp_path, raising=False)
import collections
fake = collections.OrderedDict()
for s in (same_profile_child, foreign_child):
fake[s.session_id] = s
@@ -82,3 +83,94 @@ def test_continuation_lookup_is_profile_scoped(tmp_path, monkeypatch):
fake2[foreign_child.session_id] = foreign_child
monkeypatch.setattr(routes, "SESSIONS", fake2, raising=False)
assert routes._pre_compression_continuation_session_id(snapshot) is None
def test_continuation_lookup_uses_index_without_scanning_sidecars(tmp_path, monkeypatch):
"""Indexed continuation metadata should avoid an O(all sidecars) recovery scan."""
from api import routes, config
class _S:
def __init__(self, sid, profile, snap=False):
self.session_id = sid
self.profile = profile
self.parent_session_id = None
self.pre_compression_snapshot = snap
self.updated_at = 100.0
self.created_at = 100.0
snapshot = _S("snapindex001", "work", snap=True)
index_file = tmp_path / "_index.json"
(tmp_path / "childindex01.json").write_text("{}", encoding="utf-8")
for idx in range(50):
(tmp_path / f"noise{idx:08d}.json").write_text("{}", encoding="utf-8")
index_file.write_text(
json.dumps(
[
{
"session_id": "childindex01",
"profile": "work",
"parent_session_id": "snapindex001",
"pre_compression_snapshot": False,
"updated_at": 300.0,
"created_at": 200.0,
}
]
),
encoding="utf-8",
)
monkeypatch.setattr(config, "SESSION_DIR", tmp_path, raising=False)
monkeypatch.setattr(routes, "SESSION_DIR", tmp_path, raising=False)
monkeypatch.setattr(routes, "SESSION_INDEX_FILE", index_file, raising=False)
monkeypatch.setattr(routes, "SESSIONS", collections.OrderedDict(), raising=False)
monkeypatch.setattr(
routes.Session,
"load_metadata_only",
staticmethod(lambda _sid: (_ for _ in ()).throw(AssertionError("must not scan sidecars"))),
)
assert routes._pre_compression_continuation_session_id(snapshot) == "childindex01"
def test_indexed_continuation_lookup_keeps_profile_scope(tmp_path, monkeypatch):
"""The index fast path must preserve the cross-profile continuation guard."""
from api import routes, config
class _S:
def __init__(self, sid, profile, snap=False):
self.session_id = sid
self.profile = profile
self.parent_session_id = None
self.pre_compression_snapshot = snap
self.updated_at = 100.0
self.created_at = 100.0
snapshot = _S("snapindex002", "work", snap=True)
index_file = tmp_path / "_index.json"
(tmp_path / "foreignidx01.json").write_text("{}", encoding="utf-8")
index_file.write_text(
json.dumps(
[
{
"session_id": "foreignidx01",
"profile": "personal",
"parent_session_id": "snapindex002",
"updated_at": 300.0,
"created_at": 200.0,
}
]
),
encoding="utf-8",
)
monkeypatch.setattr(config, "SESSION_DIR", tmp_path, raising=False)
monkeypatch.setattr(routes, "SESSION_DIR", tmp_path, raising=False)
monkeypatch.setattr(routes, "SESSION_INDEX_FILE", index_file, raising=False)
monkeypatch.setattr(routes, "SESSIONS", collections.OrderedDict(), raising=False)
monkeypatch.setattr(
routes.Session,
"load_metadata_only",
staticmethod(lambda _sid: (_ for _ in ()).throw(AssertionError("must not scan sidecars"))),
)
assert routes._pre_compression_continuation_session_id(snapshot) is None