From 40fbbf392b5acc3147b7fab51913111334b33635 Mon Sep 17 00:00:00 2001 From: Piyush Makhija Date: Sun, 9 Aug 2026 08:22:28 +0530 Subject: [PATCH] perf: bound the session history preference tracking reads A stack sampler on a real pipeline run found where the pipeline's time was going. 106 seconds of a 184-second run were inside _collect_session_texts, which read every .md/.txt/.json/.log file under every session directory, fully into memory, with no bound of any kind. The directory is ~/.claude/memory/logs/sessions. On the machine that found this it held 4.4 GB across 1,658 session directories -- 11,050 matching text files, read in full, on every run. The engine writes a session directory per run, so each run made the next one slower, permanently and without limit. One run had already been pushed past the entry point's 300-second orchestration cap. What the reading is for sets the bound, and it is a modest purpose for 4.4 GB of I/O: track_preferences lowercases the text and asks whether known skill and agent names appear in it, keeping anything seen in at least MIN_OCCURRENCES sessions. That is a frequency heuristic over recent behaviour. The hundred-and-fifty-first megabyte of one log cannot establish a preference that three separate sessions did not, and a session from six months ago is not evidence about what the user prefers now. So: the newest MAX_SESSIONS_SCANNED directories by mtime, and at most MAX_FILE_BYTES per file. Sorted by mtime rather than by name because session directory names are UUID-bearing and do not sort chronologically -- sorting by name would have dropped an arbitrary set rather than the oldest. Measured end to end on the same machine and task: 174s -> 51s. The collection step itself went 106s -> 1.0s. In that run the recency cap did all the work and the per-file cap never engaged, which is the expected shape: the bulk is many sessions, not one enormous file. The per-file cap is there for the other shape. The cap reports itself on stderr -- "skipped 1633 older session(s), truncated 0 file(s)". This repository's rule is that a cap must not be silent, and a sampled history that does not say it was sampled reads as a complete one. stderr rather than a logger because this module is loaded by path as a standalone script. Ten tests, each bound paired with a case proving the bound is what makes the difference: without the cap all 20 fixture sessions are read, with it 5 are, and the three dropped are the oldest by mtime. They assert on HOW MUCH is read, not on what comes back, because the function's output was always correct -- correctness was never the problem, and no correctness test could have seen this. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 4 +- README.md | 10 +- .../architecture/preference_tracker.py | 85 +++++++-- tests/test_preference_tracker_scan_bounds.py | 162 ++++++++++++++++++ 4 files changed, 242 insertions(+), 19 deletions(-) create mode 100644 tests/test_preference_tracker_scan_bounds.py diff --git a/CLAUDE.md b/CLAUDE.md index b0862d23..965fa7c1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,8 +20,8 @@ Claude Workflow Engine is a 3-level LangGraph-based orchestration pipeline for a | **Status** | Active Development | | **Primary Location** | langgraph_engine/ | | **MCP Servers** | 13 servers -- all in separate repos under [techdeveloper-org](https://github.com/orgs/techdeveloper-org/repositories); 1 also keeps an in-engine copy in `src/mcp/` | -| **Total Python Files** | 248 (langgraph_engine/); 491 repo-wide | -| **Test Files** | 103 (94 unit, 5 integration, 3 e2e, 1 load) | +| **Total Python Files** | 248 (langgraph_engine/); 492 repo-wide | +| **Test Files** | 104 (95 unit, 5 integration, 3 e2e, 1 load) | | **Call Graph** | 578 classes, 3,985 methods, 4 languages (Python/Java/TS/Kotlin) | --- diff --git a/README.md b/README.md index 7eec65df..99bcdf92 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ [![Python](https://img.shields.io/badge/Python-3.10%2B-green)](https://python.org) [![CI](https://github.com/techdeveloper-org/claude-workflow-engine/actions/workflows/ci.yml/badge.svg)](https://github.com/techdeveloper-org/claude-workflow-engine/actions/workflows/ci.yml) [![License](https://img.shields.io/badge/License-MIT-yellow)](LICENSE) -[![Tests](https://img.shields.io/badge/Tests-103%20files%20·%205%20integration-brightgreen)](tests/) +[![Tests](https://img.shields.io/badge/Tests-104%20files%20·%205%20integration-brightgreen)](tests/) [![Discussions](https://img.shields.io/badge/Discussions-GitHub-blueviolet)](../../discussions) --- @@ -469,8 +469,8 @@ claude-workflow-engine/ # 369 Python files total ├── policies/ # non-.md policy state only (failure-kb.json) — see note below │ └── 03-execution-system/failure-prevention/failure-kb.json │ -├── tests/ # 103 test_*.py files (124 total Python files) -│ ├── test_*.py # 94 unit tests +├── tests/ # 104 test_*.py files (125 total Python files) +│ ├── test_*.py # 95 unit tests │ ├── integration/ # 5 integration tests (GitHub, MCP, runtime verification) │ ├── e2e/ # 3 end-to-end scenario tests │ └── load/ # 1 concurrency / load test @@ -628,11 +628,11 @@ ENABLE_HEALTH_SERVER=1 ENABLE_METRICS=1 LOG_FORMAT=json \ ## Testing -103 test_*.py files (124 total Python files in `tests/` including conftest and `__init__` files): +104 test_*.py files (125 total Python files in `tests/` including conftest and `__init__` files): | Category | Files | Notes | |---|:---:|---| -| Unit tests | 94 | In `tests/test_*.py` — no external dependencies required | +| Unit tests | 95 | In `tests/test_*.py` — no external dependencies required | | Integration tests | 5 | `tests/integration/` — require live GitHub token or MCP servers | | E2E tests | 3 | `tests/e2e/` — require full pipeline environment | | Load tests | 1 | `tests/load/` — enabled with `RUN_LOAD_TESTS=1` | diff --git a/langgraph_engine/context_sync/architecture/preference_tracker.py b/langgraph_engine/context_sync/architecture/preference_tracker.py index 61bb4d88..1f240ffd 100644 --- a/langgraph_engine/context_sync/architecture/preference_tracker.py +++ b/langgraph_engine/context_sync/architecture/preference_tracker.py @@ -37,6 +37,27 @@ # Minimum number of appearances across sessions to register as a preference MIN_OCCURRENCES = 3 +# Bounds on how much session history one run may read. +# +# This function used to read EVERY .md/.txt/.json/.log file under every session +# directory, fully into memory, with no bound of any kind. A stack sampler run on +# 2026-08-09 measured 106 of a 184-second pipeline run inside _read_file. The +# directory it walks is ~/.claude/memory/logs/sessions, which on the machine that +# found this held 4.4 GB across 1,658 session directories -- 11,050 matching text +# files, read in full, on every run. The cost grows with every run the engine +# performs, because each run writes another session directory that the next run +# then reads, and it had already pushed one run past the entry point's +# 300-second orchestration cap. +# +# What the reading is FOR sets the bound. track_preferences lowercases the text and +# asks whether known skill and agent names appear in it, then keeps anything seen +# in at least MIN_OCCURRENCES sessions. That is a frequency heuristic over recent +# behaviour: the hundred-and-fifty-first megabyte of a single log cannot change a +# preference that three separate sessions did not already establish, and a session +# from six months ago is not evidence about what the user prefers now. +MAX_SESSIONS_SCANNED = 25 +MAX_FILE_BYTES = 1_000_000 + # Output file for storing learned preferences PREFERENCES_FILE = Path.home() / ".claude" / "memory" / "user-preferences.json" @@ -207,26 +228,66 @@ def _collect_session_texts(sessions_dir: Path) -> List[str]: return texts text_extensions = {".md", ".txt", ".json", ".log"} + truncated_files = 0 + skipped_sessions = 0 def _read_file(path: Path) -> Optional[str]: + """Read at most MAX_FILE_BYTES of one file.""" try: - return path.read_text(encoding="utf-8", errors="replace") + with path.open("r", encoding="utf-8", errors="replace") as handle: + content = handle.read(MAX_FILE_BYTES) except OSError: return None + return content + + try: + entries = list(sessions_dir.iterdir()) + except OSError: + return texts + + loose_files = [e for e in entries if e.is_file() and e.suffix.lower() in text_extensions] + session_dirs = [e for e in entries if e.is_dir()] - for item in sessions_dir.iterdir(): - if item.is_file(): - if item.suffix.lower() in text_extensions: - content = _read_file(item) + # Newest first, so the cap drops the oldest sessions rather than an arbitrary + # set. mtime is used rather than the name because session directory names are + # UUID-bearing and do not sort chronologically. + try: + session_dirs.sort(key=lambda p: p.stat().st_mtime, reverse=True) + except OSError: + pass + if len(session_dirs) > MAX_SESSIONS_SCANNED: + skipped_sessions = len(session_dirs) - MAX_SESSIONS_SCANNED + session_dirs = session_dirs[:MAX_SESSIONS_SCANNED] + + for item in loose_files: + content = _read_file(item) + if content: + if len(content) == MAX_FILE_BYTES: + truncated_files += 1 + texts.append(content) + + for session_dir in session_dirs: + try: + subitems = list(session_dir.iterdir()) + except OSError: + continue + for subitem in subitems: + if subitem.is_file() and subitem.suffix.lower() in text_extensions: + content = _read_file(subitem) if content: + if len(content) == MAX_FILE_BYTES: + truncated_files += 1 texts.append(content) - elif item.is_dir(): - # One level deeper (session subdirectory) - for subitem in item.iterdir(): - if subitem.is_file() and subitem.suffix.lower() in text_extensions: - content = _read_file(subitem) - if content: - texts.append(content) + + # A cap that hides its own size turns a sampled result into one that reads as + # complete. Reported on stderr rather than through a logger because this module + # is loaded by path as a standalone script and has no logger of its own. + if skipped_sessions or truncated_files: + print( + "[preference_tracker] sampled history: skipped %d older session(s), " + "truncated %d file(s) at %d bytes" % (skipped_sessions, truncated_files, MAX_FILE_BYTES), + file=sys.stderr, + ) return texts diff --git a/tests/test_preference_tracker_scan_bounds.py b/tests/test_preference_tracker_scan_bounds.py new file mode 100644 index 00000000..ac6cc64b --- /dev/null +++ b/tests/test_preference_tracker_scan_bounds.py @@ -0,0 +1,162 @@ +"""Preference tracking must not read the whole session history. + +`_collect_session_texts` read every .md/.txt/.json/.log file under every session +directory, fully into memory, with no bound of any kind. A stack sampler on a real +pipeline run (2026-08-09) measured 106 seconds of a 184-second run inside it. The +directory is ~/.claude/memory/logs/sessions, which held 4.4 GB across 1,658 session +directories -- 11,050 matching text files, read in full, every run. + +The cost grew with every run the engine performed, because each run writes another +session directory that the next run then reads. One run had already been pushed +past the entry point's 300-second orchestration cap. No test could see it -- the +function returned correct texts, and correctness was never the problem. + +These tests therefore assert on HOW MUCH is read, not on what comes back, and each +bound is paired with a case proving the bound is what makes the difference. + +The bounds are justified by what the reading is for: `track_preferences` lowercases +the text and asks whether known skill and agent names appear, keeping anything seen +in at least MIN_OCCURRENCES sessions. That is a frequency heuristic over recent +behaviour, so old sessions and the deep tail of a single log cannot change its +answer in a way worth 106 seconds a run. +""" + +import importlib.util +import sys +from pathlib import Path + +import pytest + +_PROJECT_ROOT = Path(__file__).resolve().parent.parent +if str(_PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(_PROJECT_ROOT)) + +_MODULE_PATH = _PROJECT_ROOT / "langgraph_engine" / "context_sync" / "architecture" / "preference_tracker.py" + + +def _load_module(): + """Load preference_tracker by path, as session_loader does at runtime.""" + spec = importlib.util.spec_from_file_location("preference_tracker_under_test", _MODULE_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture() +def tracker(): + """A freshly loaded module, so constant patching cannot leak between tests.""" + return _load_module() + + +def _make_sessions(root, count, marker="skill-marker", body_bytes=64): + """Create ``count`` session directories, oldest first by mtime.""" + import os + import time + + for i in range(count): + d = root / ("SESSION-%03d" % i) + d.mkdir() + (d / "session.md").write_text(marker + " " + ("x" * body_bytes), encoding="utf-8") + # Stagger mtimes so "newest first" has a defined answer. + stamp = time.time() - (count - i) * 60 + os.utime(d, (stamp, stamp)) + return root + + +class TestTheSessionCountIsBounded: + """An unbounded scan is what made the pipeline slower on every run.""" + + def test_only_the_newest_sessions_are_read(self, tracker, tmp_path): + tracker.MAX_SESSIONS_SCANNED = 5 + _make_sessions(tmp_path, 20) + texts = tracker._collect_session_texts(tmp_path) + assert len(texts) == 5 + + def test_without_the_cap_all_of_them_would_be_read(self, tracker, tmp_path): + """The half that proves the cap is doing the work. + + Without this, the test above passes on any history small enough to fit + under the cap anyway -- which is every history, right up until the one + that matters. + """ + tracker.MAX_SESSIONS_SCANNED = 10**9 + _make_sessions(tmp_path, 20) + texts = tracker._collect_session_texts(tmp_path) + assert len(texts) == 20 + + def test_the_oldest_sessions_are_the_ones_dropped(self, tracker, tmp_path): + """Recency is the point: preferences are about behaviour now.""" + import os + import time + + tracker.MAX_SESSIONS_SCANNED = 2 + for i, name in enumerate(("oldest", "middle", "newest")): + d = tmp_path / name + d.mkdir() + (d / "s.md").write_text(name, encoding="utf-8") + stamp = time.time() - (3 - i) * 3600 + os.utime(d, (stamp, stamp)) + texts = tracker._collect_session_texts(tmp_path) + assert set(texts) == {"newest", "middle"} + assert "oldest" not in texts + + +class TestPerFileBytesAreBounded: + """One 159 MB log must not be read whole to find a skill name in it.""" + + def test_a_large_file_is_truncated(self, tracker, tmp_path): + tracker.MAX_FILE_BYTES = 100 + d = tmp_path / "SESSION-big" + d.mkdir() + (d / "huge.log").write_text("y" * 5000, encoding="utf-8") + texts = tracker._collect_session_texts(tmp_path) + assert len(texts) == 1 + assert len(texts[0]) == 100 + + def test_a_small_file_is_read_whole(self, tracker, tmp_path): + """The half that fails if truncation were unconditional.""" + tracker.MAX_FILE_BYTES = 100 + d = tmp_path / "SESSION-small" + d.mkdir() + (d / "small.md").write_text("z" * 40, encoding="utf-8") + texts = tracker._collect_session_texts(tmp_path) + assert texts == ["z" * 40] + + +class TestTheCapAnnouncesItself: + """A cap that hides its size makes a sample read as if it were complete.""" + + def test_skipped_sessions_are_reported(self, tracker, tmp_path, capsys): + tracker.MAX_SESSIONS_SCANNED = 3 + _make_sessions(tmp_path, 10) + tracker._collect_session_texts(tmp_path) + assert "skipped 7 older session" in capsys.readouterr().err + + def test_nothing_is_reported_when_nothing_was_cut(self, tracker, tmp_path, capsys): + """The half that fails if the notice were unconditional.""" + tracker.MAX_SESSIONS_SCANNED = 100 + _make_sessions(tmp_path, 3) + tracker._collect_session_texts(tmp_path) + assert "preference_tracker" not in capsys.readouterr().err + + +class TestItStillWorksOnOrdinaryInput: + """The bounds must not change the answer for a normal history.""" + + def test_loose_files_and_session_dirs_are_both_read(self, tracker, tmp_path): + (tmp_path / "loose.md").write_text("loose-content", encoding="utf-8") + d = tmp_path / "SESSION-1" + d.mkdir() + (d / "inner.json").write_text("inner-content", encoding="utf-8") + texts = tracker._collect_session_texts(tmp_path) + assert set(texts) == {"loose-content", "inner-content"} + + def test_a_missing_directory_returns_empty(self, tracker, tmp_path): + assert tracker._collect_session_texts(tmp_path / "nope") == [] + + def test_non_text_extensions_are_ignored(self, tracker, tmp_path): + d = tmp_path / "SESSION-1" + d.mkdir() + (d / "keep.md").write_text("keep", encoding="utf-8") + (d / "skip.pkl").write_text("skip", encoding="utf-8") + assert tracker._collect_session_texts(tmp_path) == ["keep"]