Bound the session-history reads in preference tracking

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) |
 
 ---

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` |

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
 

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"]