From e4342a65cfaf0020c9032ad3f2fe5e55d007ba4a Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 9 Jul 2026 19:05:54 -0700 Subject: [PATCH] Release: split anchor_activity_scenes after messages to fix #4633 server memory growth (#5858, #5854) (#5863) Reasoning-heavy sessions accumulate 250-480KB of anchor_activity_scenes serialized BEFORE the messages array, so the 64KB metadata-prefix read always overflowed -> full-parse fallback on every read -> memory growth. Move scene data AFTER messages; metadata prefix now ~2KB, modern reads skip the full parse. Old-layout files still load (one authoritative parse then cached); round-trip preserves scenes exactly. Independent review: nesquena APPROVED (end-to-end, on head). Gate-pass: Codex SAFE (933KB legacy sidecar loads clean + round-trip verified), 17 own + 2945 regression green. Co-authored-by: t Co-authored-by: nesquena-hermes --- CHANGELOG.md | 2 + api/models.py | 418 ++++++++++++++++++--- tests/test_issue5854_anchor_scene_split.py | 346 +++++++++++++++++ 3 files changed, 724 insertions(+), 42 deletions(-) create mode 100644 tests/test_issue5854_anchor_scene_split.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f720f72..50d8f773 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ ### Fixed +- **Server memory growth on reasoning-heavy sessions is fixed (#4633 backend root cause).** Reasoning-heavy sessions accumulate 250–480 KB of `anchor_activity_scenes` that were serialized *before* the `messages` array in the session sidecar, so the cheap 64 KB metadata-prefix read always overflowed and fell back to a full parse on every read — driving memory growth. The scene data is now serialized *after* `messages`, so the metadata prefix stays small (≈2 KB) and modern reads need no full parse. Existing session files in the old layout still load correctly (one authoritative parse, then cached), and round-trips preserve scene content exactly. Thanks @nesquena-hermes. (#5858, #5854, #4633) + - **Uploading an Office document now writes a Markdown sidecar so the agent can read its text.** When you upload a `.docx`, `.xlsx`, or `.pptx` to a workspace, a best-effort `.md` sidecar with the extracted text is written alongside it (reusing the existing Office-preview extractor — no new dependency). The write is path-contained (`O_NOFOLLOW` openat-walk within the workspace), never overwrites an existing `.md` (`O_CREAT|O_EXCL`), and is best-effort — a sidecar failure leaves the uploaded Office file fully intact. Thanks @rodboev. (#5734, #540) - **Security: the embedded-terminal local-origin gate can no longer be bypassed via a spoofed `X-Forwarded-For` header, and no longer locks out direct LAN clients.** Previously, with `HERMES_WEBUI_TRUST_FORWARDED_FOR=1`, the gate honored the forwarded client IP without first verifying the request actually came through a trusted proxy — so a public client connecting directly to a passwordless WebUI could send `X-Forwarded-For: 127.0.0.1` and reach the embedded terminal (which runs a PTY shell). The forwarded chain is now consulted **only when the un-spoofable raw socket peer is a trusted proxy** (loopback, or an address in the new `HERMES_WEBUI_TRUSTED_PROXY_CIDRS` allowlist), it consumes the full `X-Forwarded-For` chain right-to-left (skipping trusted-proxy hops to find the real client), and it fails closed on malformed/empty chains. Direct loopback/LAN clients with no proxy header remain local, so onboarding, first-password/passkey setup, and passwordless terminal access keep working on the common direct-LAN deployment. Thanks @Hinotoi-agent for the original report. (#5764) diff --git a/api/models.py b/api/models.py index a0e7d94b..02f4f2b6 100644 --- a/api/models.py +++ b/api/models.py @@ -93,6 +93,17 @@ _SIDECAR_METADATA_CACHE_LOCK = threading.Lock() _SIDECAR_METADATA_CACHE: "collections.OrderedDict[tuple, dict]" = collections.OrderedDict() _SIDECAR_METADATA_CACHE_MAX = 2000 +# #5854: authoritative facts for a LEGACY (pre-#5854) sidecar whose scenes +# serialize before `messages`, so the cheap metadata-prefix read can't recover +# its message_count or scene fingerprint. Without this, an unchanged legacy +# large-scene session would full-parse on every poll (recreating the #4633 +# churn for legacy files) and could not be LRU-evicted. Populated once per file +# from a full Session.load(); keyed by the sidecar's stat signature so any edit +# invalidates it. Bounded. Value: {"message_count": int, "scene_index": dict}. +_LEGACY_SIDECAR_FACTS_LOCK = threading.Lock() +_LEGACY_SIDECAR_FACTS: "collections.OrderedDict[tuple, dict]" = collections.OrderedDict() +_LEGACY_SIDECAR_FACTS_MAX = 2000 + # --------------------------------------------------------------------------- # Stale temp-file cleanup # --------------------------------------------------------------------------- @@ -979,7 +990,18 @@ def _read_file_head(path: Path, max_prefix_bytes: int = 4096) -> str: def _read_metadata_json_prefix(path, max_prefix_bytes=65536): - """Read only the metadata portion before the top-level messages array.""" + """Read only the metadata portion before the large arrays. + + #5854: stop at the top-level ``messages`` key OR the top-level + ``anchor_activity_scenes`` key, whichever appears first. On the modern + layout scenes serialize AFTER ``messages`` so this stops at ``messages`` as + before (the prefix is now small — scene bodies are no longer in it). On the + LEGACY layout scenes serialize BEFORE ``messages`` and can be 250-480KB, so + stopping at ``anchor_activity_scenes`` keeps the read cheap and — critically + — still captures ``message_count`` (which is written before both). Without + the scenes-stop a legacy large-scene sidecar overflows ``max_prefix_bytes`` + and forces a full multi-MB parse on every poll (the #4633 churn). + """ buf = '' with open(path, 'r', encoding='utf-8') as f: while len(buf.encode('utf-8')) < max_prefix_bytes: @@ -987,10 +1009,13 @@ def _read_metadata_json_prefix(path, max_prefix_bytes=65536): if not chunk: return None buf += chunk - messages_pos = _find_top_level_json_key(buf, 'messages') - if messages_pos is None: + stop_pos = _find_top_level_json_key(buf, 'messages') + scenes_pos = _find_top_level_json_key(buf, 'anchor_activity_scenes') + if scenes_pos is not None and (stop_pos is None or scenes_pos < stop_pos): + stop_pos = scenes_pos + if stop_pos is None: continue - prefix = buf[:messages_pos].rstrip() + prefix = buf[:stop_pos].rstrip() if prefix.endswith(','): prefix = prefix[:-1].rstrip() return f'{prefix}\n}}' @@ -1172,6 +1197,14 @@ class Session: self.enabled_toolsets = enabled_toolsets # List[str] or None — per-session toolset override self.composer_draft = composer_draft if isinstance(composer_draft, dict) else {} self.anchor_activity_scenes = anchor_activity_scenes if isinstance(anchor_activity_scenes, dict) else {} + # #5854: a compact fingerprint of anchor_activity_scenes ({scene_key: + # updated_at}) persisted BEFORE the messages array so the sidebar-poll + # freshness check can compare scene freshness without parsing the full + # (often 250-480KB) scene bodies, which serialize AFTER messages. None + # on legacy sidecars (scenes-before-messages, no fingerprint) — callers + # fall back to reading keys/updated_at off anchor_activity_scenes. + _raw_scene_index = kwargs.get('anchor_scene_index') + self._anchor_scene_index = _raw_scene_index if isinstance(_raw_scene_index, dict) else None raw_message_count = kwargs.get('message_count') parsed_message_count = None if raw_message_count is not None: @@ -1231,15 +1264,31 @@ class Session: 'parent_session_id', 'worktree_path', 'worktree_branch', 'worktree_repo_root', 'worktree_created_at', 'is_cli_session', 'source_tag', 'raw_source', 'session_source', 'source_label', 'read_only', - 'enabled_toolsets', 'composer_draft', 'anchor_activity_scenes', + 'enabled_toolsets', 'composer_draft', ] meta = {k: getattr(self, k, None) for k in METADATA_FIELDS} + # #5854: message_count and a compact anchor-scene fingerprint go in the + # metadata prefix (BEFORE messages) so load_metadata_only() and the + # sidebar-poll freshness check never have to parse the full (250-480KB) + # scene bodies. message_count is placed BEFORE anchor_scene_index so a + # legacy-format reader that stops at a scene key still finds the count. + # The full anchor_activity_scenes bodies serialize AFTER messages. meta['message_count'] = len(self.messages or []) + meta['anchor_scene_index'] = _anchor_scene_index_from_records(self.anchor_activity_scenes) + # Keep the in-memory fingerprint aligned with what we just persisted, so a + # later metadata-only reload of THIS object (or any fingerprint reader) + # sees the current value rather than a stale load-time snapshot (#5854 + # defense-in-depth; the cached-side freshness check reads real records, + # not this, so this is belt-and-suspenders). + self._anchor_scene_index = dict(meta['anchor_scene_index']) meta['messages'] = self.messages meta['tool_calls'] = self.tool_calls - # Fields not in METADATA_FIELDS (e.g. last_usage) go at the end + meta['anchor_activity_scenes'] = self.anchor_activity_scenes if isinstance(self.anchor_activity_scenes, dict) else {} + # Fields not in METADATA_FIELDS (e.g. last_usage) go at the end. Exclude + # the keys we placed explicitly above so they aren't emitted twice. + _placed = {'message_count', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'} extra = {k: v for k, v in self.__dict__.items() - if k not in METADATA_FIELDS and k not in ('messages', 'tool_calls') + if k not in METADATA_FIELDS and k not in _placed and not k.startswith('_')} payload = json.dumps({**meta, **extra}, ensure_ascii=False, indent=2) @@ -1356,6 +1405,10 @@ class Session: p = SESSION_DIR / f'{sid}.json' if not p.exists(): return None + # #5854: snapshot the stat signature BEFORE reading so a legacy-facts + # cache write is only committed if the file didn't change under us + # during the parse (TOCTOU guard against an atomic replace mid-read). + _pre_read_sig = _sidecar_stat_signature(p) data = json.loads(p.read_text(encoding='utf-8')) data['messages'], _collapsed_partials = _collapse_adjacent_duplicate_partials(data.get('messages')) session = cls(**data) @@ -1367,6 +1420,27 @@ class Session: session.save(touch_updated_at=False, skip_index=True) except Exception: logger.debug("Failed to persist collapsed duplicate partials for %s", sid, exc_info=True) + else: + # #5854: for a LEGACY sidecar (no modern anchor_scene_index key), the + # cheap metadata-prefix read cannot recover message_count/scenes when + # scenes serialize before them, so cache the authoritative facts we + # just parsed. This keeps the metadata-only path and the eviction + # check from full-parsing this unchanged file again on every poll. + # Keyed by stat signature, so any edit invalidates it; the next + # save() rewrites the modern layout and the fallback stops firing. + # expected_sig guards against an atomic replace during the read. + # (When _collapsed_partials fired, save() above already rewrote the + # modern layout, so no legacy caching is needed.) + if 'anchor_scene_index' not in data: + try: + _legacy_sidecar_facts_put( + sid, + len(getattr(session, 'messages', None) or []), + _anchor_scene_index_from_records(getattr(session, 'anchor_activity_scenes', None)), + expected_sig=_pre_read_sig, + ) + except Exception: + logger.debug("legacy sidecar facts cache populate failed for %s", sid, exc_info=True) return session @classmethod @@ -1403,6 +1477,38 @@ class Session: index_message_count = index_message_counts.get(str(sid)) else: index_message_count = _lookup_index_message_count(sid) + # #5854 legacy-layout recovery: a pre-#5854 sidecar serialized + # anchor_activity_scenes BEFORE message_count, so on a large-scene + # legacy file the cheap prefix now stops at the scenes key and + # captures NO message_count. The sidebar _index.json count can lag + # behind external sidecar appends, so trusting it here would report a + # stale/zero count and could drop an unsaved user tail on the next + # get_session cache-replace. When the prefix carries NEITHER + # message_count NOR the modern anchor_scene_index key (⇒ a legacy + # file whose count fell after the scenes), recover the authoritative + # facts. To avoid re-parsing an unchanged legacy file on every poll + # (which would recreate the #4633 churn for legacy sidecars that are + # never re-saved), consult a bounded stat-signature cache first and + # only full-load on a miss, caching the result. The next save() + # rewrites the modern layout so the fallback stops firing entirely. + # A MODERN file always carries message_count in the prefix, so it + # never reaches here — a genuine 0 stays 0. + if ( + sidecar_message_count is None + and 'anchor_scene_index' not in parsed + ): + _facts = _legacy_sidecar_facts_get(sid) + if _facts is not None: + parsed['anchor_scene_index'] = _facts.get('scene_index') or {} + session = cls(**parsed) + session._metadata_message_count = _parse_nonnegative_int(_facts.get('message_count')) + session._loaded_metadata_only = True + return session + # Cache miss → full-load. cls.load() itself populates the legacy + # facts cache with a TOCTOU-guarded write (expected_sig), so we + # do NOT re-cache here (an unguarded second write could stamp + # stale facts under a replacement file's signature — Codex r5). + return cls.load(sid) # Modern sidecars carry an accurate message_count, so it is the # source of truth and we skip the per-row _index.json read in the # common case. The sidebar index is only a cache (it can lag behind @@ -3064,6 +3170,136 @@ def _cache_has_stale_unsaved_user_tail(cached, disk_session) -> bool: return _message_content_text(cached_tail) == _message_content_text(previous_disk_user) +def _anchor_scene_index_from_records(records) -> dict: + """Build the compact anchor-scene fingerprint {scene_key: updated_at} (#5854). + + This is the freshness signal the sidebar-poll comparison needs — scene keys + plus each scene's ``updated_at`` — WITHOUT the 250-480KB bodies. Persisted in + the metadata prefix (before ``messages``) so ``load_metadata_only`` and + ``_persisted_session_meta_prefix`` stay cheap. Mirrors exactly what + ``_anchor_scene_record_keys`` / ``_anchor_scene_records_updated_at`` read off + the full records, so the fingerprint comparison is behavior-identical. + """ + if not isinstance(records, dict): + return {} + index = {} + for key, value in records.items(): + if not key or not isinstance(value, dict): + continue + try: + updated_at = float(value.get('updated_at') or 0) + except (TypeError, ValueError): + updated_at = 0.0 + index[str(key)] = updated_at + return index + + +def _disk_scene_fingerprint(disk_meta_prefix: dict): + """Resolve the (scene_keys, max_updated_at) freshness signal from a parsed + metadata prefix dict, preferring the modern ``anchor_scene_index`` and + falling back to the full ``anchor_activity_scenes`` bodies for legacy files. + + Returns ``None`` when the prefix carries NEITHER field, so callers can tell + "no scenes" (empty dict/index present) apart from "couldn't determine" + (legacy file whose scenes serialize after ``messages`` and so aren't in the + prefix) and fall through to the full metadata load instead of assuming zero. + """ + if not isinstance(disk_meta_prefix, dict): + return None + if 'anchor_scene_index' in disk_meta_prefix: + raw = disk_meta_prefix.get('anchor_scene_index') + raw = raw if isinstance(raw, dict) else {} + keys = {str(k) for k in raw} + latest = 0.0 + for v in raw.values(): + try: + fv = float(v or 0) + except (TypeError, ValueError): + fv = 0.0 + if fv > latest: + latest = fv + return keys, latest + if 'anchor_activity_scenes' in disk_meta_prefix: + records = disk_meta_prefix.get('anchor_activity_scenes') + records = records if isinstance(records, dict) else {} + keys = {str(k) for k, val in records.items() if k and isinstance(val, dict)} + latest = 0.0 + for val in records.values(): + if not isinstance(val, dict): + continue + try: + fv = float(val.get('updated_at') or 0) + except (TypeError, ValueError): + fv = 0.0 + if fv > latest: + latest = fv + return keys, latest + return None + + +def _sidecar_stat_signature(path): + """Stat signature for a sidecar path, or None if it can't be stat'd. + + Any edit (atomic-rename or in-place) changes at least one component, so a + cached entry keyed by this signature is auto-invalidated on the next write. + """ + try: + st = path.stat() + except OSError: + return None + return (str(path), int(getattr(st, 'st_mtime_ns', int(st.st_mtime * 1_000_000_000))), + int(st.st_size), int(getattr(st, 'st_ctime_ns', int(st.st_ctime * 1_000_000_000)))) + + +def _legacy_sidecar_facts_get(sid): + """Return cached authoritative facts for a LEGACY sidecar, or None (#5854). + + Only returns a hit when the file's current stat signature matches the cached + one, so a stale entry can never be served after an edit. + """ + if not is_safe_session_id(sid): + return None + sig = _sidecar_stat_signature(SESSION_DIR / f'{sid}.json') + if sig is None: + return None + with _LEGACY_SIDECAR_FACTS_LOCK: + hit = _LEGACY_SIDECAR_FACTS.get(sig) + if hit is not None: + _LEGACY_SIDECAR_FACTS.move_to_end(sig) + return dict(hit) + return None + + +def _legacy_sidecar_facts_put(sid, message_count, scene_index, *, expected_sig): + """Cache authoritative facts for a legacy sidecar keyed by its stat signature. + + #5854 TOCTOU guard: ``expected_sig`` (MANDATORY) is the signature captured + BEFORE the caller parsed the file. The facts were derived from that snapshot, + so we only cache when the file's CURRENT signature still equals it — + otherwise the file was atomically replaced during the parse and these facts + describe the old content; caching them under the new signature would serve + stale data. Pass ``None`` explicitly only if the caller genuinely has no + snapshot (then this is a no-op, refusing to cache unverified facts). + """ + if not is_safe_session_id(sid): + return + if expected_sig is None: + return + sig = _sidecar_stat_signature(SESSION_DIR / f'{sid}.json') + if sig is None: + return + if sig != expected_sig: + # File changed under us during the parse — do not cache stale facts. + return + entry = {"message_count": message_count, + "scene_index": dict(scene_index) if isinstance(scene_index, dict) else {}} + with _LEGACY_SIDECAR_FACTS_LOCK: + _LEGACY_SIDECAR_FACTS[sig] = entry + _LEGACY_SIDECAR_FACTS.move_to_end(sig) + while len(_LEGACY_SIDECAR_FACTS) > _LEGACY_SIDECAR_FACTS_MAX: + _LEGACY_SIDECAR_FACTS.popitem(last=False) + + def _anchor_scene_record_keys(session) -> set[str]: records = getattr(session, 'anchor_activity_scenes', None) if not isinstance(records, dict): @@ -3088,6 +3324,44 @@ def _anchor_scene_records_updated_at(session) -> float: return latest +def _session_scene_keys(session) -> set[str]: + """Scene keys for a session object, fingerprint-aware (#5854). + + The ``_anchor_scene_index`` fingerprint is authoritative ONLY on a + metadata-only stub (whose full ``anchor_activity_scenes`` are not + materialized because they serialize after ``messages``). A FULLY-LOADED + session carries a load-time fingerprint that goes stale the moment its + records are mutated in place (the scene-persist path does exactly that + without refreshing it), so for a full session we MUST read the real records. + """ + if getattr(session, '_loaded_metadata_only', False): + index = getattr(session, '_anchor_scene_index', None) + if isinstance(index, dict): + return {str(k) for k in index} + return _anchor_scene_record_keys(session) + + +def _session_scene_updated_at(session) -> float: + """Max scene ``updated_at`` for a session object, fingerprint-aware (#5854). + + Fingerprint is authoritative only on a metadata-only stub; a fully-loaded + session always compares its real records (see ``_session_scene_keys``). + """ + if getattr(session, '_loaded_metadata_only', False): + index = getattr(session, '_anchor_scene_index', None) + if isinstance(index, dict): + latest = 0.0 + for v in index.values(): + try: + fv = float(v or 0) + except (TypeError, ValueError): + fv = 0.0 + if fv > latest: + latest = fv + return latest + return _anchor_scene_records_updated_at(session) + + def _cached_session_lags_disk(cached) -> bool: """Return True when a cached full session is older than its sidecar. @@ -3134,58 +3408,54 @@ def _cached_session_lags_disk(cached) -> bool: _scene_check_inconclusive = False if cached_scenes: disk_meta_quick = _persisted_session_meta_prefix(sid) - if disk_meta_quick is not None: - disk_scenes = disk_meta_quick.get('anchor_activity_scenes') or {} - if not isinstance(disk_scenes, dict): - disk_scenes = {} - if disk_scenes: + # #5854: modern prefixes carry only the anchor_scene_index + # fingerprint (keys + updated_at), not the full scene bodies (those + # now serialize after `messages`). _disk_scene_fingerprint resolves + # the (keys, max_updated_at) signal from either the modern + # fingerprint or a legacy file's inline bodies, and returns None + # when the prefix carries NEITHER (a legacy large-scene file whose + # scenes fall after the prefix) so we fall through rather than + # assuming "no scenes". + disk_fp = _disk_scene_fingerprint(disk_meta_quick) if disk_meta_quick is not None else None + if disk_fp is not None: + disk_keys, disk_latest = disk_fp + if disk_keys: # Directional: only reload when disk is strictly ahead # of cache. Mirror master's subset comparison — cache # that is ahead of disk must NOT force a reload, or # un-persisted scene data is silently dropped. - disk_keys = {str(key) for key, value in disk_scenes.items() - if key and isinstance(value, dict)} cached_keys = _anchor_scene_record_keys(cached) - if disk_keys and not disk_keys.issubset(cached_keys): + if not disk_keys.issubset(cached_keys): return True # Same key set (or disk is subset): check the latest # updated_at timestamp. - def _max_updated(records): - latest = 0.0 - for record in records.values(): - if not isinstance(record, dict): - continue - try: - ua = float(record.get('updated_at') or 0) - except (TypeError, ValueError): - ua = 0.0 - if ua > latest: - latest = ua - return latest - if _max_updated(disk_scenes) > _anchor_scene_records_updated_at(cached): + if disk_latest > _anchor_scene_records_updated_at(cached): return True # disk has no scenes, cache does -> cache is ahead; keep it. else: - # Can't cheaply verify scene freshness from disk. The prefix - # reader may have failed while _persisted_message_count - # succeeded via index fallback. Fall through to the full - # metadata load so we don't serve stale scenes on the next - # equal-count inactive session path. Greptile P1. + # Can't cheaply verify scene freshness from disk (prefix carried + # neither fingerprint nor inline scenes, or the prefix read + # failed while _persisted_message_count succeeded via index + # fallback). Fall through to the full metadata load so we don't + # serve stale scenes on the next equal-count inactive path. + # Greptile P1. _scene_check_inconclusive = True else: # Cached session has no scene records. Check if disk has gained # the first scene record — without this the fast-path would miss # a newly persisted scene and return the stale cache. Greptile P1. disk_meta_quick = _persisted_session_meta_prefix(sid) - if disk_meta_quick is not None: - disk_scenes = disk_meta_quick.get('anchor_activity_scenes') or {} - if isinstance(disk_scenes, dict) and disk_scenes: + disk_fp = _disk_scene_fingerprint(disk_meta_quick) if disk_meta_quick is not None else None + if disk_fp is not None: + disk_keys, _disk_latest = disk_fp + if disk_keys: return True else: - # Prefix read failed (may still succeed via index fallback for - # message count). Mark inconclusive so we fall through to the - # full metadata comparison instead of returning False with - # stale cache. Greptile P1 (discussion_r3548650345). + # Prefix read failed / carried no scene signal (may still + # succeed via index fallback for message count). Mark + # inconclusive so we fall through to the full metadata + # comparison instead of returning False with stale cache. + # Greptile P1 (discussion_r3548650345). _scene_check_inconclusive = True if getattr(cached, 'active_stream_id', None) or getattr(cached, 'pending_user_message', None): # Active session: messages may be in flight; fall through to the @@ -3213,13 +3483,37 @@ def _cached_session_lags_disk(cached) -> bool: if disk_count is not None and disk_count > cached_count: return True if not getattr(cached, 'active_stream_id', None) and not getattr(cached, 'pending_user_message', None): + # #5854: disk_meta is a metadata-only stub whose scenes now live after + # `messages` and so are NOT materialized on it — read its scene freshness + # from the _anchor_scene_index fingerprint. `cached` is ALWAYS a full + # in-memory session (get_session never caches metadata-only stubs), and + # its records are mutated in place by the scene-persist path without + # refreshing the load-time fingerprint — so the cached side must read the + # REAL records (master parity), never the fingerprint, or a parity cache + # looks disk-behind after a scene write and forces a spurious full reload. + # + # A LEGACY stub (no anchor_scene_index fingerprint AND scenes not in the + # prefix) carries no scene signal at all — comparing it blind would miss + # a genuine disk-ahead scene change (stale worklog served). Full-load the + # sidecar once to compare real scene records; the next save() rewrites + # the modern layout so this legacy full-load doesn't recur. + if ( + getattr(disk_meta, '_loaded_metadata_only', False) + and getattr(disk_meta, '_anchor_scene_index', None) is None + ): + try: + disk_full = Session.load(sid) + except Exception: + disk_full = None + if disk_full is not None: + disk_meta = disk_full cached_scene_keys = _anchor_scene_record_keys(cached) - disk_scene_keys = _anchor_scene_record_keys(disk_meta) + disk_scene_keys = _session_scene_keys(disk_meta) if disk_scene_keys and not disk_scene_keys.issubset(cached_scene_keys): return True if ( disk_scene_keys - and _anchor_scene_records_updated_at(disk_meta) > _anchor_scene_records_updated_at(cached) + and _session_scene_updated_at(disk_meta) > _anchor_scene_records_updated_at(cached) ): return True return False @@ -3246,6 +3540,46 @@ def _persisted_message_count(sid) -> int | None: count = _parse_nonnegative_int(parsed.get('message_count')) if count is not None: return count + # #5854: a legacy sidecar (scenes-before-count) yields a prefix with + # no message_count and no modern anchor_scene_index. The _index.json + # count can lag behind external appends, so trusting it here risks + # under-reporting (and dropping an unsaved tail on cache-replace). + # Consult the authoritative legacy-facts cache (populated by a full + # load) before giving up; only return None (→ caller full-loads) on + # a miss. This keeps clean legacy sessions LRU-evictable without + # trusting a stale index. A MODERN prefix always carries + # message_count, so it returns above. + if 'anchor_scene_index' not in parsed: + _facts = _legacy_sidecar_facts_get(sid) + if _facts is not None: + cached_count = _parse_nonnegative_int(_facts.get('message_count')) + if cached_count is not None: + return cached_count + # Cache miss (never full-loaded yet, or the facts LRU evicted + # this entry). Returning None here would make the session + # non-evictable (the eviction check treats an unknown disk count + # as "do not evict"), which can let new_session() evict its own + # unsaved session and 404 the first send. Full-parse to get the + # authoritative count (Session.load re-caches the facts for + # legacy files under a TOCTOU-guarded signature). Re-verify the + # stat signature is stable across the parse so we never return a + # count from a file that was atomically replaced mid-read; retry + # boundedly on mismatch, then fall through to None. Bounded + # overall: the next save() rewrites the modern layout. + for _attempt in range(3): + sig_before = _sidecar_stat_signature(p) + try: + _full = Session.load(sid) + except Exception: + _full = None + sig_after = _sidecar_stat_signature(p) + if _full is None: + return None + if sig_before is not None and sig_before == sig_after: + return len(getattr(_full, 'messages', None) or []) + # File changed during the parse — the count is uncertain; + # retry with a fresh snapshot. + return None except Exception: # Fall through to the index-based fallback below. pass diff --git a/tests/test_issue5854_anchor_scene_split.py b/tests/test_issue5854_anchor_scene_split.py new file mode 100644 index 00000000..11da496e --- /dev/null +++ b/tests/test_issue5854_anchor_scene_split.py @@ -0,0 +1,346 @@ +"""Regression coverage for #5854: move anchor_activity_scenes out of the metadata prefix. + +Root of #4633 (server RSS climb → OOM) and #5839 (browser freeze): reasoning-heavy +sessions store 250–480 KB of `anchor_activity_scenes`, and `save()` serialized them +BEFORE `messages`, so the cheap metadata-prefix read (`_read_metadata_json_prefix`, +64 KB cap) overflowed and fell back to a full multi-MB `json.loads` on every +sidebar poll — the allocation churn that drove the RSS high-water crash. + +The fix persists a compact `anchor_scene_index` fingerprint ({scene_key: updated_at}) +in the metadata prefix (after `message_count`, before `messages`) and moves the full +scene bodies AFTER `messages`. The sidebar-poll freshness check reads the fingerprint; +the full-session GET still reads full bodies. These tests lock: + * the new serialization order (message_count < scene_index < messages < scenes), + * the cheap prefix stays small for a large-scene session (the #4633 fix), + * metadata-only stubs carry the fingerprint and a truthful message_count, + * full load round-trips scene bodies unchanged, + * the freshness comparison stays behavior-identical (keys + max updated_at), + * legacy-layout back-compat (scenes-before-messages) + the message_count recovery. +""" +import json + +import pytest + +import api.models as M + + +@pytest.fixture +def session_store(tmp_path, monkeypatch): + sdir = tmp_path / "sessions" + monkeypatch.setattr(M, "SESSION_DIR", sdir) + sdir.mkdir(parents=True, exist_ok=True) + return sdir + + +def _big_scenes(n_scenes=3, n_rows=80, base_ts=1000.0): + rows = [{"row_id": f"r{i}", "kind": "tool", "text": "X" * 3000, + "thinking": {"text": ""}, "tool": {"name": "read_file"}} for i in range(n_rows)] + return {f"scene{j}": {"version": 1, "updated_at": base_ts + j, + "scene": {"activity_rows": rows}} for j in range(n_scenes)} + + +def _make(session_store, sid="s1", scenes=None, msgs=None): + s = M.Session(session_id=sid, title="T", workspace=str(session_store.parent), + model="glm", messages=msgs if msgs is not None else [ + {"role": "user", "content": "hi"}, {"role": "assistant", "content": "yo"}]) + if scenes is not None: + s.anchor_activity_scenes = scenes + s.save() + return s + + +# ── Serialization order + cheap prefix (the #4633 fix) ────────────────────── + +def test_scene_bodies_serialize_after_messages_with_fingerprint_before(session_store): + _make(session_store, "s1", scenes=_big_scenes()) + raw = (session_store / "s1.json").read_text(encoding="utf-8") + ci = raw.find('"message_count"') + xi = raw.find('"anchor_scene_index"') + mi = raw.find('"messages"') + si = raw.find('"anchor_activity_scenes"') + assert -1 < ci < xi < mi < si, "order must be message_count < scene_index < messages < scenes" + + +def test_large_scene_session_keeps_cheap_prefix_small(session_store): + """The #4633 root: a 250KB+ scene session must NOT force a full parse.""" + _make(session_store, "s1", scenes=_big_scenes()) + prefix = M._read_metadata_json_prefix(session_store / "s1.json") + assert prefix is not None, "cheap prefix must succeed (not fall back to full load)" + assert len(prefix.encode("utf-8")) < 65536 + parsed = json.loads(prefix) + assert parsed["message_count"] == 2 + assert set(parsed["anchor_scene_index"]) == {"scene0", "scene1", "scene2"} + assert "anchor_activity_scenes" not in parsed, "full scenes must not be in the cheap prefix" + + +def test_metadata_only_stub_is_cheap_and_correct(session_store): + _make(session_store, "s1", scenes=_big_scenes()) + stub = M.Session.load_metadata_only("s1") + assert stub is not None + assert stub.messages == [] + assert stub._metadata_message_count == 2 + assert stub._loaded_metadata_only is True + # fingerprint available on the stub; full scenes not materialized + assert M._session_scene_keys(stub) == {"scene0", "scene1", "scene2"} + assert M._session_scene_updated_at(stub) == 1002.0 + + +def test_full_load_round_trips_scene_bodies(session_store): + scenes = _big_scenes(n_scenes=3, n_rows=80) + _make(session_store, "s1", scenes=scenes) + full = M.Session.load("s1") + assert set(full.anchor_activity_scenes) == {"scene0", "scene1", "scene2"} + assert len(full.anchor_activity_scenes["scene0"]["scene"]["activity_rows"]) == 80 + # bodies unchanged + assert full.anchor_activity_scenes["scene0"]["scene"]["activity_rows"][0]["text"] == "X" * 3000 + + +def test_save_reload_save_is_idempotent_on_scene_index(session_store): + _make(session_store, "s1", scenes=_big_scenes()) + full = M.Session.load("s1") + full.save(touch_updated_at=False) + prefix = json.loads(M._read_metadata_json_prefix(session_store / "s1.json")) + assert set(prefix["anchor_scene_index"]) == {"scene0", "scene1", "scene2"} + assert prefix["message_count"] == 2 + + +# ── Freshness comparison stays behavior-identical ─────────────────────────── + +def test_disk_scene_fingerprint_modern_and_legacy_equivalent(): + # modern: fingerprint present + modern = {"anchor_scene_index": {"a": 5.0, "b": 9.0}} + keys, latest = M._disk_scene_fingerprint(modern) + assert keys == {"a", "b"} and latest == 9.0 + # legacy: full bodies inline (scenes-before-messages layout) + legacy = {"anchor_activity_scenes": {"a": {"updated_at": 5.0}, "b": {"updated_at": 9.0}}} + keys2, latest2 = M._disk_scene_fingerprint(legacy) + assert keys2 == {"a", "b"} and latest2 == 9.0 + # neither present → None (must fall through, not assume "no scenes") + assert M._disk_scene_fingerprint({"title": "x"}) is None + # empty modern index present → ("no scenes", 0.0), NOT None + assert M._disk_scene_fingerprint({"anchor_scene_index": {}}) == (set(), 0.0) + + +def test_cached_lags_disk_detects_new_scene_and_newer_timestamp(session_store): + # persist a session with scene0@1000 + _make(session_store, "s1", scenes={"scene0": {"updated_at": 1000.0, "scene": {}}}) + # cached copy that is BEHIND (no scenes) → must lag + cached_behind = M.Session(session_id="s1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + assert M._cached_session_lags_disk(cached_behind) is True + # cached copy AT PARITY → must NOT lag + cached_parity = M.Session(session_id="s1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + cached_parity.anchor_activity_scenes = {"scene0": {"updated_at": 1000.0, "scene": {}}} + assert M._cached_session_lags_disk(cached_parity) is False + # cached copy AHEAD of disk (extra un-persisted scene) → must NOT force reload + cached_ahead = M.Session(session_id="s1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + cached_ahead.anchor_activity_scenes = {"scene0": {"updated_at": 1000.0, "scene": {}}, + "scene1": {"updated_at": 2000.0, "scene": {}}} + assert M._cached_session_lags_disk(cached_ahead) is False + + +def test_cached_lags_disk_detects_newer_scene_updated_at(session_store): + _make(session_store, "s1", scenes={"scene0": {"updated_at": 2000.0, "scene": {}}}) + cached_stale = M.Session(session_id="s1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + cached_stale.anchor_activity_scenes = {"scene0": {"updated_at": 1000.0, "scene": {}}} + assert M._cached_session_lags_disk(cached_stale) is True + + +def test_parity_cache_not_forced_reload_when_prefix_read_fails(session_store): + """MUST-FIX regression (Opus round-1): when the cheap prefix read fails + (oversized metadata field > 64KB), the slow path must compare the CACHED + side against its REAL records, not a stale load-time fingerprint. A cache at + exact parity with disk must NOT be judged as lagging (which would force a + spurious full reload of the heavy session this PR optimizes). + + Simulate an oversized metadata prefix by giving the session a large + `compression_anchor_summary` so `_read_metadata_json_prefix` overflows and + the slow path (Session.load_metadata_only) is taken. + """ + scenes = {"scene0": {"updated_at": 1000.0, "scene": {}}} + s = M.Session(session_id="p1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + s.anchor_activity_scenes = scenes + s.compression_anchor_summary = "Z" * 80000 # push metadata prefix > 64KB + s.save() + # Sanity: the cheap prefix genuinely fails for this file (slow path taken). + assert M._read_metadata_json_prefix(session_store / "p1.json") is None + # A cached full session at exact parity (same scene key + updated_at) whose + # load-time fingerprint is stale/empty must NOT be judged as lagging. + cached_parity = M.Session(session_id="p1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + cached_parity.anchor_activity_scenes = {"scene0": {"updated_at": 1000.0, "scene": {}}} + cached_parity._anchor_scene_index = {} # stale/empty load-time fingerprint + cached_parity.compression_anchor_summary = "Z" * 80000 + assert M._cached_session_lags_disk(cached_parity) is False + # But a genuinely-behind cache (disk has a newer scene) still reloads. + cached_behind = M.Session(session_id="p1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + cached_behind.anchor_activity_scenes = {} + assert M._cached_session_lags_disk(cached_behind) is True + + +# ── Legacy back-compat ────────────────────────────────────────────────────── + +def _write_legacy(session_store, sid, scenes, message_count, with_msgs=True): + """Emulate a pre-#5854 sidecar: scenes BEFORE message_count/messages, no fingerprint.""" + doc = {"session_id": sid, "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "anchor_activity_scenes": scenes, + "message_count": message_count, + "messages": [{"role": "user", "content": "hi"}] if with_msgs else []} + (session_store / f"{sid}.json").write_text(json.dumps(doc, indent=2), encoding="utf-8") + + +def test_legacy_layout_metadata_only_reads_count_and_scene_keys(session_store): + _write_legacy(session_store, "leg1", {"sc0": {"updated_at": 5.0}}, message_count=7) + stub = M.Session.load_metadata_only("leg1") + assert stub is not None + # legacy message_count survives (it's before scenes in this legacy doc, but even + # when after, the recovery path covers it — see next test) + assert stub._metadata_message_count == 7 + + +def test_legacy_large_scene_missing_count_recovers_via_full_load(session_store): + """The Codex CORE hazard: legacy scenes-before-count + large scenes. The cheap + prefix stops at scenes and misses message_count; with no index count, a naive + stub would report 0 msgs and the session could vanish. Must recover via full load.""" + big = _big_scenes(n_scenes=3, n_rows=80) # >64KB so prefix stops at scenes + # legacy order: scenes FIRST, then message_count AFTER (pre-#5854 shape) + doc = {"session_id": "leg2", "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "anchor_activity_scenes": big, + "message_count": 5, + "messages": [{"role": "user", "content": f"m{i}"} for i in range(5)]} + (session_store / "leg2.json").write_text(json.dumps(doc, indent=2), encoding="utf-8") + # No _index.json entry → must fall back to full load and get the real count. + stub = M.Session.load_metadata_only("leg2") + assert stub is not None + assert (stub._metadata_message_count or len(stub.messages)) == 5, "must not report 0 messages" + + +def test_legacy_missing_count_uses_authoritative_not_stale_index(session_store): + """Codex round-2/3 CORE: on a legacy sidecar whose prefix lacks message_count, + _persisted_message_count must NOT trust a (possibly lagging) _index.json. It + full-parses once to get the AUTHORITATIVE count (and caches it), rather than + under-reporting from a stale index or returning None (which would pin the + session in memory as non-evictable and can 404 the first send).""" + big = _big_scenes(n_scenes=3, n_rows=80) + doc = {"session_id": "leg3", "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "anchor_activity_scenes": big, "message_count": 4, + "messages": [{"role": "user", "content": f"m{i}"} for i in range(4)]} + (session_store / "leg3.json").write_text(json.dumps(doc, indent=2), encoding="utf-8") + M._LEGACY_SIDECAR_FACTS.clear() + # Prefix stops at scenes (no count, no fingerprint) → must return the real + # authoritative count (4), never None and never a stale index value. + assert M._persisted_message_count("leg3") == 4 + + +def test_legacy_disk_ahead_scene_detected_via_full_load(session_store): + """Codex round-2 SILENT: a legacy stub carries no scene signal (no fingerprint, + scenes after the prefix). The slow path must full-load to detect a genuine + disk-ahead scene change instead of serving a stale cached worklog.""" + # legacy file WITH a prefix-captured count (small enough) but scenes present; + # emulate the stub-with-no-scene-signal by writing legacy layout. + doc = {"session_id": "leg4", "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 2, + "anchor_activity_scenes": {"sceneNEW": {"updated_at": 5000.0, "scene": {}}}, + "messages": [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "yo"}]} + (session_store / "leg4.json").write_text(json.dumps(doc, indent=2), encoding="utf-8") + # cached copy is BEHIND: it has an older/different scene set. + cached_behind = M.Session(session_id="leg4", title="Legacy", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}]) + cached_behind.anchor_activity_scenes = {} + assert M._cached_session_lags_disk(cached_behind) is True + + +def test_fully_loaded_session_never_uses_stale_fingerprint(session_store): + """Codex round-2 SILENT: the scene helpers must read REAL records for a + fully-loaded session, never its (possibly stale) load-time _anchor_scene_index.""" + full = M.Session(session_id="f1", title="T", workspace=str(session_store.parent), + model="glm", messages=[{"role": "user", "content": "hi"}]) + full.anchor_activity_scenes = {"sceneA": {"updated_at": 30.0}} + full._anchor_scene_index = {"sceneOLD": 10.0} # stale load-time fingerprint + # _loaded_metadata_only is falsy → must read the REAL records, not the index. + assert M._session_scene_keys(full) == {"sceneA"} + assert M._session_scene_updated_at(full) == 30.0 + # A metadata-only stub, by contrast, DOES use its fingerprint. + stub = M.Session(session_id="f2", title="T", workspace=str(session_store.parent), + model="glm", messages=[]) + stub._loaded_metadata_only = True + stub.anchor_activity_scenes = {} + stub._anchor_scene_index = {"sceneB": 42.0} + assert M._session_scene_keys(stub) == {"sceneB"} + assert M._session_scene_updated_at(stub) == 42.0 + + +def _write_legacy_large(session_store, sid, n_msgs=5): + """Write a legacy large-scene sidecar (scenes before count, >64KB prefix).""" + big = _big_scenes(n_scenes=3, n_rows=80) + doc = {"session_id": sid, "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "anchor_activity_scenes": big, + "message_count": n_msgs, + "messages": [{"role": "user", "content": f"m{i}"} for i in range(n_msgs)]} + (session_store / f"{sid}.json").write_text(json.dumps(doc, indent=2), encoding="utf-8") + + +def test_legacy_large_scene_not_reparsed_on_every_read(session_store, monkeypatch): + """Codex round-3 SILENT: an unchanged legacy large-scene sidecar must be + full-parsed at most once (cached by stat signature), not on every poll — + otherwise the #4633 churn returns for legacy files that are never re-saved.""" + M._LEGACY_SIDECAR_FACTS.clear() + _write_legacy_large(session_store, "legcache", n_msgs=5) + # Count real full-loads via a spy on the raw file read inside load(). + calls = {"n": 0} + real_load = M.Session.load.__func__ + + def _counting_load(cls, sid, *a, **k): + if sid == "legcache": + calls["n"] += 1 + return real_load(cls, sid, *a, **k) + + monkeypatch.setattr(M.Session, "load", classmethod(_counting_load)) + s1 = M.Session.load_metadata_only("legcache") + s2 = M.Session.load_metadata_only("legcache") + s3 = M.Session.load_metadata_only("legcache") + assert s1._metadata_message_count == 5 + assert s2._metadata_message_count == 5 + assert s3._metadata_message_count == 5 + assert calls["n"] == 1, f"legacy file full-loaded {calls['n']}x; must be cached after the first" + + +def test_legacy_session_stays_evictable(session_store): + """Codex round-3 SILENT: a clean legacy session must remain LRU-evictable — + _persisted_message_count must yield its authoritative count (via the facts + cache) rather than None (which would pin it in memory forever).""" + M._LEGACY_SIDECAR_FACTS.clear() + _write_legacy_large(session_store, "legevict", n_msgs=3) + # Prime the facts cache via a full load (mirrors get_session materializing it). + M.Session.load("legevict") + assert M._persisted_message_count("legevict") == 3 + + +def test_legacy_count_recovers_on_facts_cache_miss(session_store): + """Codex round-3 CORE: even with the legacy-facts cache EMPTY (never loaded, + or LRU-evicted), _persisted_message_count must not return None — it full-parses + once and returns the authoritative count, so the session stays evictable and a + fresh new_session() can't evict its own unsaved session and 404 the first send.""" + _write_legacy_large(session_store, "legmiss", n_msgs=6) + M._LEGACY_SIDECAR_FACTS.clear() # simulate cold / evicted facts cache + assert M._persisted_message_count("legmiss") == 6 + # and the miss re-populated the cache for the next call + assert M._legacy_sidecar_facts_get("legmiss") is not None