Merge remote-tracking branch 'origin/master' into paperclip/pr-6222

This commit is contained in:
Frank Song committed 2026-07-20 09:30:56 +08:00
commit 80d40d6531
5 files changed
+176 -26

No files matched your search

+2
View File
@@ -15,6 +15,8 @@
### Fixed
- **Content search no longer evicts your open sessions from the cache.** A content search scans many sessions, and each scan used to promote/cold-load sessions into the in-memory LRU — evicting the sessions you actually have open. Search now reads through a scan-only accessor that keeps the same disk-freshness, journal-recovery, and state.db reconciliation (so recent content stays searchable) but doesn't disturb the working-set cache. Thanks @ai-ag2026. (#6084, #6083)
- **Watching the same terminal from two tabs no longer splits the output stream.** Terminal output is now broadcast to every subscribed viewer instead of draining from one shared queue, so a second tab (or a reconnect) sees the full stream rather than a random half. SSE responses carry event `id:`s and a reconnecting viewer replays only events after its `Last-Event-ID` cursor (fresh viewers still get the full bounded backlog). Thanks @ai-ag2026. (#5836)
- **A turn that ends in an error no longer shows impossible "still running" tool work or a wrong duration.** On a terminal error, the settled (and reloaded) Worklog now freezes the turn's duration before cleanup and seals any still-projected tool rows as done, so a settled/reloaded transcript can't disagree with the terminal state. Successful and cancel paths are unchanged. Thanks @webtecnica. (#6323, #6309)
+44 -24
View File
@@ -4404,17 +4404,33 @@ def _evict_sessions_over_cap(cap: int | None = None) -> int:
return evicted
def get_session(sid, metadata_only=False):
"""Load a session, optionally with metadata only (skipping the messages array).
def get_session_for_scan(sid):
"""Read a session for a one-pass scan without disturbing the LRU.
Metadata-only loads intentionally do not populate the full-session cache.
Otherwise a later full load could return a compact object with an empty
messages list. Use this when you only need compact() metadata and not the
actual message history (e.g., for fast sidebar switching).
The scan uses the full canonical resolver so stale sidecars, pending journal
recovery, and newer state.db transcript rows remain searchable. It only
differs from ``get_session`` in cache policy: no hit promotion, no cold-load
insertion, and therefore no scan-triggered eviction.
"""
try:
return _resolve_session(sid, promote_cache=False, cache_on_miss=False)
except Exception:
logger.debug("scan load failed for session %s", sid, exc_info=True)
return None
def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_miss=True):
"""Resolve a session through the canonical freshness/recovery path.
``get_session_for_scan`` shares this resolver with normal reads so that
content search sees the same disk freshness, journal-retry, and state.db
recovery behavior. Its policy merely suppresses LRU promotion and cold-load
caching; reconciliation still updates an already-resident entry when normal
resolution replaces stale contents.
"""
with LOCK:
cached = SESSIONS.get(sid)
if cached is not None:
if cached is not None and promote_cache:
SESSIONS.move_to_end(sid) # LRU: mark as recently used
if cached is not None:
# Defensive cache ownership check: compression/continuation and recovery
@@ -4438,12 +4454,12 @@ def get_session(sid, metadata_only=False):
disk_session = Session.load(sid)
with LOCK:
SESSIONS[sid] = disk_session
SESSIONS.move_to_end(sid)
if promote_cache:
SESSIONS.move_to_end(sid)
cached = disk_session
except Exception:
logger.debug(
"cached session disk-freshness check failed for session %s",
sid, exc_info=True,
"cached session disk-freshness check failed for session %s", sid, exc_info=True,
)
if not metadata_only and _inactive_cache_tail_needs_disk_check(cached):
try:
@@ -4451,28 +4467,26 @@ def get_session(sid, metadata_only=False):
if _cache_has_stale_unsaved_user_tail(cached, disk_session):
with LOCK:
SESSIONS[sid] = disk_session
SESSIONS.move_to_end(sid)
if promote_cache:
SESSIONS.move_to_end(sid)
cached = disk_session
except Exception:
logger.debug(
"stale cached user-tail check failed for session %s",
sid, exc_info=True,
"stale cached user-tail check failed for session %s", sid, exc_info=True,
)
if not metadata_only and _session_has_pending_journal_retry(cached):
try:
_try_retry_journal_recovery_in_place(cached)
except Exception:
logger.debug(
"lazy journal-retry failed on cache hit for session %s",
sid, exc_info=True,
"lazy journal-retry failed on cache hit for session %s", sid, exc_info=True,
)
if not metadata_only:
try:
_sync_sidecar_from_state_db_if_newer(cached)
except Exception:
logger.debug(
"state.db newer-sidecar sync failed on cache hit for session %s",
sid, exc_info=True,
"state.db newer-sidecar sync failed on cache hit for session %s", sid, exc_info=True,
)
return cached
if metadata_only:
@@ -4482,10 +4496,12 @@ def get_session(sid, metadata_only=False):
else:
s = Session.load(sid)
if s:
with LOCK:
SESSIONS[sid] = s
SESSIONS.move_to_end(sid)
_evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved)
if cache_on_miss:
with LOCK:
SESSIONS[sid] = s
if promote_cache:
SESSIONS.move_to_end(sid)
_evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved)
if not metadata_only:
try:
synced_from_state = _sync_sidecar_from_state_db_if_newer(s)
@@ -4499,14 +4515,13 @@ def get_session(sid, metadata_only=False):
_try_retry_journal_recovery_in_place(s)
except Exception:
logger.debug(
"lazy journal-retry failed on cold load for session %s",
sid, exc_info=True,
"lazy journal-retry failed on cold load for session %s", sid, exc_info=True,
)
# If repair had to bail because the per-session lock was held,
# do not pin the still-stale sidecar in the LRU cache forever.
# Leaving it cached would prevent future get_session() calls from
# re-entering the cache-miss repair path after the lock holder exits.
if not repaired and (len(s.messages) == 0
if cache_on_miss and not repaired and (len(s.messages) == 0
and s.pending_user_message
and s.active_stream_id
and s.active_stream_id not in _active_stream_ids()):
@@ -4519,6 +4534,11 @@ def get_session(sid, metadata_only=False):
raise KeyError(sid)
def get_session(sid, metadata_only=False):
"""Load a session, optionally with metadata only (skipping messages)."""
return _resolve_session(sid, metadata_only=metadata_only)
_COMPRESSION_RECOVERY_PROFILE_UNSET = object()
+7 -1
View File
@@ -9307,6 +9307,7 @@ def _keep_latest_messaging_session_per_source(
from api.models import (
Session,
get_session,
get_session_for_scan,
find_compression_recovery_session,
get_session_for_file_ops,
new_session,
@@ -16777,7 +16778,12 @@ def _handle_sessions_search(handler, parsed):
continue
if content_search:
try:
sess = get_session(s["session_id"])
# Scan accessor, not get_session(): a content search walks every
# session, and routing that through the LRU would evict the
# user's working set on every keystroke-debounced search.
sess = get_session_for_scan(s["session_id"])
if sess is None:
continue
msgs = sess.messages[:depth] if depth else sess.messages
for m in msgs:
c = _session_search_message_text(m)
@@ -413,3 +413,122 @@ def test_stale_unsaved_shell_with_draft_stays_resident(isolated_session_env):
assert _session_is_evictable(shell) is False, (
"a stale shell with an active composer draft must stay resident"
)
def test_content_search_scan_does_not_evict_the_working_set(isolated_session_env):
"""A content search must not push the user's open sessions out of the cache.
/api/sessions/search?content=1 walks EVERY session. Routing that through
get_session() inserted each one into the LRU and marked it recently-used, so
a single search over an install with more sessions than the cap flushed the
whole cache — the classic buffer-pool scan-pollution problem. The sessions
the user actually had open were the ones evicted.
get_session_for_scan() reads without promoting or inserting, so a scan is
transparent to the LRU.
"""
from api import config as _cfg
from api.config import SESSIONS
from api.models import get_session, get_session_for_scan
_cfg.SESSIONS_MAX = 5
working = []
for i in range(4):
s = _make_persisted_session(900 + i)
get_session(s.session_id) # the user opens it -> legitimately cached
working.append(s.session_id)
corpus = [_make_persisted_session(i).session_id for i in range(60)]
for sid in corpus: # what the content search does
assert get_session_for_scan(sid) is not None
for sid in working:
assert sid in SESSIONS, "a scan evicted the user's working set"
assert not any(sid in SESSIONS for sid in corpus), "the scan polluted the LRU"
assert len(SESSIONS) <= _cfg.SESSIONS_MAX
def test_scan_accessor_reuses_resident_sessions_without_promoting(isolated_session_env):
"""A scan hit must reuse the cached object but must not refresh its recency."""
from api import config as _cfg
from api.config import SESSIONS
from api.models import get_session, get_session_for_scan
_cfg.SESSIONS_MAX = 50
first = _make_persisted_session(801)
second = _make_persisted_session(802)
get_session(first.session_id)
get_session(second.session_id) # second is now the most-recent entry
order_before = list(SESSIONS.keys())
scanned = get_session_for_scan(first.session_id)
assert scanned is SESSIONS[first.session_id], "scan should reuse the resident object"
assert list(SESSIONS.keys()) == order_before, "scan must not promote in the LRU"
def test_content_search_scan_recovers_newer_state_db_without_lru_churn(
isolated_session_env, monkeypatch,
):
"""The real search path must find state.db-only recovery text without LRU churn."""
from types import SimpleNamespace
from urllib.parse import urlparse
from api import models, routes
from api.config import SESSIONS
working = _make_persisted_session(950)
stale = _make_persisted_session(
951,
messages=[{"role": "user", "content": "old prompt", "timestamp": 100.0}],
)
stale.active_stream_id = "dead-stream"
stale.pending_user_message = "recover me"
stale.pending_started_at = 102.0
stale.save()
_insert(working)
_insert(stale)
order_before = list(SESSIONS.keys())
size_before = len(SESSIONS)
recovered = [
{"role": "user", "content": "old prompt", "timestamp": 100.0},
{"role": "user", "content": "recover me", "timestamp": 102.0},
{"role": "assistant", "content": "state-db-only needle", "timestamp": 103.0},
]
monkeypatch.setattr(
models,
"get_state_db_session_summary",
lambda sid, profile=None: {"message_count": len(recovered), "last_message_at": 103.0},
)
monkeypatch.setattr(
models,
"get_state_db_session_messages",
lambda sid, **kwargs: list(recovered),
)
monkeypatch.setattr(
routes,
"all_sessions",
lambda: [{"session_id": stale.session_id, "title": stale.title, "profile": "default"}],
)
monkeypatch.setattr(routes, "load_settings", lambda: {"api_redact_enabled": False})
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "default")
captured = {}
monkeypatch.setattr(
routes,
"j",
lambda handler, payload, status=200, extra_headers=None: captured.update(
payload=payload, status=status,
),
)
routes._handle_sessions_search(
SimpleNamespace(),
urlparse("/api/sessions/search?q=needle&content=1&depth=0"),
)
assert captured["status"] == 200
assert captured["payload"]["count"] == 1
assert captured["payload"]["sessions"][0]["session_id"] == stale.session_id
assert list(SESSIONS.keys()) == order_before, "scan recovery must not promote the LRU"
assert len(SESSIONS) == size_before, "scan recovery must not insert or evict cache entries"
@@ -40,8 +40,11 @@ def _run_search(query):
captured["status"] = status
captured["payload"] = payload
# The content search reads through get_session_for_scan (the LRU-transparent
# scan accessor), not get_session — patching the wrong one here would make
# the depth assertions pass vacuously on an empty result set.
with patch("api.routes.all_sessions", return_value=list(sessions_meta)), patch(
"api.routes.get_session", return_value=session
"api.routes.get_session_for_scan", return_value=session
), patch("api.profiles.get_active_profile_name", return_value="default"), patch(
"api.routes.j", side_effect=fake_j
):