From 6335afa4793ec4c06f5b748abaa388f57906ffcc Mon Sep 17 00:00:00 2001 From: ArkNill <48707894+ArkNill@users.noreply.github.com> Date: Wed, 20 May 2026 14:59:36 +0900 Subject: [PATCH 1/5] refactor(composition): extract _classify_messages_delta helper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pull the per-message classification loop out of _reconstruct_and_classify into a reusable helper that takes any message list and returns (totals, tool_call_counts, tool_byte_totals, thinking_count). No behavior change — final-state classification still happens once over the fully reconstructed `accumulated`. Sets up the incremental cache work for #16 by giving the fold path a single function to call on each delta of new messages. --- src/llm_relay/proxy/composition.py | 54 ++++++++++++++++++------------ 1 file changed, 33 insertions(+), 21 deletions(-) diff --git a/src/llm_relay/proxy/composition.py b/src/llm_relay/proxy/composition.py index 1ee2d2f..70aeb4e 100644 --- a/src/llm_relay/proxy/composition.py +++ b/src/llm_relay/proxy/composition.py @@ -166,6 +166,37 @@ def _count_thinking_blocks(msg: dict) -> int: # ── Reconstruction ── +def _classify_messages_delta( + messages: List[dict], +) -> Tuple[Dict[str, int], Dict[str, int], Dict[str, Dict[str, int]], int]: + """Classify a list of messages into category byte totals + tool details. + + Returns (totals, tool_call_counts, tool_byte_totals, thinking_count) where each + value is the contribution from `messages` alone. Callers either use the result + directly (final-state classification) or fold it into a running state + (incremental classification). + """ + totals: Dict[str, int] = {cat: 0 for cat in CATEGORIES} + tool_call_counts: Dict[str, int] = defaultdict(int) + tool_byte_totals: Dict[str, Dict[str, int]] = defaultdict(lambda: {"use": 0, "result": 0}) + thinking_count = 0 + + for msg in messages: + sizes = _classify_message(msg) + for cat in CATEGORIES: + totals[cat] += sizes.get(cat, 0) + for name in _extract_tool_names(msg): + tool_call_counts[name] += 1 + for name, bytes_info in _extract_tool_use_bytes(msg).items(): + if name == "__result__": + continue + tool_byte_totals[name]["use"] += bytes_info["use"] + tool_byte_totals[name]["result"] += bytes_info["result"] + thinking_count += _count_thinking_blocks(msg) + + return totals, dict(tool_call_counts), dict(tool_byte_totals), thinking_count + + def _reconstruct_and_classify( turns_data: List[dict], ) -> Tuple[Dict[str, int], int, Dict[str, int], Dict[str, int], Dict[str, Dict[str, int]], int]: @@ -196,35 +227,16 @@ def _reconstruct_and_classify( else: accumulated.extend(messages) - # Track reads and tool calls for msg in messages: for target in _extract_read_targets(msg): read_counts[target] += 1 - # Classify accumulated context + extract tool details - totals: Dict[str, int] = {cat: 0 for cat in CATEGORIES} - tool_call_counts: Dict[str, int] = defaultdict(int) - tool_byte_totals: Dict[str, Dict[str, int]] = defaultdict(lambda: {"use": 0, "result": 0}) - thinking_count = 0 - - for msg in accumulated: - sizes = _classify_message(msg) - for cat in CATEGORIES: - totals[cat] += sizes.get(cat, 0) - # Per-tool counting - for name in _extract_tool_names(msg): - tool_call_counts[name] += 1 - for name, bytes_info in _extract_tool_use_bytes(msg).items(): - if name == "__result__": - continue - tool_byte_totals[name]["use"] += bytes_info["use"] - tool_byte_totals[name]["result"] += bytes_info["result"] - thinking_count += _count_thinking_blocks(msg) + totals, tool_call_counts, tool_byte_totals, thinking_count = _classify_messages_delta(accumulated) total_bytes = sum(totals.values()) dupes = {k: v for k, v in read_counts.items() if v > 1} - return totals, total_bytes, dupes, dict(tool_call_counts), dict(tool_byte_totals), thinking_count + return totals, total_bytes, dupes, tool_call_counts, tool_byte_totals, thinking_count # ── Public API ── From ab9163a72a828995864a2fb77fc2733c0e338d53 Mon Sep 17 00:00:00 2001 From: ArkNill <48707894+ArkNill@users.noreply.github.com> Date: Wed, 20 May 2026 15:00:22 +0900 Subject: [PATCH 2/5] fix(composition): reset read_counts on compaction (storage_mode=full) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A compaction event (storage_mode="full") replaces the accumulated context wholesale, so reads that happened in pre-compaction turns are no longer present in the context. Previously read_counts kept accumulating across the compaction boundary, which made duplicate_reads report files that weren't even in the current state as duplicates of post-compaction reads. Reset read_counts together with `accumulated` on each "full" turn so duplicate_reads reflects only files still in the live context. Behavior visible in two places: - /api/v1/turns and /api/v1/display: duplicate_reads dict no longer carries entries for files only read pre-compaction - duplicate_read_count drops accordingly; duplicate_read_warning may flip off if the only triggering file was pre-compaction Existing test_duplicate_read_detection still passes — its first turn is the only "full" turn, so the reset has no effect there. New test_compaction_resets_read_counts covers the pre/post-compaction separation. --- src/llm_relay/proxy/composition.py | 4 ++++ tests/test_api/test_composition.py | 27 +++++++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/llm_relay/proxy/composition.py b/src/llm_relay/proxy/composition.py index 70aeb4e..cd0e0b5 100644 --- a/src/llm_relay/proxy/composition.py +++ b/src/llm_relay/proxy/composition.py @@ -224,6 +224,10 @@ def _reconstruct_and_classify( if storage_mode == "full": accumulated = list(messages) + # Compaction replaces the accumulated context, so prior reads + # are no longer present. Reset read_counts so duplicate_reads + # reflects only files still in the current accumulated state. + read_counts = defaultdict(int) else: accumulated.extend(messages) diff --git a/tests/test_api/test_composition.py b/tests/test_api/test_composition.py index 8b7e1dd..95c5857 100644 --- a/tests/test_api/test_composition.py +++ b/tests/test_api/test_composition.py @@ -178,6 +178,33 @@ def test_duplicate_read_detection(self): assert "/tmp/test.py" in dupes assert dupes["/tmp/test.py"] == 3 + def test_compaction_resets_read_counts(self): + """Pre-compaction reads should not appear in duplicate_reads after a + storage_mode='full' turn — those files are no longer in the accumulated + context, so counting them as duplicates of post-compaction reads is wrong. + """ + read_a = { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "/a.py"}}, + ], + } + read_b = { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "t2", "name": "Read", "input": {"file_path": "/b.py"}}, + ], + } + turns = [ + _make_turn(1, "full", [read_a]), + _make_turn(2, "delta", [read_a]), # /a.py reads = 2 (pre-compaction) + _make_turn(3, "full", [read_b]), # compaction → reset + _make_turn(4, "delta", [read_b]), # /b.py reads = 2 (post-compaction) + ] + _, _, dupes, _tc, _tb, _thc = _reconstruct_and_classify(turns) + assert "/a.py" not in dupes, "pre-compaction reads should not survive" + assert dupes.get("/b.py") == 2 + def test_invalid_json_skipped(self): turns = [ {"turn_number": 1, "storage_mode": "full", "request_messages": "not valid json"}, From 091e13a19ed9639b724ced23a6d2f8dfd4c22326 Mon Sep 17 00:00:00 2001 From: ArkNill <48707894+ArkNill@users.noreply.github.com> Date: Wed, 20 May 2026 15:04:30 +0900 Subject: [PATCH 3/5] perf(composition): incremental cache fold for both analyze functions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous cache for analyze_session_composition and analyze_session_composition_per_turn keyed on (session_id, max_turn). Any new turn invalidated the entry, forcing a full re-walk of every prior turn from the conversation_turns table. For long-running sessions, this made cache lookups effectively O(n²) in turn count — RSS could balloon to 10GB+ within 30 minutes of normal multi-agent traffic, and /api/v1/turns responses stretched to tens of seconds. This commit switches both caches to dataclass-backed state keyed on session_id alone: _CompositionState — accumulated, totals, tool_call_counts, tool_byte_totals, thinking_count, read_counts, max_turn_processed _PerTurnState — accumulated, turn_results, max_turn_processed On each call: 1. Probe MAX(turn_number) to detect new arrivals 2. If cache matches max_turn exactly, build result from state and return 3. Otherwise fetch only WHERE turn_number > max_turn_processed 4. Fold those delta turns into state in place 5. Cache the updated state Compaction handling (storage_mode="full") matches the semantics of _reconstruct_and_classify: accumulated/totals/tool counts/thinking_count all reset, then re-absorb the snapshot. read_counts also reset (the behavior change introduced in the prior commit). Failure handling: any exception during incremental fold falls back to a full rebuild via _full_rebuild_composition_state / _full_rebuild_per_turn_state. A max_turn regression (e.g., turn deletion) also triggers full rebuild with a warning log — the prior cache state is presumed stale. Result identity is no longer preserved across cache-hit calls (the result dict is rebuilt from state each time). Updated test_cache_hit to assert value equality, which is the actual contract. Tests: existing 64 composition tests pass; full suite 522 pass. Resolves #16. --- src/llm_relay/proxy/composition.py | 327 ++++++++++++++++++++++++----- tests/test_api/test_composition.py | 5 +- 2 files changed, 280 insertions(+), 52 deletions(-) diff --git a/src/llm_relay/proxy/composition.py b/src/llm_relay/proxy/composition.py index cd0e0b5..fa425b4 100644 --- a/src/llm_relay/proxy/composition.py +++ b/src/llm_relay/proxy/composition.py @@ -15,6 +15,7 @@ import re import sqlite3 from collections import defaultdict +from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple logger = logging.getLogger("llm-relay.composition") @@ -28,8 +29,34 @@ "thinking_overhead", ] -# In-memory cache: session_id -> (max_turn_number, result_dict) -_cache: Dict[str, Tuple[int, dict]] = {} + +@dataclass +class _CompositionState: + """Cumulative state for incremental composition analysis. + + Built by folding turns one at a time. On storage_mode="full" (compaction), + accumulated/totals/tool counts/thinking_count/read_counts all reset and + then absorb the snapshot. On "delta", everything is appended to. + """ + max_turn_processed: int = 0 + accumulated: List[dict] = field(default_factory=list) + read_counts: Dict[str, int] = field(default_factory=dict) + totals: Dict[str, int] = field(default_factory=lambda: {cat: 0 for cat in CATEGORIES}) + tool_call_counts: Dict[str, int] = field(default_factory=dict) + tool_byte_totals: Dict[str, Dict[str, int]] = field(default_factory=dict) + thinking_count: int = 0 + + +@dataclass +class _PerTurnState: + """Cumulative state for incremental per-turn composition analysis.""" + max_turn_processed: int = 0 + accumulated: List[dict] = field(default_factory=list) + turn_results: List[dict] = field(default_factory=list) + + +# In-memory cache: session_id -> _CompositionState +_cache: Dict[str, _CompositionState] = {} # ── Classification ── @@ -243,6 +270,100 @@ def _reconstruct_and_classify( return totals, total_bytes, dupes, tool_call_counts, tool_byte_totals, thinking_count +# ── Incremental folding ── + + +def _parse_turn_messages(row: dict) -> Optional[List[dict]]: + """Parse request_messages from a turn row. Returns None if unusable.""" + req_json = row["request_messages"] + if not req_json: + return None + try: + messages = req_json if isinstance(req_json, list) else json.loads(req_json) + except (json.JSONDecodeError, TypeError): + return None + if not isinstance(messages, list): + return None + return messages + + +def _fold_delta_into_composition_state( + state: _CompositionState, + new_turns: List[dict], +) -> None: + """Fold delta turns into the state in place. + + Matches the semantics of _reconstruct_and_classify: storage_mode="full" + resets accumulated/totals/tool_calls/tool_bytes/thinking/read_counts and + re-absorbs the snapshot; other modes append. + """ + for row in new_turns: + turn_number = row["turn_number"] + storage_mode = row["storage_mode"] + messages = _parse_turn_messages(row) + if messages is None: + state.max_turn_processed = turn_number + continue + + if storage_mode == "full": + state.accumulated = list(messages) + state.read_counts = {} + state.totals = {cat: 0 for cat in CATEGORIES} + state.tool_call_counts = {} + state.tool_byte_totals = {} + state.thinking_count = 0 + else: + state.accumulated.extend(messages) + + d_totals, d_calls, d_bytes, d_thinking = _classify_messages_delta(messages) + for cat in CATEGORIES: + state.totals[cat] += d_totals[cat] + for name, count in d_calls.items(): + state.tool_call_counts[name] = state.tool_call_counts.get(name, 0) + count + for name, info in d_bytes.items(): + existing = state.tool_byte_totals.setdefault(name, {"use": 0, "result": 0}) + existing["use"] += info["use"] + existing["result"] += info["result"] + state.thinking_count += d_thinking + + for msg in messages: + for target in _extract_read_targets(msg): + state.read_counts[target] = state.read_counts.get(target, 0) + 1 + + state.max_turn_processed = turn_number + + +def _build_composition_result_from_state(state: _CompositionState) -> dict: + """Render a result dict from accumulated state.""" + dupes = {k: v for k, v in state.read_counts.items() if v > 1} + return _build_composition_result( + state.totals, + dupes=dupes, + tool_calls=state.tool_call_counts, + tool_bytes=state.tool_byte_totals, + thinking_count=state.thinking_count, + ) + + +def _full_rebuild_composition_state( + conn: sqlite3.Connection, + session_id: str, +) -> Optional[_CompositionState]: + """Fetch all turns for a session and fold them into a fresh state.""" + turns = conn.execute( + """SELECT turn_number, storage_mode, request_messages + FROM conversation_turns + WHERE session_id = ? + ORDER BY turn_number ASC""", + (session_id,), + ).fetchall() + if not turns: + return None + state = _CompositionState() + _fold_delta_into_composition_state(state, [dict(t) for t in turns]) + return state + + # ── Public API ── @@ -252,56 +373,65 @@ def analyze_session_composition( ) -> Optional[dict]: """Analyze context composition for a session. Returns cached result if available. - Returns dict with: categories, total_bytes, est_tokens, snr, duplicate_read_count + Uses an incremental cache: when new turns arrive, only those past + max_turn_processed are fetched and folded into existing state. Compaction + events (storage_mode="full") reset accumulated/totals/read_counts then + re-absorb the snapshot. On any fold exception, falls back to a full + rebuild so a corrupted increment can't poison the cache permanently. + + Returns dict with: categories, total_bytes, est_tokens, snr, duplicate_read_count. Returns None if no history data exists. """ - # Check current max turn row = conn.execute( "SELECT MAX(turn_number) AS max_turn FROM conversation_turns WHERE session_id = ?", (session_id,), ).fetchone() - if not row or row["max_turn"] is None: return None - max_turn = row["max_turn"] - # Cache check cached = _cache.get(session_id) - if cached and cached[0] == max_turn: - return cached[1] - # Fetch turns - turns = conn.execute( + if cached is not None and max_turn < cached.max_turn_processed: + logger.warning( + "max_turn regressed for %s (cached=%d, now=%d) — rebuilding composition cache", + session_id, cached.max_turn_processed, max_turn, + ) + cached = None + + if cached is not None and max_turn == cached.max_turn_processed: + return _build_composition_result_from_state(cached) + + start_after = cached.max_turn_processed if cached is not None else 0 + new_turns = conn.execute( """SELECT turn_number, storage_mode, request_messages FROM conversation_turns - WHERE session_id = ? + WHERE session_id = ? AND turn_number > ? ORDER BY turn_number ASC""", - (session_id,), + (session_id, start_after), ).fetchall() - if not turns: + if not new_turns and cached is None: return None - turns_data = [dict(t) for t in turns] - category_bytes, total_bytes, dupes, tool_calls, tool_bytes, thinking_count = ( - _reconstruct_and_classify(turns_data) - ) - - result = _build_composition_result( - category_bytes, dupes, tool_calls, tool_bytes, - thinking_count=thinking_count, - ) - - # Cache - _cache[session_id] = (max_turn, result) + state = cached if cached is not None else _CompositionState() + try: + _fold_delta_into_composition_state(state, [dict(t) for t in new_turns]) + except Exception: + logger.exception("Incremental composition fold failed for %s — rebuilding", session_id) + rebuilt = _full_rebuild_composition_state(conn, session_id) + if rebuilt is None: + _cache.pop(session_id, None) + return None + state = rebuilt - return result + _cache[session_id] = state + return _build_composition_result_from_state(state) # ── Per-turn analysis ── -_per_turn_cache: Dict[str, Tuple[int, dict]] = {} +_per_turn_cache: Dict[str, _PerTurnState] = {} def _reconstruct_per_turn( @@ -378,12 +508,94 @@ def _sample_turns(turns: List[dict]) -> List[dict]: return [turns[i] for i in sorted(indices)] +def _classify_accumulated_categories(accumulated: List[dict]) -> Tuple[Dict[str, int], int]: + """Classify an accumulated message list into category byte totals. + + Used by the per-turn fold path which only needs categories, not the + extended tool-call/tool-bytes/thinking breakdown. + """ + totals: Dict[str, int] = {cat: 0 for cat in CATEGORIES} + for msg in accumulated: + sizes = _classify_message(msg) + for cat in CATEGORIES: + totals[cat] += sizes.get(cat, 0) + return totals, sum(totals.values()) + + +def _fold_delta_into_per_turn_state( + state: _PerTurnState, + new_turns: List[dict], +) -> None: + """Fold delta turns into the per-turn state in place. + + For each turn, classifies the full accumulated context at that point and + appends a per-turn entry. Compaction events reset `accumulated` to the + snapshot before classifying. + """ + for row in new_turns: + turn_number = row["turn_number"] + storage_mode = row["storage_mode"] + messages = _parse_turn_messages(row) + if messages is None: + state.max_turn_processed = turn_number + continue + + compacted = False + if storage_mode == "full": + if state.accumulated and len(messages) < len(state.accumulated): + compacted = True + state.accumulated = list(messages) + else: + state.accumulated.extend(messages) + + totals, total_bytes = _classify_accumulated_categories(state.accumulated) + categories = {} + for cat in CATEGORIES: + b = totals[cat] + pct = round(b / total_bytes * 100, 1) if total_bytes > 0 else 0.0 + categories[cat] = {"bytes": b, "pct": pct} + + state.turn_results.append({ + "turn": turn_number, + "msgs": len(state.accumulated), + "compacted": compacted, + "composition": { + "total_bytes": total_bytes, + "est_tokens": total_bytes // 4, + "categories": categories, + }, + }) + state.max_turn_processed = turn_number + + +def _full_rebuild_per_turn_state( + conn: sqlite3.Connection, + session_id: str, +) -> Optional[_PerTurnState]: + """Fetch all turns and fold them into a fresh per-turn state.""" + turns = conn.execute( + """SELECT turn_number, storage_mode, request_messages + FROM conversation_turns + WHERE session_id = ? + ORDER BY turn_number ASC""", + (session_id,), + ).fetchall() + if not turns: + return None + state = _PerTurnState() + _fold_delta_into_per_turn_state(state, [dict(t) for t in turns]) + return state + + def analyze_session_composition_per_turn( conn: sqlite3.Connection, session_id: str, ) -> Optional[dict]: """Analyze per-turn composition for a session. Returns sampled data for large sessions. + Uses an incremental cache: only turns past max_turn_processed are fetched + and folded. Sampling is applied on each call against the full result list. + Returns dict with: session_id, total_turns, sampled, turns[]. Returns None if no history data exists. """ @@ -391,44 +603,57 @@ def analyze_session_composition_per_turn( "SELECT MAX(turn_number) AS max_turn FROM conversation_turns WHERE session_id = ?", (session_id,), ).fetchone() - if not row or row["max_turn"] is None: return None - max_turn = row["max_turn"] - # Cache check cached = _per_turn_cache.get(session_id) - if cached and cached[0] == max_turn: - return cached[1] - turns = conn.execute( - """SELECT turn_number, storage_mode, request_messages - FROM conversation_turns - WHERE session_id = ? - ORDER BY turn_number ASC""", - (session_id,), - ).fetchall() + if cached is not None and max_turn < cached.max_turn_processed: + logger.warning( + "max_turn regressed for %s (per-turn cached=%d, now=%d) — rebuilding", + session_id, cached.max_turn_processed, max_turn, + ) + cached = None + + if cached is None or max_turn > cached.max_turn_processed: + start_after = cached.max_turn_processed if cached is not None else 0 + new_turns = conn.execute( + """SELECT turn_number, storage_mode, request_messages + FROM conversation_turns + WHERE session_id = ? AND turn_number > ? + ORDER BY turn_number ASC""", + (session_id, start_after), + ).fetchall() + + if not new_turns and cached is None: + return None - if not turns: - return None + state = cached if cached is not None else _PerTurnState() + try: + _fold_delta_into_per_turn_state(state, [dict(t) for t in new_turns]) + except Exception: + logger.exception("Per-turn incremental fold failed for %s — rebuilding", session_id) + rebuilt = _full_rebuild_per_turn_state(conn, session_id) + if rebuilt is None: + _per_turn_cache.pop(session_id, None) + return None + state = rebuilt - turns_data = [dict(t) for t in turns] - all_turns = _reconstruct_per_turn(turns_data) + _per_turn_cache[session_id] = state + else: + state = cached - sampled = len(all_turns) > 50 - sampled_turns = _sample_turns(all_turns) if sampled else all_turns + sampled = len(state.turn_results) > 50 + sampled_turns = _sample_turns(state.turn_results) if sampled else state.turn_results - result = { + return { "session_id": session_id, - "total_turns": len(all_turns), + "total_turns": len(state.turn_results), "sampled": sampled, "turns": sampled_turns, } - _per_turn_cache[session_id] = (max_turn, result) - return result - def clear_cache(session_id: Optional[str] = None) -> None: """Clear composition cache for a session or all sessions.""" diff --git a/tests/test_api/test_composition.py b/tests/test_api/test_composition.py index 95c5857..6785bcd 100644 --- a/tests/test_api/test_composition.py +++ b/tests/test_api/test_composition.py @@ -274,12 +274,15 @@ def test_returns_none_for_missing_session(self): assert result is None def test_cache_hit(self): + # Cache hit: second call returns an equal result. The incremental + # cache rebuilds the result dict each call from state, so identity is + # not preserved, but value must match exactly. conn = self._make_conn([ _make_turn(1, "full", [{"role": "user", "content": "test"}]), ]) r1 = analyze_session_composition(conn, "test-session") r2 = analyze_session_composition(conn, "test-session") - assert r1 is r2 # Same object = cache hit + assert r1 == r2 def test_cache_invalidation_on_new_turn(self): conn = self._make_conn([ From 69c79a8f323f640de2ffdb8b2ac0632dfb76f5ef Mon Sep 17 00:00:00 2001 From: ArkNill <48707894+ArkNill@users.noreply.github.com> Date: Wed, 20 May 2026 15:06:44 +0900 Subject: [PATCH 4/5] test(composition): incremental cache scenario coverage (#16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds TestIncrementalComposition class covering the new incremental fold path: test_incremental_matches_full_rebuild_all_scenarios — equivalence contract across 6 scenarios (delta-only, full-only, mixed, compaction-then-delta, delta-then-compaction, repeated-compaction). For each scenario at each turn, compares the result built from the live incremental cache against a from-scratch rebuild. test_incremental_fetches_only_delta — wraps the connection to log queries and confirms the second call issues exactly one `turn_number > ?` fetch and zero full-history fetches. test_compaction_within_incremental_path_resets_state — verifies that a "full" turn arriving on top of cached state wipes pre-compact reads from duplicate_reads and resets totals. test_max_turn_regression_triggers_rebuild — deleting a turn after the cache is populated causes the next call to rebuild cleanly rather than serve stale state. test_no_change_call_is_a_pure_cache_read — repeated calls without new turns return equal results. test_per_turn_incremental_appends_one_entry — adding one delta turn appends exactly one new entry to total_turns and preserves earlier per-turn entries verbatim. test_per_turn_sampling_window_shifts_with_new_turns — for sessions over the 50-turn sampling threshold, the last-5 window slides correctly as new turns arrive (re-sampling happens on each call). --- tests/test_api/test_composition.py | 252 +++++++++++++++++++++++++++++ 1 file changed, 252 insertions(+) diff --git a/tests/test_api/test_composition.py b/tests/test_api/test_composition.py index 6785bcd..0d20dd1 100644 --- a/tests/test_api/test_composition.py +++ b/tests/test_api/test_composition.py @@ -509,6 +509,258 @@ def test_compaction_detection(self): assert result["turns"][1]["compacted"] is True +# ── Incremental cache coverage (#16) ── + + +def _add_turn(conn, session_id, turn_number, storage_mode, messages): + """Append one turn to a test DB connection.""" + conn.execute( + """INSERT INTO conversation_turns + (ts, session_id, turn_number, storage_mode, request_messages) + VALUES (?, ?, ?, ?, ?)""", + (time.time(), session_id, turn_number, storage_mode, json.dumps(messages)), + ) + conn.commit() + + +def _make_incremental_conn(): + """Create a DB with the same schema as TestAnalyzeSessionComposition uses.""" + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + conn.execute( + """CREATE TABLE conversation_turns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts REAL, session_id TEXT, turn_number INTEGER, + storage_mode TEXT, request_messages TEXT, + request_size_bytes INTEGER DEFAULT 0, + response_size_bytes INTEGER DEFAULT 0, + provider TEXT DEFAULT 'anthropic' + )""" + ) + return conn + + +class TestIncrementalComposition: + """Coverage for the incremental cache path introduced for #16. + + Equivalence with full rebuild is the primary contract — for any sequence + of delta/full turns, the incremental path must produce the same result + dict that a from-scratch rebuild produces. + """ + + def setup_method(self): + clear_cache() + + # ── Equivalence with full rebuild ── + + def _scenarios(self): + def u(t): + return {"role": "user", "content": t} + def a(t): + return {"role": "assistant", "content": [{"type": "text", "text": t}]} + return { + "delta_only": [ + ("full", [u("first")]), + ("delta", [a("reply 1")]), + ("delta", [u("second"), a("reply 2")]), + ], + "full_only": [ + ("full", [u("turn 1")]), + ("full", [u("turn 1"), a("reply 1")]), + ("full", [u("turn 1"), a("reply 1"), u("turn 2")]), + ], + "mixed": [ + ("full", [u("opening")]), + ("delta", [a("answer")]), + ("delta", [u("follow"), a("response")]), + ("delta", [u("third")]), + ], + "compaction_then_delta": [ + ("full", [u("preamble"), a("background " * 20)]), + ("delta", [u("topic shift")]), + ("full", [u("post-compact")]), + ("delta", [a("post-compact reply")]), + ], + "delta_then_compaction": [ + ("full", [u("start")]), + ("delta", [a("growing " * 30)]), + ("delta", [u("more"), a("output " * 20)]), + ("full", [u("compacted summary")]), + ], + "repeated_compaction": [ + ("full", [u("v1")]), + ("delta", [a("v1 ans")]), + ("full", [u("v2")]), + ("delta", [a("v2 ans")]), + ("full", [u("v3")]), + ], + } + + def test_incremental_matches_full_rebuild_all_scenarios(self): + sid = "equiv-session" + for label, turns in self._scenarios().items(): + conn = _make_incremental_conn() + for i, (mode, msgs) in enumerate(turns, start=1): + _add_turn(conn, sid, i, mode, msgs) + + # Incremental path (cache survives between calls) + incr = analyze_session_composition(conn, sid) + + # Full rebuild path (fresh state every call) + clear_cache() + full = analyze_session_composition(conn, sid) + + assert incr == full, ( + f"scenario={label} turn={i}: incremental result diverged from full rebuild" + ) + # Carry incremental cache forward to next turn iteration + clear_cache() + clear_cache() + + # ── Behavior verification ── + + def test_incremental_fetches_only_delta(self): + sid = "fetch-bounds" + conn = _make_incremental_conn() + _add_turn(conn, sid, 1, "full", [{"role": "user", "content": "first"}]) + analyze_session_composition(conn, sid) + + # Add a new turn; track which queries fire + _add_turn(conn, sid, 2, "delta", + [{"role": "assistant", "content": [{"type": "text", "text": "answer"}]}]) + + captured: list = [] + original_execute = conn.execute + + def tracking_execute(query, *args, **kwargs): + captured.append(query) + return original_execute(query, *args, **kwargs) + + # sqlite3.Connection.execute is read-only; wrap the conn in a passthrough + class _Wrapper: + def __init__(self, c): + self._c = c + def execute(self, query, *args, **kwargs): + captured.append(query) + return self._c.execute(query, *args, **kwargs) + + analyze_session_composition(_Wrapper(conn), sid) + + delta_fetches = [q for q in captured if "turn_number > ?" in q] + full_fetches = [ + q for q in captured + if "FROM conversation_turns" in q and "turn_number > ?" not in q + and "MAX(turn_number)" not in q + ] + assert len(delta_fetches) == 1, "expected exactly one delta fetch" + assert len(full_fetches) == 0, "must not fetch the full turn history on incremental path" + + def test_compaction_within_incremental_path_resets_state(self): + sid = "compact-incremental" + conn = _make_incremental_conn() + u = {"role": "user", "content": "pre-compact"} + a = {"role": "assistant", "content": [{"type": "text", "text": "long " * 200}]} + read_a = { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "/pre.py"}}, + ], + } + _add_turn(conn, sid, 1, "full", [u, a, read_a]) + _add_turn(conn, sid, 2, "delta", [read_a]) # /pre.py = 2 before compact + before_compact = analyze_session_composition(conn, sid) + assert "/pre.py" in before_compact["duplicate_reads"] + assert before_compact["categories"]["assistant_text"]["bytes"] > 0 + + # Compaction event arrives + _add_turn(conn, sid, 3, "full", [{"role": "user", "content": "compacted"}]) + after_compact = analyze_session_composition(conn, sid) + assert "/pre.py" not in after_compact["duplicate_reads"] + # assistant_text was wiped (only user message in the new snapshot) + assert after_compact["categories"]["assistant_text"]["bytes"] == 0 + assert after_compact["categories"]["user_text"]["bytes"] > 0 + + def test_max_turn_regression_triggers_rebuild(self): + sid = "regress" + conn = _make_incremental_conn() + _add_turn(conn, sid, 1, "full", [{"role": "user", "content": "a"}]) + _add_turn(conn, sid, 2, "delta", + [{"role": "assistant", "content": [{"type": "text", "text": "b"}]}]) + analyze_session_composition(conn, sid) + + # Simulate a delete: drop turn 2 + conn.execute("DELETE FROM conversation_turns WHERE turn_number = 2") + conn.commit() + + result = analyze_session_composition(conn, sid) + # After rebuild from remaining turn 1 only, assistant_text is 0 + assert result["categories"]["assistant_text"]["bytes"] == 0 + assert result["categories"]["user_text"]["bytes"] > 0 + + def test_no_change_call_is_a_pure_cache_read(self): + sid = "noop" + conn = _make_incremental_conn() + _add_turn(conn, sid, 1, "full", [{"role": "user", "content": "x"}]) + r1 = analyze_session_composition(conn, sid) + r2 = analyze_session_composition(conn, sid) + assert r1 == r2 + + # ── Per-turn variant ── + + def _per_turn_conn(self): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + conn.execute("""CREATE TABLE conversation_turns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts REAL, session_id TEXT, turn_number INTEGER, + storage_mode TEXT, request_messages TEXT, + response_message TEXT, thinking_blocks TEXT, + model TEXT, temperature REAL, max_tokens INTEGER, + total_message_count INTEGER DEFAULT 0, + previous_message_count INTEGER DEFAULT 0, + request_size_bytes INTEGER DEFAULT 0, + response_size_bytes INTEGER DEFAULT 0, + request_id INTEGER + )""") + return conn + + def test_per_turn_incremental_appends_one_entry(self): + sid = "pt-append" + conn = self._per_turn_conn() + _add_turn(conn, sid, 1, "full", [{"role": "user", "content": "t1"}]) + first = analyze_session_composition_per_turn(conn, sid) + assert first["total_turns"] == 1 + + _add_turn(conn, sid, 2, "delta", + [{"role": "assistant", "content": [{"type": "text", "text": "t2"}]}]) + second = analyze_session_composition_per_turn(conn, sid) + assert second["total_turns"] == 2 + # Existing turn entry preserved verbatim (append-only) + assert second["turns"][0] == first["turns"][0] + + def test_per_turn_sampling_window_shifts_with_new_turns(self): + """The last-5 sampling window must follow new turns even though state + keeps the full result list. Each call re-samples on access. + """ + sid = "pt-window" + conn = self._per_turn_conn() + # Seed 55 turns (>50 → sampling kicks in) + for n in range(1, 56): + _add_turn(conn, sid, n, "full" if n == 1 else "delta", + [{"role": "user", "content": f"t{n}"}]) + r1 = analyze_session_composition_per_turn(conn, sid) + assert r1["sampled"] is True + last5_v1 = [t["turn"] for t in r1["turns"][-5:]] + assert last5_v1 == [51, 52, 53, 54, 55] + + # Add one more turn; last-5 should slide + _add_turn(conn, sid, 56, "delta", + [{"role": "assistant", "content": [{"type": "text", "text": "t56"}]}]) + r2 = analyze_session_composition_per_turn(conn, sid) + last5_v2 = [t["turn"] for t in r2["turns"][-5:]] + assert last5_v2 == [52, 53, 54, 55, 56] + + # ── File-based composition tests (Codex / Gemini) ── From efb801dbeb00c8fae815096bc327886998437eba Mon Sep 17 00:00:00 2001 From: ArkNill <48707894+ArkNill@users.noreply.github.com> Date: Wed, 20 May 2026 15:07:27 +0900 Subject: [PATCH 5/5] docs(changelog): incremental composition perf + read_counts behavior change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two Unreleased entries documenting the #16 work: - Performance: incremental composition cache (resolves the O(n²) replay that produced 10 GB+ RSS balloons and 46s response times on multi-agent hosts with thousands of turns per session). - Changed: duplicate_reads no longer accumulates across compaction, surfacing the latent overcounting bug that was masking real duplicate-read warnings on post-compact context. --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c4af5c5..5bcf1ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,9 +4,13 @@ All notable changes to llm-relay are documented here. ## [Unreleased] +### Performance +- **Incremental composition cache** (`composition.py`, #16): `analyze_session_composition` and `analyze_session_composition_per_turn` now fold delta turns into cached state instead of re-walking the full session on every new turn. Previously the cache invalidated whenever any new turn arrived, forcing an O(n²) replay; on long-running sessions this caused `/api/v1/turns` to balloon to tens of seconds and daemon RSS to climb above 10 GB within ~30 min of normal multi-agent traffic. Cache now keys on `session_id` alone with `(max_turn_processed, accumulated, totals, …)` state; new turns trigger a `WHERE turn_number > max_turn_processed` fetch. Compaction events (`storage_mode="full"`) reset accumulated state and re-absorb the snapshot. Exception during fold falls back to a full rebuild so an incremental error can't poison the cache. + ### Changed - **Claude Code zone defaults rescaled to 665K ceiling**: Yellow 332K / Orange 465K / Red 600K / Hard 665K (`_zones.py`, `tui.py`, `display.py`, `.env.public`). Rationale: Claude Code's client-side auto-compact (re-introduced in v2.1.139) triggers around 650-670K cumulative context; the new defaults let operators hand off to a new session before compaction degrades context continuity. Override via `LLM_TOKEN_A_*` / `LLM_TOKEN_CEILING` env vars if a different ceiling is needed (e.g. `500000` for public deployments without 1M entitlement). - **Red-zone message split for CC vs Codex** (`i18n.py`): new keys `zone.abs.red.cc` and `zone.ratio.red.cc` carry an explicit "Auto-compact imminent — hand off to a new session" guideline used by CC paths. Codex paths keep the existing "Session rotation required" wording (Codex has no client-side compaction; the 400K hard limit semantics differ). +- **`duplicate_reads` no longer accumulates across compaction** (`composition.py`): when a `storage_mode="full"` turn arrives, read counts reset together with the accumulated context. Previously reads from pre-compaction turns were counted as duplicates of post-compaction reads, inflating `duplicate_read_count` and triggering spurious `duplicate_read_warning` flips. Visible in `/api/v1/turns`, `/api/v1/display`, dashboard Context Health, and TUI duplicate-file listings. ### Tests - `tests/test_api/test_turns.py::_zone_env` autouse fixture now patches both `_zones` and `routes` modules so legacy 1M-scale assertions stay stable against new production defaults.