diff --git a/api/routes.py b/api/routes.py index f25e80c5..740cbed5 100644 --- a/api/routes.py +++ b/api/routes.py @@ -28,7 +28,7 @@ import time import uuid import http.client import socket as _socket -from collections import defaultdict, deque +from collections import defaultdict, deque, OrderedDict from pathlib import Path from contextlib import closing from urllib.parse import parse_qs, quote, unquote, urljoin, urlsplit @@ -8940,7 +8940,82 @@ def _limited_webui_messages_for_display(session, state_db_messages) -> list: ) -def _limited_webui_messages_for_display_with_sidecar(session, sidecar_messages, state_db_messages) -> list: +def _display_merge_session_is_active(session) -> bool: + """Return whether any canonical in-memory projection is active/pending.""" + if getattr(session, "active_stream_id", None) or getattr( + session, "pending_user_message", None + ): + return True + sid = str(getattr(session, "session_id", "") or "") + if not sid: + return True + with LOCK: + live = SESSIONS.get(sid) + if live is None or live is session: + return False + if str(getattr(live, "session_id", "") or "") != sid: + return True + return bool( + getattr(live, "active_stream_id", None) + or getattr(live, "pending_user_message", None) + ) + + +def _display_merge_cached_messages(session, sidecar_messages, *, msg_before=None): + """Return the memoized merged transcript, or None when it can't be reused. + + Lets GET /api/session skip loading the state.db rows entirely on a hit. That + is only sound because the cache key can be built from the existing + commit-reliable DB/WAL/SHM signature (`_state_db_session_signature`) rather + than a fingerprint computed FROM the loaded rows -- otherwise the key could + not be built without paying exactly the cost we are trying to avoid. + + Fail-closed by construction: returns None whenever the session is active, + the key cannot be built, or the cached entry does not match, and the caller + then performs the normal full load + merge. + """ + if msg_before is not None or _display_merge_session_is_active(session): + return None + sid = str(getattr(session, "session_id", "") or "") + if not sid: + return None + with _display_merge_cache_lock: + entry = _display_merge_cache.get(sid) + if entry is None: + return None + # Resolve the sidecar exactly like the merge helper does: it treats None as + # "load the lineage myself", and the cache entry was keyed on that RESOLVED + # list. Probing with a bare None would key on an empty sidecar and miss + # every time -- silently reverting this optimisation. + if sidecar_messages is None: + sidecar_messages = _webui_sidecar_lineage_messages_for_display(session) + else: + sidecar_messages = list(sidecar_messages or []) + # Building the key requires the sidecar rows (cheap: already in memory or + # served from the lineage cache) but not the state.db rows -- that + # asymmetry is the whole point. + cache_key = _display_merge_cache_key(session, sidecar_messages, None) + if cache_key is None: + return None + with _display_merge_cache_lock: + entry = _display_merge_cache.get(sid) + if not _display_merge_cache_entry_usable(entry, cache_key): + return None + _display_merge_cache.move_to_end(sid, last=True) + return [dict(m) if isinstance(m, dict) else m for m in entry["messages"]] + + +_DISPLAY_STATE_SIGNATURE_UNSET = object() + + +def _limited_webui_messages_for_display_with_sidecar( + session, + sidecar_messages, + state_db_messages, + *, + state_db_signature=_DISPLAY_STATE_SIGNATURE_UNSET, + msg_before=None, +) -> list: if sidecar_messages is None: sidecar_messages = _webui_sidecar_lineage_messages_for_display(session) else: @@ -8956,12 +9031,471 @@ def _limited_webui_messages_for_display_with_sidecar(session, sidecar_messages, # the paginated load). The append-only merge is O(n) over already-bounded # in-memory lists; the real latency win here is skipping the lineage-parent # DISK load above, which we still skip. (#4070 ship-review) - return merge_session_messages_append_only( + # + # perf: the merge itself is still expensive for multi-thousand-message + # historical transcripts (~2-3s per request: json.dumps merge keys + + # loose-content probes per row), and GET /api/session re-runs it on every + # open/poll. Memoize per session id. Validity is fail-closed: + # - only INACTIVE sessions (no active stream, no pending user message): + # an active session's in-memory tail can be ahead of its disk + # signature, so it always recomputes; + # - the child sidecar's exact stat signature plus every lineage parent + # signature recorded by _webui_sidecar_lineage_messages_for_display; + # - the sidecar row count and last timestamp (guards unsaved in-memory + # appends that have not reached disk yet); + # - a content fingerprint of the (bounded) state.db rows. + # Any uncertainty (missing signature, fingerprint failure) skips caching. + cache_key = None + # A msg_before request deliberately reads a different (uncapped) state.db + # scope than the initial tail request. It must bypass both cache layers: + # skipping only the pre-load probe still let this inner lookup reuse the + # initial 50k-row backstop merge and made the oldest row unreachable. + if msg_before is None and not _display_merge_session_is_active(session): + if state_db_signature is _DISPLAY_STATE_SIGNATURE_UNSET: + _state_key = _state_db_rows_fingerprint(state_db_messages) + else: + _state_key = state_db_signature + if _state_key is not None: + _current_key = _state_db_session_signature( + getattr(session, "session_id", None), + getattr(session, "profile", None) or None, + ) + if _current_key != _state_key: + _state_key = None + if _state_key is not None: + cache_key = _display_merge_cache_key( + session, + sidecar_messages, + state_db_messages, + state_db_signature=_state_key, + ) + if cache_key is not None: + sid = str(getattr(session, "session_id", "") or "") + with _display_merge_cache_lock: + entry = _display_merge_cache.get(sid) + if _display_merge_cache_entry_usable(entry, cache_key): + _display_merge_cache.move_to_end(sid, last=True) + return [dict(m) if isinstance(m, dict) else m for m in entry["messages"]] + merged = merge_session_messages_append_only( sidecar_messages, state_db_messages, truncation_watermark=getattr(session, "truncation_watermark", None), truncation_boundary=getattr(session, "truncation_boundary", None), ) + if cache_key is not None: + _state_key = cache_key[4] + _streaming_key = ( + isinstance(_state_key, (list, tuple)) + and bool(_state_key) + and _state_key[0] == "streaming" + ) + if ( + state_db_signature is not _DISPLAY_STATE_SIGNATURE_UNSET + and not _streaming_key + and _state_db_session_signature( + getattr(session, "session_id", None), + getattr(session, "profile", None) or None, + ) + != state_db_signature + ): + cache_key = None + if cache_key is not None: + sid = str(getattr(session, "session_id", "") or "") + with _display_merge_cache_lock: + _display_merge_cache[sid] = { + "key": cache_key, + "messages": merged, + "stored_at": time.monotonic(), + } + _display_merge_cache.move_to_end(sid, last=True) + while len(_display_merge_cache) > _DISPLAY_MERGE_CACHE_MAX: + _display_merge_cache.popitem(last=False) + # Same shallow-copy contract as the cache-hit path (and as the lineage + # cache): callers may attach display metadata to the returned rows. + return [dict(m) if isinstance(m, dict) else m for m in merged] + return merged + + +# perf: memoized sidecar↔state.db display merges for GET /api/session. +# See _limited_webui_messages_for_display_with_sidecar for the validity rules. +_DISPLAY_MERGE_CACHE_MAX = 16 +# Legacy streaming-freeze keys are still accepted defensively and remain +# tightly bounded. Production streaming keys now carry an exact target-session +# digest, so unrelated deltas stay stable without hiding target mutations. +_DISPLAY_MERGE_STREAMING_TTL_SECONDS = 5.0 +_display_merge_cache: "OrderedDict[str, dict]" = OrderedDict() +_display_merge_cache_lock = threading.Lock() + + +def _display_merge_cache_entry_usable(entry, cache_key) -> bool: + if entry is None or entry.get("key") != cache_key: + return False + try: + state_key = cache_key[4] + except (IndexError, TypeError): + return False + is_streaming_key = ( + isinstance(state_key, (list, tuple)) + and bool(state_key) + and state_key[0] == "streaming" + ) + if not is_streaming_key: + return True + try: + age = time.monotonic() - float(entry["stored_at"]) + except (KeyError, TypeError, ValueError): + return False + return 0.0 <= age <= _DISPLAY_MERGE_STREAMING_TTL_SECONDS + + +def _display_merge_requires_lineage_provenance(session) -> bool: + """Return whether this sidecar view depends on a stitched snapshot parent.""" + parent_id = str(getattr(session, "parent_session_id", "") or "").strip() + if not parent_id: + return False + if not is_safe_session_id(parent_id): + return True + try: + parent = Session.load(parent_id) + except Exception: + return True + if parent is None: + return True + if not getattr(parent, "pre_compression_snapshot", False): + return False + source = str(getattr(session, "session_source", "") or "").strip().lower() + parent_source = str( + getattr(parent, "session_source", "") or "" + ).strip().lower() + if source == "fork" and parent_source != "fork": + return False + return not _messages_start_with_visible_prefix( + list(getattr(session, "messages", []) or []), + list(getattr(parent, "messages", []) or []), + ) + + +def _evict_lineage_display_cache_entry(sid, expected_entry) -> None: + """Evict only the lineage entry that this caller validated as stale.""" + with _lineage_display_cache_lock: + if _lineage_display_cache.get(sid) is expected_entry: + _lineage_display_cache.pop(sid, None) + + +def _display_merge_cache_key( + session, + sidecar_messages, + state_db_messages, + *, + state_db_signature=_DISPLAY_STATE_SIGNATURE_UNSET, +): + """Return a fail-closed validity key for the display-merge cache, or None. + + None means "do not cache": any component that cannot be resolved exactly + (missing sidecar signature, unfingerprintable state rows) disables the + cache for this request rather than risking a stale transcript. + """ + from api.models import _sidecar_stat_signature + + sid = str(getattr(session, "session_id", "") or "") + if not sid or not is_safe_session_id(sid): + return None + self_sig = _sidecar_stat_signature(SESSION_DIR / f"{sid}.json") + if self_sig is None: + return None + # Lineage parents: reuse the signatures recorded by the (already memoized) + # lineage stitch so a write to any parent snapshot invalidates this cache + # too. A lineage without snapshot parents records no entry — empty tuple. + parent_sigs = () + with _lineage_display_cache_lock: + lineage_entry = _lineage_display_cache.get(sid) + if lineage_entry is not None: + if ( + lineage_entry.get("provenance_complete") is not True + or lineage_entry.get("self_sig") != self_sig + ): + _evict_lineage_display_cache_entry(sid, lineage_entry) + return None + parent_sigs = tuple( + (str(path), tuple(sig) if isinstance(sig, (list, tuple)) else sig) + for path, sig in (lineage_entry.get("parent_sigs") or []) + ) + for parent_path, parent_sig in parent_sigs: + if _sidecar_stat_signature(Path(parent_path)) != parent_sig: + _evict_lineage_display_cache_entry(sid, lineage_entry) + return None + with _lineage_display_cache_lock: + if _lineage_display_cache.get(sid) is not lineage_entry: + return None + if not parent_sigs and _display_merge_requires_lineage_provenance(session): + return None + last_ts = None + if sidecar_messages: + last = sidecar_messages[-1] + if isinstance(last, dict): + last_ts = last.get("timestamp") + # Prefer the existing commit-reliable DB/WAL/SHM signature outside streams. + # While another turn streams, use an exact digest scoped to this target + # session so unrelated per-delta commits do not churn the key. Fail closed + # onto the exact row fingerprint whenever the signature cannot be read. + if state_db_signature is _DISPLAY_STATE_SIGNATURE_UNSET: + state_fp = _state_db_session_signature( + sid, getattr(session, "profile", None) or None + ) + else: + state_fp = state_db_signature + if state_fp is None: + # state_db_messages is None on the cache-probe path, where the rows were + # deliberately not loaded. Fingerprinting None would key on the empty + # row set and could match an entry built from real rows, so fail closed. + if state_db_messages is None: + return None + state_fp = _state_db_rows_fingerprint(state_db_messages) + if state_fp is None: + return None + return ( + self_sig, + parent_sigs, + len(sidecar_messages), + last_ts, + state_fp, + getattr(session, "truncation_watermark", None), + getattr(session, "truncation_boundary", None), + ) + + +def _state_db_target_session_signature(db_path, session_id): + """Hash every target-session row without materialising display dictionaries.""" + try: + uri_path = quote(str(Path(db_path).resolve()), safe="/") + with closing( + sqlite3.connect(f"file:{uri_path}?mode=ro", uri=True, timeout=5.0) + ) as conn: + conn.execute("PRAGMA query_only=ON") + columns = [str(row[1]) for row in conn.execute("PRAGMA table_info(messages)")] + if "session_id" not in columns or "id" not in columns: + return None + quoted_columns = ", ".join( + '"' + column.replace('"', '""') + '"' for column in columns + ) + conn.text_factory = lambda raw: ("text", raw) + rows = conn.execute( + f'SELECT {quoted_columns} FROM messages ' + 'WHERE session_id = ? ORDER BY id', + (str(session_id),), + ) + digest = hashlib.blake2b(digest_size=32) + digest.update("\x1f".join(columns).encode("utf-8")) + row_count = 0 + for row in rows: + row_count += 1 + for value in row: + if value is None: + tag, payload = b"n", b"" + elif isinstance(value, tuple) and value[:1] == ("text",): + tag, payload = b"t", value[1] + elif isinstance(value, bytes): + tag, payload = b"b", value + elif isinstance(value, int): + tag, payload = b"i", str(value).encode("ascii") + elif isinstance(value, float): + tag, payload = b"f", value.hex().encode("ascii") + else: + return None + digest.update(tag) + digest.update(len(payload).to_bytes(8, "big")) + digest.update(payload) + return ("streaming-target", row_count, digest.hexdigest()) + except (OSError, sqlite3.Error, ValueError): + return None + + +def _state_db_target_session_revision(db_path, session_id): + """Return a bounded cross-process revision for one session's display rows. + + Supported writers update ``sessions.last_activity_at``/``message_count``. + The indexed tail revision additionally catches direct appends, tail deletes, + and raw changes to timestamp, row flags, or byte lengths on the newest row. + Legacy stores without a sessions row use full numeric/length aggregates. A + same-length raw SQL rewrite of an older row that bypasses session metadata + is outside the state-store writer contract; detecting it exactly would + require scanning and hashing every message payload (148 MB on the production + long session), which would make the cache slower than the uncached path. + """ + message_length_columns = ( + "role", + "content", + "tool_call_id", + "tool_calls", + "tool_name", + "finish_reason", + "reasoning", + "reasoning_content", + "reasoning_details", + "codex_reasoning_items", + "codex_message_items", + "effect_disposition", + "api_content", + "display_kind", + "display_metadata", + ) + message_sum_columns = ("observed", "active", "compacted") + session_revision_columns = ( + "message_count", + "last_activity_at", + "ended_at", + "end_reason", + "rewind_count", + "archived", + ) + try: + uri_path = quote(str(Path(db_path).resolve()), safe="/") + with closing( + sqlite3.connect(f"file:{uri_path}?mode=ro", uri=True, timeout=0.25) + ) as conn: + conn.execute("PRAGMA query_only=ON") + conn.execute("PRAGMA busy_timeout=250") + message_columns = { + str(row[1]) for row in conn.execute("PRAGMA table_info(messages)") + } + if "session_id" not in message_columns or "id" not in message_columns: + return None + session_columns = { + str(row[1]) for row in conn.execute("PRAGMA table_info(sessions)") + } + available_session_columns = [ + column + for column in session_revision_columns + if column in session_columns + ] + session_revision = None + if "id" in session_columns and available_session_columns: + session_revision = conn.execute( + "SELECT " + + ", ".join( + f'"{column}"' for column in available_session_columns + ) + + " FROM sessions WHERE id = ?", + (str(session_id),), + ).fetchone() + if session_revision is not None: + latest_parts = ["id"] + latest_parts.append( + "timestamp" if "timestamp" in message_columns else "NULL" + ) + latest_parts.extend( + f'LENGTH(COALESCE("{column}", \'\'))' + for column in message_length_columns + if column in message_columns + ) + latest_parts.extend( + f'COALESCE("{column}", 0)' + for column in message_sum_columns + if column in message_columns + ) + latest_revision = conn.execute( + f"SELECT {', '.join(latest_parts)} FROM messages " + "WHERE session_id = ? ORDER BY id DESC LIMIT 1", + (str(session_id),), + ).fetchone() + return ( + "target-session-revision-v2", + tuple(available_session_columns), + tuple(session_revision), + tuple(latest_revision) if latest_revision is not None else None, + ) + + # Legacy state stores without a sessions row have no supported + # O(1) activity revision. Keep them conservative by scanning compact + # numeric/length aggregates instead of trusting a global DB stamp. + aggregate_parts = ["COUNT(*)", "MAX(id)"] + aggregate_parts.append( + "MAX(timestamp)" if "timestamp" in message_columns else "NULL" + ) + aggregate_parts.extend( + f'SUM(LENGTH(COALESCE("{column}", \'\')))' + for column in message_length_columns + if column in message_columns + ) + aggregate_parts.extend( + f'SUM(COALESCE("{column}", 0))' + for column in message_sum_columns + if column in message_columns + ) + message_revision = conn.execute( + f"SELECT {', '.join(aggregate_parts)} FROM messages " + "WHERE session_id = ?", + (str(session_id),), + ).fetchone() + return ( + "target-session-revision-v1-legacy", + (), + None, + tuple(message_revision) if message_revision is not None else None, + ) + except (OSError, sqlite3.Error, ValueError): + return None + + +def _state_db_session_signature(session_id, profile=None): + """Return a cross-process target-session cache revision, fail-closed. + + The session-scoped revision avoids DB/WAL false invalidations caused by a + different conversation streaming in another WebUI process. If the schema + cannot provide that revision, fall back to the existing global file key. + """ + from api.models import _agent_state_db_path, _sqlite_file_stat_cache_key + + sid = str(session_id or "") + if not sid or not is_safe_session_id(sid): + return None + try: + db_path = _agent_state_db_path(profile=profile) + if not db_path or not Path(db_path).exists(): + return None + except Exception: + return None + target_revision = _state_db_target_session_revision(db_path, sid) + if target_revision is not None: + return target_revision + try: + signature = _sqlite_file_stat_cache_key(Path(db_path)) + except Exception: + return None + if signature is None: + return None + # ``_sqlite_file_stat_cache_key`` is a tuple of the content fingerprint and + # DB/WAL/SHM stat stamps. A completely empty result is not a valid key. + try: + if not any(component is not None for component in signature): + return None + except TypeError: + return None + return signature + + +def _load_state_db_messages_with_stable_signature(session_id, profile, reader_kwargs): + """Load rows and return the database signature that brackets that read.""" + before = _state_db_session_signature(session_id, profile) + rows = get_state_db_session_messages(session_id, **dict(reader_kwargs or {})) + after = _state_db_session_signature(session_id, profile) + stable = before if before is not None and before == after else None + return rows, stable + + +def _state_db_rows_fingerprint(rows) -> str | None: + """Content fingerprint of the state.db display rows, or None on failure.""" + try: + h = hashlib.sha256() + h.update(str(len(rows)).encode("utf-8")) + for row in rows: + if isinstance(row, dict): + h.update(json.dumps(row, sort_keys=True, default=str).encode("utf-8", "replace")) + else: + h.update(repr(row).encode("utf-8", "replace")) + return h.hexdigest() + except Exception: + return None def _sidecar_file_exceeds_threshold(session_id, threshold_bytes) -> bool: @@ -9059,6 +9593,16 @@ def _messages_start_with_visible_prefix(messages, prefix) -> bool: return False +# perf: memoized lineage-stitch results for GET /api/session. Keyed by session +# id; validity = exact stat signature of the child sidecar AND every snapshot +# parent involved in the stitch. Any write to any involved sidecar changes its +# signature and invalidates the entry. Bounded LRU — historical lineages are +# few but their merges cost seconds each. +_LINEAGE_DISPLAY_CACHE_MAX = 16 +_lineage_display_cache: "OrderedDict[str, dict]" = OrderedDict() +_lineage_display_cache_lock = threading.Lock() + + def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20) -> list: """Return WebUI sidecar messages stitched across compression snapshots. @@ -9067,20 +9611,80 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20) child sidecar. Opening the child alone makes older turns look lost. Stitch only those snapshot parents for display; ordinary forks also carry ``parent_session_id`` but must remain independent conversations. + + perf: the stitched merge is O(total messages) with expensive per-row keys + (json.dumps of tool_calls, loose-content regex). For multi-thousand-message + lineages it costs seconds per request, and GET /api/session re-runs it on + every open/poll. The result is cached per session id, keyed by the stat + signature of every sidecar involved (child + each snapshot parent), so an + idle historical lineage merges once and any write to any involved sidecar + invalidates naturally. Cache hits return shallow-copied rows so callers can + attach display metadata without corrupting the cache. """ + from api.models import _sidecar_stat_signature + + cache_allowed = not _display_merge_session_is_active(session) + sid = str(getattr(session, "session_id", "") or "") + self_sig = None + if sid and is_safe_session_id(sid): + self_sig = _sidecar_stat_signature(SESSION_DIR / f"{sid}.json") + if cache_allowed and self_sig is not None: + with _lineage_display_cache_lock: + entry = _lineage_display_cache.get(sid) + if ( + entry is not None + and entry.get("provenance_complete") is True + and entry.get("self_sig") == self_sig + ): + stale = False + for parent_path, parent_sig in entry.get("parent_sigs") or []: + if _sidecar_stat_signature(Path(parent_path)) != parent_sig: + stale = True + break + if not stale: + with _lineage_display_cache_lock: + current_entry = _lineage_display_cache.get(sid) + if current_entry is entry: + _lineage_display_cache.move_to_end(sid, last=True) + return [ + dict(m) if isinstance(m, dict) else m + for m in entry["messages"] + ] + else: + _evict_lineage_display_cache_entry(sid, entry) + segments = [] current = session session_messages = list(getattr(session, "messages", []) or []) source = str(getattr(session, "session_source", "") or "").strip().lower() root_is_fork = source == "fork" seen = {str(getattr(session, "session_id", "") or "")} + parent_sigs: list[tuple[str, tuple]] = [] + parent_signatures_complete = True for _ in range(max(0, int(max_hops))): parent_id = str(getattr(current, "parent_session_id", "") or "").strip() - if not parent_id or parent_id in seen or not is_safe_session_id(parent_id): + if not parent_id: break + if parent_id in seen or not is_safe_session_id(parent_id): + parent_signatures_complete = False + break + parent_path = SESSION_DIR / f"{parent_id}.json" + parent_sig_before = _sidecar_stat_signature(parent_path) parent = Session.load(parent_id) - if not parent or not getattr(parent, "pre_compression_snapshot", False): + if not parent: + parent_signatures_complete = False break + if not getattr(parent, "pre_compression_snapshot", False): + break + parent_sig = _sidecar_stat_signature(parent_path) + if ( + parent_sig_before is None + or parent_sig is None + or parent_sig_before != parent_sig + ): + parent_signatures_complete = False + else: + parent_sigs.append((str(parent_path), parent_sig)) parent_source = str(getattr(parent, "session_source", "") or "").strip().lower() if root_is_fork and parent_source != "fork": break @@ -9092,6 +9696,10 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20) segments.append(parent) seen.add(parent_id) current = parent + else: + # Exhausting max_hops means the declared ancestry may continue beyond + # the signatures captured above. Never publish partial provenance. + parent_signatures_complete = False if not segments: return list(getattr(session, "messages", []) or []) @@ -9104,11 +9712,31 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20) truncation_watermark=getattr(segment, "truncation_watermark", None), truncation_boundary=getattr(segment, "truncation_boundary", None), ) - return merge_session_messages_append_only( + merged = merge_session_messages_append_only( merged, getattr(session, "messages", []) or [], truncation_watermark=None, ) + if ( + cache_allowed + and self_sig is not None + and parent_sigs + and parent_signatures_complete + ): + with _lineage_display_cache_lock: + _lineage_display_cache[sid] = { + "self_sig": self_sig, + "parent_sigs": parent_sigs, + "provenance_complete": True, + "messages": merged, + } + _lineage_display_cache.move_to_end(sid, last=True) + while len(_lineage_display_cache) > _LINEAGE_DISPLAY_CACHE_MAX: + _lineage_display_cache.popitem(last=False) + # Hand out copies so caller-side metadata mutation cannot corrupt + # the cached rows (same contract as the cache-hit path). + return [dict(m) if isinstance(m, dict) else m for m in merged] + return merged def _merged_session_messages_for_display(session, cli_messages=None) -> list: @@ -12904,6 +13532,11 @@ def handle_get(handler, parsed) -> bool: metadata_summary = None limited_sidecar_messages = None state_db_since_timestamp = None + # Set by the limited-display path when the memoized merge can be + # reused without loading the state.db rows; must exist for every + # branch below, including the ones that never probe the cache. + _display_cache_hit = None + _display_state_db_signature = None if is_messaging_session: cli_messages = get_cli_session_messages(sid) elif load_messages: @@ -12927,10 +13560,51 @@ def handle_get(handler, parsed) -> bool: _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, - ) + # perf: on the limited-display path the state.db rows are only + # consumed by the memoized merge below. Now that the cache key + # is a bounded SQL signature rather than a fingerprint OF these + # rows, a hit no longer needs them -- and materialising tens of + # thousands of dicts was the dominant remaining cost (~2.3s on a + # 36k-row session) even when the merge itself was served from + # cache. Probe the cache first and skip the load on a hit. + # + # Deliberately narrow: only when msg_limit is set (the merge + # helper below is the sole consumer) and only for inactive + # sessions, matching the cache's own validity rule. Any miss + # falls through to the normal full load, so this can only skip + # work that would have produced an identical merged result. + _display_cache_hit = None + if ( + msg_limit is not None + and not getattr(s, "active_stream_id", None) + and not getattr(s, "pending_user_message", None) + ): + _display_cache_hit = _display_merge_cached_messages( + s, + limited_sidecar_messages, + msg_before=msg_before, + ) + if _display_cache_hit is not None: + state_db_messages = [] + else: + if ( + msg_limit is not None + and not getattr(s, "active_stream_id", None) + and not getattr(s, "pending_user_message", None) + ): + ( + state_db_messages, + _display_state_db_signature, + ) = _load_state_db_messages_with_stable_signature( + sid, + _session_profile, + _state_db_reader_kwargs, + ) + else: + state_db_messages = get_state_db_session_messages( + sid, + **_state_db_reader_kwargs, + ) elif not is_messaging_session: # Metadata-only callers still need the same append-only # reconciliation contract as full loads so stale/replayed @@ -12963,11 +13637,16 @@ def handle_get(handler, parsed) -> bool: # them chronologically and dedupe exact repeats. _all_msgs = _merged_session_messages_for_display(s, cli_messages) elif msg_limit is not None: - _all_msgs = _limited_webui_messages_for_display_with_sidecar( - s, - limited_sidecar_messages, - state_db_messages, - ) + if _display_cache_hit is not None: + _all_msgs = _display_cache_hit + else: + _all_msgs = _limited_webui_messages_for_display_with_sidecar( + s, + limited_sidecar_messages, + state_db_messages, + state_db_signature=_display_state_db_signature, + msg_before=msg_before, + ) else: _all_msgs = merge_session_messages_append_only( _webui_sidecar_lineage_messages_for_display(s), diff --git a/tests/test_display_merge_cache.py b/tests/test_display_merge_cache.py new file mode 100644 index 00000000..57318997 --- /dev/null +++ b/tests/test_display_merge_cache.py @@ -0,0 +1,119 @@ +"""The display merge for paginated loads must be memoized for inactive sessions. + +Before the fix, GET /api/session re-ran merge_session_messages_append_only on +every request (~2-3s for multi-thousand-message transcripts). The cache must: + +- return the same merged transcript as a direct merge (correctness), +- serve copies (caller mutation cannot corrupt the cache), +- invalidate when the sidecar file changes on disk, +- invalidate when the state.db rows change, +- never cache ACTIVE sessions (active_stream_id / pending_user_message). +""" + +import time +from types import SimpleNamespace + +import pytest + +@pytest.fixture() +def routes_env(tmp_path, monkeypatch): + import api.config as config + import api.models as models + import api.routes as routes + + home = tmp_path / "home" + state_dir = tmp_path / "state" + session_dir = state_dir / "sessions" + session_dir.mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_WEBUI_STATE_DIR", str(state_dir)) + monkeypatch.setattr(config, "STATE_DIR", state_dir) + monkeypatch.setattr(config, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(routes, "SESSION_DIR", session_dir) + with routes._display_merge_cache_lock: + routes._display_merge_cache.clear() + yield SimpleNamespace(config=config, models=models, routes=routes) + with routes._display_merge_cache_lock: + routes._display_merge_cache.clear() + + +def _make_session(routes_env, sid="20260101_000000_cache1", n=6): + models = routes_env.models + s = models.Session(session_id=sid) + now = time.time() + for i in range(n): + role = "user" if i % 2 == 0 else "assistant" + s.messages.append({"role": role, "content": f"turn {i}", "timestamp": now + i}) + s.save() + return s + + +def _state_rows(base_ts, n=2): + return [ + {"role": "assistant", "content": f"state row {i}", "timestamp": base_ts + 100 + i} + for i in range(n) + ] + + +def test_merge_cached_and_equal_to_direct_merge(routes_env): + routes = routes_env.routes + s = _make_session(routes_env) + rows = _state_rows(s.messages[-1]["timestamp"]) + + first = routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + assert routes._display_merge_cache, "expected a cache entry for an inactive session" + second = routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + assert [m.get("content") for m in first] == [m.get("content") for m in second] + assert any(m.get("content") == "state row 0" for m in second) + + +def test_cache_hit_returns_copies(routes_env): + routes = routes_env.routes + s = _make_session(routes_env, sid="20260101_000000_cache2") + rows = _state_rows(s.messages[-1]["timestamp"]) + + first = routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + first[0]["content"] = "MUTATED" + second = routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + assert second[0]["content"] != "MUTATED" + + +def test_cache_invalidated_by_sidecar_write(routes_env): + routes = routes_env.routes + s = _make_session(routes_env, sid="20260101_000000_cache3") + rows = _state_rows(s.messages[-1]["timestamp"]) + + routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + s.messages.append({ + "role": "assistant", "content": "new turn after write", + "timestamp": time.time() + 50, + }) + s.save() + merged = routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + assert any(m.get("content") == "new turn after write" for m in merged) + + +def test_cache_invalidated_by_state_rows_change(routes_env): + routes = routes_env.routes + s = _make_session(routes_env, sid="20260101_000000_cache4") + rows = _state_rows(s.messages[-1]["timestamp"]) + + routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + rows2 = rows + [{ + "role": "assistant", "content": "brand new state row", + "timestamp": rows[-1]["timestamp"] + 5, + }] + merged = routes._limited_webui_messages_for_display_with_sidecar(s, None, rows2) + assert any(m.get("content") == "brand new state row" for m in merged) + + +def test_active_session_never_cached(routes_env): + routes = routes_env.routes + s = _make_session(routes_env, sid="20260101_000000_cache5") + s.active_stream_id = "stream-live" + rows = _state_rows(s.messages[-1]["timestamp"]) + + routes._display_merge_cache.clear() + routes._limited_webui_messages_for_display_with_sidecar(s, None, rows) + assert s.session_id not in routes._display_merge_cache diff --git a/tests/test_display_merge_cache_shortcut.py b/tests/test_display_merge_cache_shortcut.py new file mode 100644 index 00000000..6e663175 --- /dev/null +++ b/tests/test_display_merge_cache_shortcut.py @@ -0,0 +1,353 @@ +"""Coverage for skipping the state.db load on a display-merge cache hit. + +GET /api/session used to materialise every state.db row for a session even when +the merged transcript was already memoized -- ~2.3s of pure waste on a 36k-row +session. The rows are only consumed by the merge, so a cache hit can skip the +load entirely. + +That is only sound because the cache key no longer depends on the loaded rows. +These tests pin the two properties that make the shortcut safe: + 1. a hit returns EXACTLY what the full load+merge path returns; + 2. anything uncertain (active session, unbuildable key, changed data) falls + back to the full load rather than serving a stale transcript. +""" + +import sys +import time +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from api import routes # noqa: E402 + +SID = "20260301_090000_cache1" + + +class _Session: + """Minimal session stand-in for the cache-probe contract.""" + + def __init__(self, sid=SID, active=None, pending=None): + self.session_id = sid + self.profile = None + self.active_stream_id = active + self.pending_user_message = pending + self.truncation_watermark = None + self.truncation_boundary = None + + +@pytest.fixture(autouse=True) +def _clean_cache(): + with routes._display_merge_cache_lock: + routes._display_merge_cache.clear() + yield + with routes._display_merge_cache_lock: + routes._display_merge_cache.clear() + + +@pytest.fixture() +def stable_key(monkeypatch): + """Make the cache key deterministic without touching the filesystem.""" + monkeypatch.setattr( + "api.models._sidecar_stat_signature", lambda p: ("sig", 1, 2, 3), raising=False) + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: "SIG-A") + return "SIG-A" + + +def _seed(session, sidecar, merged): + key = routes._display_merge_cache_key(session, sidecar, None) + assert key is not None + with routes._display_merge_cache_lock: + routes._display_merge_cache[session.session_id] = { + "key": key, + "messages": merged, + "stored_at": time.monotonic(), + } + return key + + +# -------------------------------------------------------------------------- +# The shortcut must be exact. +# -------------------------------------------------------------------------- + +def test_cache_hit_returns_the_memoized_transcript(stable_key): + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + merged = [ + {"role": "user", "content": "hi", "timestamp": 10.0}, + {"role": "assistant", "content": "hello", "timestamp": 11.0}, + ] + _seed(session, sidecar, merged) + + got = routes._display_merge_cached_messages(session, sidecar) + assert got == merged + + +def test_cache_hit_returns_copies_not_shared_rows(stable_key): + """Callers attach display metadata; the cached rows must not be mutated.""" + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + merged = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, merged) + + got = routes._display_merge_cached_messages(session, sidecar) + got[0]["injected"] = True + + again = routes._display_merge_cached_messages(session, sidecar) + assert "injected" not in again[0], "cache entry was mutated by a caller" + + +def test_shortcut_matches_the_full_merge_path(monkeypatch, stable_key): + """The whole point: identical output with and without the shortcut. + + Populate the cache through the real merge helper, then assert the probe + returns the same transcript the helper would have recomputed. + """ + session = _Session() + sidecar = [{"role": "user", "content": "one", "timestamp": 1.0}] + rows = [ + {"role": "user", "content": "one", "timestamp": 1.0}, + {"role": "assistant", "content": "two", "timestamp": 2.0}, + ] + + full = routes._limited_webui_messages_for_display_with_sidecar( + session, + sidecar, + rows, + state_db_signature=stable_key, + ) + probed = routes._display_merge_cached_messages(session, sidecar) + + assert probed is not None, "merge helper did not populate the cache" + assert probed == full + + +# -------------------------------------------------------------------------- +# Fail-closed: never serve a stale transcript. +# -------------------------------------------------------------------------- + +def test_active_stream_never_uses_the_cache(stable_key): + """An active session's in-memory tail can be ahead of its disk signature.""" + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "hi", "timestamp": 10.0}]) + + session.active_stream_id = "stream-123" + assert routes._display_merge_cached_messages(session, sidecar) is None + + +def test_pending_user_message_never_uses_the_cache(stable_key): + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "hi", "timestamp": 10.0}]) + + session.pending_user_message = {"content": "not yet on disk"} + assert routes._display_merge_cached_messages(session, sidecar) is None + + +def test_live_memory_session_blocks_hit_from_stale_idle_object(stable_key): + """A stale disk object must not bypass the canonical in-memory active queue.""" + stale = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + merged = [{"role": "user", "content": "cached", "timestamp": 10.0}] + _seed(stale, sidecar, merged) + live = _Session(active="stream-live") + + with routes.LOCK: + previous = routes.SESSIONS.get(stale.session_id) + routes.SESSIONS[stale.session_id] = live + try: + assert routes._display_merge_cached_messages(stale, sidecar) is None + finally: + with routes.LOCK: + if previous is None: + routes.SESSIONS.pop(stale.session_id, None) + else: + routes.SESSIONS[stale.session_id] = previous + + +def test_changed_state_db_signature_misses(monkeypatch, stable_key): + """New state.db rows must invalidate the entry, not be silently skipped.""" + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "hi", "timestamp": 10.0}]) + + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: "SIG-B") + assert routes._display_merge_cached_messages(session, sidecar) is None + + +def test_streaming_freeze_key_hits_before_ttl(monkeypatch, stable_key): + marker = ("streaming", ("run-1",)) + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: marker) + now = [1000.0] + monkeypatch.setattr(routes.time, "monotonic", lambda: now[0]) + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + merged = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, merged) + + now[0] += routes._DISPLAY_MERGE_STREAMING_TTL_SECONDS - 0.1 + assert routes._display_merge_cached_messages(session, sidecar) == merged + + +def test_streaming_freeze_key_expires_after_ttl(monkeypatch, stable_key): + marker = ("streaming", ("run-1",)) + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: marker) + now = [1000.0] + monkeypatch.setattr(routes.time, "monotonic", lambda: now[0]) + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "hi", "timestamp": 10.0}]) + + now[0] += routes._DISPLAY_MERGE_STREAMING_TTL_SECONDS + 0.1 + assert routes._display_merge_cached_messages(session, sidecar) is None + + +def test_changed_sidecar_tail_misses(stable_key): + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "hi", "timestamp": 10.0}]) + + grown = sidecar + [{"role": "assistant", "content": "new", "timestamp": 12.0}] + assert routes._display_merge_cached_messages(session, grown) is None + + +def test_empty_cache_misses(stable_key): + assert routes._display_merge_cached_messages(_Session(), []) is None + + +def test_msg_before_pagination_never_uses_tail_cache(stable_key): + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "cached tail", "timestamp": 10.0}]) + + assert routes._display_merge_cached_messages( + session, + sidecar, + msg_before=100, + ) is None + + +def test_msg_before_full_read_cannot_hit_inner_tail_cache(stable_key): + """The merge helper must not reuse the initial backstop result after paging.""" + session = _Session() + tail_rows = [ + {"role": "user", "content": "row-2", "timestamp": 2.0}, + {"role": "assistant", "content": "row-3", "timestamp": 3.0}, + ] + full_rows = [ + {"role": "assistant", "content": "row-1", "timestamp": 1.0}, + *tail_rows, + ] + + initial = routes._limited_webui_messages_for_display_with_sidecar( + session, + [], + tail_rows, + state_db_signature=stable_key, + ) + assert [row["content"] for row in initial] == ["row-2", "row-3"] + + paging = routes._limited_webui_messages_for_display_with_sidecar( + session, + [], + full_rows, + state_db_signature=stable_key, + msg_before=2, + ) + assert [row["content"] for row in paging] == ["row-1", "row-2", "row-3"] + + +def test_probe_fails_closed_when_bounded_signature_unavailable(monkeypatch): + """Without the bounded signature the key would need the rows we skipped. + + Fingerprinting the absent rows would key on an empty row set and could match + an entry built from real rows -- so the probe must refuse instead. + """ + monkeypatch.setattr( + "api.models._sidecar_stat_signature", lambda p: ("sig", 1, 2, 3), raising=False) + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: "SIG-A") + + session = _Session() + sidecar = [{"role": "user", "content": "hi", "timestamp": 10.0}] + _seed(session, sidecar, [{"role": "user", "content": "hi", "timestamp": 10.0}]) + + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: None) + assert routes._display_merge_cached_messages(session, sidecar) is None + + +def test_cache_key_refuses_none_rows_without_bounded_signature(monkeypatch): + """Direct contract test for the probe-path guard in the key builder.""" + monkeypatch.setattr( + "api.models._sidecar_stat_signature", lambda p: ("sig", 1, 2, 3), raising=False) + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *a, **k: None) + + key = routes._display_merge_cache_key(_Session(), [{"timestamp": 1.0}], None) + assert key is None + + +def test_unsafe_session_id_is_refused(stable_key): + bad = _Session(sid="../../etc/passwd") + assert routes._display_merge_cached_messages(bad, []) is None + + +def test_loader_refuses_signature_when_commit_occurs_during_read(monkeypatch): + signatures = iter(("SIG-A", "SIG-B")) + monkeypatch.setattr( + routes, + "_state_db_session_signature", + lambda *args, **kwargs: next(signatures), + ) + rows = [{"role": "assistant", "content": "snapshot"}] + monkeypatch.setattr(routes, "get_state_db_session_messages", lambda *args, **kwargs: rows) + + loaded, stable = routes._load_state_db_messages_with_stable_signature( + SID, + profile=None, + reader_kwargs={}, + ) + + assert loaded == rows + assert stable is None + + +def test_loader_returns_signature_when_read_window_is_stable(monkeypatch): + monkeypatch.setattr(routes, "_state_db_session_signature", lambda *args, **kwargs: "SIG-A") + rows = [{"role": "assistant", "content": "snapshot"}] + monkeypatch.setattr(routes, "get_state_db_session_messages", lambda *args, **kwargs: rows) + + loaded, stable = routes._load_state_db_messages_with_stable_signature( + SID, + profile=None, + reader_kwargs={}, + ) + + assert loaded == rows + assert stable == "SIG-A" + + +def test_merge_does_not_publish_when_signature_changes_before_store(monkeypatch): + monkeypatch.setattr( + "api.models._sidecar_stat_signature", lambda path: ("sig", 1, 2, 3), raising=False + ) + signatures = iter(("SIG-A", "SIG-B")) + monkeypatch.setattr( + routes, + "_state_db_session_signature", + lambda *args, **kwargs: next(signatures), + ) + session = _Session() + sidecar = [{"role": "user", "content": "one", "timestamp": 1.0}] + rows = [{"role": "assistant", "content": "two", "timestamp": 2.0}] + + merged = routes._limited_webui_messages_for_display_with_sidecar( + session, + sidecar, + rows, + state_db_signature="SIG-A", + ) + + assert len(merged) == 2 + with routes._display_merge_cache_lock: + assert session.session_id not in routes._display_merge_cache diff --git a/tests/test_lineage_display_cache.py b/tests/test_lineage_display_cache.py new file mode 100644 index 00000000..aa298692 --- /dev/null +++ b/tests/test_lineage_display_cache.py @@ -0,0 +1,327 @@ +"""Perf regression tests — lineage-stitch memoization for GET /api/session. + +The compression-continuation stitch (`_webui_sidecar_lineage_messages_for_display`) +re-merged multi-thousand-message parent+child transcripts on EVERY /api/session +request (seconds per call on real sessions). The merge result is now cached, +keyed by the exact stat signature of every sidecar involved, so an idle +historical lineage merges once. These tests pin the contract: + +- warm calls return the same rows without re-merging; +- cache hits hand out copies (caller mutation cannot corrupt the cache); +- any write to child OR snapshot-parent sidecar invalidates the entry. +""" +from __future__ import annotations + +import time + +import pytest + +import api.profiles as profiles + + +@pytest.fixture +def hermes_home(tmp_path, monkeypatch): + home = tmp_path / "hermes_home" + home.mkdir() + (home / "sessions").mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", home) + return home + + +@pytest.fixture +def lineage(hermes_home, monkeypatch): + """A snapshot parent + continuation child pair persisted to disk.""" + import api.routes as routes + from api.models import Session + + monkeypatch.setattr(routes, "SESSION_DIR", hermes_home / "sessions") + import api.models as models + + monkeypatch.setattr(models, "SESSION_DIR", hermes_home / "sessions") + # Fresh cache per test. + routes._lineage_display_cache.clear() + + parent = Session( + session_id="lineage_parent", + title="parent", + messages=[ + {"role": "user", "content": f"question {i}", "timestamp": 1000 + i * 10} + if i % 2 == 0 + else {"role": "assistant", "content": f"answer {i}", "timestamp": 1000 + i * 10} + for i in range(40) + ], + ) + parent.pre_compression_snapshot = True + parent.save() + + child = Session( + session_id="lineage_child", + title="child", + messages=[ + {"role": "user", "content": "suite", "timestamp": 5000}, + {"role": "assistant", "content": "reponse suite", "timestamp": 5010}, + ], + ) + child.parent_session_id = "lineage_parent" + child.save() + return routes, Session, child + + +def test_lineage_stitch_is_cached_and_correct(lineage): + routes, Session, child = lineage + m1 = routes._webui_sidecar_lineage_messages_for_display(child) + assert routes._lineage_display_cache, "merge result must be cached" + m2 = routes._webui_sidecar_lineage_messages_for_display(child) + assert [m.get("timestamp") for m in m1] == [m.get("timestamp") for m in m2] + # Parent turns + child turns all present. + assert len(m1) == 42 + + +def test_cache_hits_return_copies_not_aliases(lineage): + routes, Session, child = lineage + routes._webui_sidecar_lineage_messages_for_display(child) + m2 = routes._webui_sidecar_lineage_messages_for_display(child) + m2[0]["__mutated"] = True + m3 = routes._webui_sidecar_lineage_messages_for_display(child) + assert "__mutated" not in m3[0], "caller mutation leaked into the cache" + + +def test_child_write_invalidates_cache(lineage, hermes_home): + routes, Session, child = lineage + routes._webui_sidecar_lineage_messages_for_display(child) + child.messages.append({"role": "user", "content": "new turn", "timestamp": 6000}) + time.sleep(0.01) + child.save() + reloaded = Session.load("lineage_child") + merged = routes._webui_sidecar_lineage_messages_for_display(reloaded) + assert any(m.get("content") == "new turn" for m in merged), ( + "stale cache served after the child sidecar changed" + ) + + +def test_parent_write_invalidates_cache(lineage, hermes_home): + routes, Session, child = lineage + routes._webui_sidecar_lineage_messages_for_display(child) + parent = Session.load("lineage_parent") + parent.messages.append( + {"role": "assistant", "content": "late parent row", "timestamp": 1999} + ) + time.sleep(0.01) + parent.save() + merged = routes._webui_sidecar_lineage_messages_for_display(child) + assert any(m.get("content") == "late parent row" for m in merged), ( + "stale cache served after the snapshot parent sidecar changed" + ) + + +@pytest.mark.parametrize( + "field,value", + [ + ("active_stream_id", "stream-live"), + ("pending_user_message", {"content": "queued"}), + ], +) +def test_active_or_pending_lineage_bypasses_reads_and_writes(lineage, field, value): + routes, Session, child = lineage + routes._webui_sidecar_lineage_messages_for_display(child) + setattr(child, field, value) + child.messages.append( + {"role": "assistant", "content": "unsaved live tail", "timestamp": 2000} + ) + + merged = routes._webui_sidecar_lineage_messages_for_display(child) + assert any(m.get("content") == "unsaved live tail" for m in merged) + + with routes._lineage_display_cache_lock: + routes._lineage_display_cache.pop(child.session_id, None) + routes._webui_sidecar_lineage_messages_for_display(child) + with routes._lineage_display_cache_lock: + assert child.session_id not in routes._lineage_display_cache + + +def test_lru_eviction_during_parent_validation_cannot_crash(lineage, monkeypatch): + """A cache entry may disappear while its parent signatures are checked.""" + routes, Session, child = lineage + import api.models as models + + expected = routes._webui_sidecar_lineage_messages_for_display(child) + real_signature = models._sidecar_stat_signature + evicted = False + + def evict_on_parent(path): + nonlocal evicted + if not evicted and path.stem == "lineage_parent": + evicted = True + with routes._lineage_display_cache_lock: + routes._lineage_display_cache.pop(child.session_id, None) + return real_signature(path) + + monkeypatch.setattr(models, "_sidecar_stat_signature", evict_on_parent) + got = routes._webui_sidecar_lineage_messages_for_display(child) + + assert evicted + assert [m.get("content") for m in got] == [m.get("content") for m in expected] + + +def test_incomplete_multihop_parent_signatures_disable_cache(hermes_home, monkeypatch): + import api.models as models + import api.routes as routes + from api.models import Session + + monkeypatch.setattr(routes, "SESSION_DIR", hermes_home / "sessions") + monkeypatch.setattr(models, "SESSION_DIR", hermes_home / "sessions") + routes._lineage_display_cache.clear() + + grandparent = Session( + session_id="lineage_grandparent", + messages=[{"role": "user", "content": "grand", "timestamp": 1}], + ) + grandparent.pre_compression_snapshot = True + grandparent.save() + parent = Session( + session_id="lineage_parent_two", + messages=[{"role": "assistant", "content": "parent", "timestamp": 2}], + ) + parent.pre_compression_snapshot = True + parent.parent_session_id = grandparent.session_id + parent.save() + child = Session( + session_id="lineage_child_two", + messages=[{"role": "user", "content": "child", "timestamp": 3}], + ) + child.parent_session_id = parent.session_id + child.save() + + real_signature = models._sidecar_stat_signature + + def incomplete_signature(path): + if path.stem == grandparent.session_id: + return None + return real_signature(path) + + monkeypatch.setattr(models, "_sidecar_stat_signature", incomplete_signature) + got = routes._webui_sidecar_lineage_messages_for_display(child) + + assert [m.get("content") for m in got] == ["grand", "parent", "child"] + with routes._lineage_display_cache_lock: + assert child.session_id not in routes._lineage_display_cache + assert routes._display_merge_cache_key( + child, + got, + [], + state_db_signature=("db", "stable"), + ) is None + + +def test_parent_replace_during_load_is_not_published(lineage, monkeypatch): + """A parent signature must bracket the snapshot used for the cached stitch.""" + routes, Session, child = lineage + real_load = Session.load + replaced = False + + def load_and_replace(cls, session_id): + nonlocal replaced + loaded = real_load(session_id) + if session_id == "lineage_parent" and not replaced: + replaced = True + replacement = real_load(session_id) + replacement.messages[0] = { + "role": "user", + "content": "parent-new", + "timestamp": 1000, + } + replacement.save() + return loaded + + monkeypatch.setattr(Session, "load", classmethod(load_and_replace)) + + first = routes._webui_sidecar_lineage_messages_for_display(child) + assert first[0]["content"] == "question 0" + + second = routes._webui_sidecar_lineage_messages_for_display(child) + assert second[0]["content"] == "parent-new" + + +def test_unverifiable_warm_parent_evicts_lineage_and_blocks_display_hit( + lineage, monkeypatch +): + """Old provenance cannot survive a parent replacement plus stat failure.""" + routes, Session, child = lineage + from api import models + + state_signature = ("target-session-revision",) + monkeypatch.setattr( + routes, "_state_db_session_signature", lambda *_a, **_k: state_signature + ) + old_messages = routes._webui_sidecar_lineage_messages_for_display(child) + old_key = routes._display_merge_cache_key( + child, + old_messages, + [], + state_db_signature=state_signature, + ) + assert old_key is not None + with routes._display_merge_cache_lock: + routes._display_merge_cache[child.session_id] = { + "key": old_key, + "messages": old_messages, + "stored_at": 0.0, + } + + parent = Session.load("lineage_parent") + parent.messages[0] = { + "role": "user", + "content": "parent-new", + "timestamp": 1000, + } + parent.save() + real_signature = models._sidecar_stat_signature + + def unavailable_parent(path): + if path.stem == "lineage_parent": + return None + return real_signature(path) + + monkeypatch.setattr(models, "_sidecar_stat_signature", unavailable_parent) + rebuilt = routes._webui_sidecar_lineage_messages_for_display(child) + + assert rebuilt[0]["content"] == "parent-new" + with routes._lineage_display_cache_lock: + assert child.session_id not in routes._lineage_display_cache + assert routes._display_merge_cached_messages(child, rebuilt) is None + + +def test_missing_declared_ancestor_disables_cache_until_it_appears(lineage): + """Negative lineage provenance must invalidate when the file appears later.""" + routes, Session, child = lineage + parent = Session.load("lineage_parent") + parent.parent_session_id = "lineage_grand" + parent.save() + + first = routes._webui_sidecar_lineage_messages_for_display(child) + assert all(row["content"] != "grand" for row in first) + with routes._lineage_display_cache_lock: + assert child.session_id not in routes._lineage_display_cache + + grand = Session(session_id="lineage_grand", workspace=".") + grand.pre_compression_snapshot = True + grand.messages = [{"role": "user", "content": "grand", "timestamp": 10}] + grand.save() + + second = routes._webui_sidecar_lineage_messages_for_display(child) + assert second[0]["content"] == "grand" + + +def test_sessions_without_lineage_do_not_pollute_cache(lineage): + routes, Session, child = lineage + routes._lineage_display_cache.clear() + solo = Session( + session_id="lineage_solo", + title="solo", + messages=[{"role": "user", "content": "hi", "timestamp": 1}], + ) + solo.save() + out = routes._webui_sidecar_lineage_messages_for_display(solo) + assert len(out) == 1 + assert "lineage_solo" not in routes._lineage_display_cache diff --git a/tests/test_state_db_session_signature.py b/tests/test_state_db_session_signature.py new file mode 100644 index 00000000..ba44cb77 --- /dev/null +++ b/tests/test_state_db_session_signature.py @@ -0,0 +1,304 @@ +"""Coverage for the commit-reliable state.db display-cache signature. + +The display-merge cache key used to embed ``_state_db_rows_fingerprint``, which +serialises every state.db display row with ``json.dumps``. On a 36k-row session +that cost ~1.4s *per request*, which became the floor cost of even a cache HIT -- +the cache could never pay for itself. + +``_state_db_session_signature`` now uses a cross-process target-session revision +derived from supported session metadata plus indexed message aggregates. These +tests pin the properties that matter: supported target mutations move the +signature, direct insert/delete/length/timestamp changes remain visible, +unrelated sessions stay stable without a process-local stream marker, and +failures return None so the caller falls back to the exact row fingerprint. +""" + +import sqlite3 +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from api import routes # noqa: E402 + +SID = "20260101_120000_abcdef" + + +def _build_db(path, rows=None): + """Minimal messages table matching the columns the signature reads.""" + conn = sqlite3.connect(str(path)) + try: + conn.execute( + "CREATE TABLE messages (" + " id INTEGER PRIMARY KEY," + " session_id TEXT," + " role TEXT," + " content TEXT," + " timestamp REAL)" + ) + conn.execute("CREATE INDEX idx_ms ON messages(session_id, id)") + payload = rows if rows is not None else [ + (i, SID, "user" if i % 2 else "assistant", + f"message number {i} with enough text to slice into thirds", + 1700000000.0 + i) + for i in range(1, 41) + ] + conn.executemany( + "INSERT INTO messages (id, session_id, role, content, timestamp) " + "VALUES (?,?,?,?,?)", payload) + conn.execute( + "CREATE TABLE sessions (" + " id TEXT PRIMARY KEY," + " message_count INTEGER," + " last_activity_at REAL," + " last_read_at REAL," + " ended_at REAL," + " end_reason TEXT," + " rewind_count INTEGER," + " archived INTEGER)" + ) + conn.execute( + "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?)", + (SID, len(payload), 1700000100.0, None, None, None, 0, 0), + ) + conn.commit() + finally: + conn.close() + return path + + +@pytest.fixture() +def db(tmp_path, monkeypatch): + path = _build_db(tmp_path / "state.db") + monkeypatch.setattr( + "api.models._agent_state_db_path", lambda profile=None: path, raising=False) + monkeypatch.setattr( + "api.models._cli_sessions_streaming_freeze_marker", lambda: None, raising=False) + return path + + +def _sig(db_path): + return routes._state_db_session_signature(SID, None) + + +def _mutate(db_path, sql, params=(), *, touch_session=False): + conn = sqlite3.connect(str(db_path)) + try: + conn.execute(sql, params) + if touch_session: + conn.execute( + "UPDATE sessions SET " + "message_count=(SELECT COUNT(*) FROM messages WHERE session_id=?), " + "last_activity_at=COALESCE(last_activity_at, 0)+1 " + "WHERE id=?", + (SID, SID), + ) + conn.commit() + finally: + conn.close() + + +# -------------------------------------------------------------------------- +# Detection matrix: every one of these must change the signature, otherwise a +# stale transcript can be served from the display-merge cache. +# -------------------------------------------------------------------------- + +MUTATIONS = [ + pytest.param( + "INSERT INTO messages (session_id, role, content, timestamp) " + "VALUES (?, 'user', 'brand new row', 1700009999.0)", (SID,), + id="append_row"), + pytest.param( + "DELETE FROM messages WHERE rowid = " + "(SELECT MAX(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="delete_tail_row"), + pytest.param( + "DELETE FROM messages WHERE rowid = " + "(SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="delete_head_row"), + pytest.param( + "UPDATE messages SET content = 'short' WHERE rowid = " + "(SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_content_different_length"), + pytest.param( + "UPDATE messages SET timestamp = 1.0 WHERE rowid = " + "(SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_timestamp"), + pytest.param( + "UPDATE messages SET role = 'tool' WHERE rowid = " + "(SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_role_only"), + pytest.param( + "UPDATE messages SET content = 'Z' || SUBSTR(content, 2) WHERE rowid = " + "(SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_first_char_same_length"), + pytest.param( + "UPDATE messages SET content = " + "SUBSTR(content,1,LENGTH(content)/2-1) || 'Z' || SUBSTR(content,LENGTH(content)/2+1) " + "WHERE rowid = (SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_middle_char_same_length"), + pytest.param( + "UPDATE messages SET content = " + "SUBSTR(content,1,LENGTH(content)/3-1) || 'Z' || SUBSTR(content,LENGTH(content)/3+1) " + "WHERE rowid = (SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_third_char_same_length"), + pytest.param( + "UPDATE messages SET content = " + "SUBSTR(content,1,LENGTH(content)-2) || 'Z' || SUBSTR(content,LENGTH(content)) " + "WHERE rowid = (SELECT MIN(rowid) FROM messages WHERE session_id = ?)", (SID,), + id="edit_tail_char_same_length"), +] + + +@pytest.mark.parametrize("sql,params", MUTATIONS) +def test_signature_detects_mutation(db, sql, params): + """A stale cache is only safe if the key moves on every visible mutation.""" + before = _sig(db) + assert before is not None + _mutate(db, sql, params, touch_session=True) + after = _sig(db) + assert after is not None + assert after != before, "signature did not move -> stale transcript risk" + + +def test_signature_is_stable_without_mutation(db): + """No spurious invalidation: repeated reads of an unchanged DB must match.""" + first = _sig(db) + assert first is not None + for _ in range(3): + assert _sig(db) == first + + +def test_signature_is_scoped_to_target_without_process_local_stream_marker(db): + """Cross-process writes to another session must not churn this cache key.""" + first = _sig(db) + assert first is not None + + _mutate( + db, + "INSERT INTO messages (session_id, role, content, timestamp) " + "VALUES ('20260202_000000_other', 'user', 'unrelated delta', 1700001111.0)", + ) + assert _sig(db) == first + + _mutate( + db, + "UPDATE messages SET content = 'Z' || SUBSTR(content, 2) WHERE rowid = " + "(SELECT MIN(rowid) FROM messages WHERE session_id = ?)", + (SID,), + touch_session=True, + ) + assert _sig(db) != first, "supported target edit did not advance session revision" + + +def test_raw_target_tail_length_change_invalidates_without_session_metadata(db): + """A direct edit of the indexed tail shape still invalidates.""" + before = _sig(db) + _mutate( + db, + "UPDATE messages SET content = 'short' WHERE rowid = " + "(SELECT MAX(rowid) FROM messages WHERE session_id = ?)", + (SID,), + ) + assert _sig(db) != before + + +def test_signature_stays_stable_on_other_session_write(db): + """Session-scoped revision avoids global DB/WAL false invalidations.""" + before = _sig(db) + _mutate( + db, + "INSERT INTO messages (session_id, role, content, timestamp) " + "VALUES ('20260202_000000_other', 'user', 'unrelated', 1700001111.0)") + assert _sig(db) == before + + +# -------------------------------------------------------------------------- +# Fail-closed behaviour: the caller must be able to fall back safely. +# -------------------------------------------------------------------------- + +def test_signature_returns_none_for_unsafe_session_id(db): + assert routes._state_db_session_signature("../../etc/passwd", None) is None + assert routes._state_db_session_signature("", None) is None + assert routes._state_db_session_signature(None, None) is None + + +def test_signature_returns_none_when_db_missing(tmp_path, monkeypatch): + missing = tmp_path / "absent.db" + monkeypatch.setattr( + "api.models._agent_state_db_path", lambda profile=None: missing, raising=False) + monkeypatch.setattr( + "api.models._cli_sessions_streaming_freeze_marker", + lambda: ("streaming", ("other-active-run",)), + raising=False, + ) + assert routes._state_db_session_signature(SID, None) is None + + +def test_signature_uses_file_fallback_on_incomplete_schema(tmp_path, monkeypatch): + """File stamps remain a valid key when content summaries are unavailable.""" + path = tmp_path / "state.db" + conn = sqlite3.connect(str(path)) + try: + conn.execute("CREATE TABLE messages (id INTEGER PRIMARY KEY)") + conn.commit() + finally: + conn.close() + monkeypatch.setattr( + "api.models._agent_state_db_path", lambda profile=None: path, raising=False) + first = routes._state_db_session_signature(SID, None) + assert first is not None + assert routes._state_db_session_signature(SID, None) == first + + +def test_signature_returns_none_when_all_fingerprints_fail(db, monkeypatch): + monkeypatch.setattr( + routes, "_state_db_target_session_revision", lambda *a, **k: None + ) + monkeypatch.setattr( + "api.models._sqlite_file_stat_cache_key", lambda path: None, raising=False) + assert routes._state_db_session_signature(SID, None) is None + + +def test_signature_returns_none_when_resolver_raises(monkeypatch): + def boom(profile=None): + raise RuntimeError("resolver exploded") + + monkeypatch.setattr("api.models._agent_state_db_path", boom, raising=False) + assert routes._state_db_session_signature(SID, None) is None + + +def test_cache_key_falls_back_to_exact_fingerprint(tmp_path, monkeypatch): + """When the bounded signature is unavailable, the key must still be built. + + This is the fail-closed contract: degraded to the old (slow but exact) + fingerprint rather than disabling the cache or trusting a partial key. + """ + monkeypatch.setattr( + routes, "_state_db_session_signature", lambda *a, **k: None) + + calls = [] + real = routes._state_db_rows_fingerprint + + def spy(rows): + calls.append(len(rows or [])) + return real(rows) + + monkeypatch.setattr(routes, "_state_db_rows_fingerprint", spy) + + class _S: + session_id = SID + profile = None + truncation_watermark = None + truncation_boundary = None + + monkeypatch.setattr( + "api.models._sidecar_stat_signature", lambda p: ("sig", 1, 2, 3), raising=False) + + rows = [{"role": "user", "content": "x", "timestamp": 1.0}] + key = routes._display_merge_cache_key(_S(), [{"timestamp": 1.0}], rows) + + assert calls, "exact fingerprint was not used as fallback" + assert key is not None