From a5d4dac0e7a13cb88b1a8ce7300d6597b0765953 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 1 Aug 2026 02:04:01 +0300 Subject: [PATCH 1/5] fix: derive watcher projects from workspace metadata --- ...2026-08-01-d2-project-derivation-design.md | 77 ++++++++++++ scripts/d2_project_backfill.py | 115 ++++++++++++++++++ src/brainlayer/watcher_bridge.py | 62 ++++++++-- tests/test_project_backfill.py | 81 ++++++++++++ tests/test_watcher_bridge.py | 55 +++++++++ 5 files changed, 380 insertions(+), 10 deletions(-) create mode 100644 docs/plans/2026-08-01-d2-project-derivation-design.md create mode 100644 scripts/d2_project_backfill.py create mode 100644 tests/test_project_backfill.py diff --git a/docs/plans/2026-08-01-d2-project-derivation-design.md b/docs/plans/2026-08-01-d2-project-derivation-design.md new file mode 100644 index 00000000..86669499 --- /dev/null +++ b/docs/plans/2026-08-01-d2-project-derivation-design.md @@ -0,0 +1,77 @@ +# D2 Project Derivation Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Derive watcher chunk projects only from trustworthy workspace signals and safely repair the existing numeric-project corruption. + +**Architecture:** Keep source-path parsing only for Claude's explicit `projects/` workspace encoding. For every other provider, obtain the project from session-record workspace metadata, normalize the workspace basename, and return `None` when no such signal is present. A standalone, resumable migration will update only bare one- or two-digit projects, export a narrow rollback TSV first, batch updates, and record its own audit/progress state. + +**Tech Stack:** Python 3, SQLite/APSW, pytest. + +--- + +### Task 1: Define derivation regressions + +**Files:** +- Modify: `tests/test_watcher_bridge.py` +- Modify: `src/brainlayer/watcher_bridge.py` + +**Step 1: Write failing tests** + +Add tests for a Codex date-partitioned source with a metadata `cwd`, a Claude `projects/` source, Cursor and Gemini sources with a metadata `cwd`, and a source with no derivable workspace. Assert no provider falls back to a numeric directory. + +**Step 2: Run to verify failure** + +Run: `pytest tests/test_watcher_bridge.py::TestProjectExtraction -v` + +Expected: FAIL because extraction has no metadata input and returns the date segment. + +**Step 3: Implement minimal derivation** + +Change `_extract_project_from_source` to accept an optional entry/metadata workspace signal, preserve the Claude `projects/` parser, and reject pure numeric candidates in all paths. Pass each watcher entry to the helper. + +**Step 4: Run to verify pass** + +Run: `pytest tests/test_watcher_bridge.py::TestProjectExtraction -v` + +Expected: PASS. + +### Task 2: Implement the narrow, idempotent backfill + +**Files:** +- Create: `src/brainlayer/project_backfill.py` +- Create: `tests/test_project_backfill.py` + +**Step 1: Write failing migration tests** + +Create a fixture DB containing numeric projects with recoverable and unrecoverable source sessions plus a non-numeric project. Test TSV rollback export, real-project/NULL/untouched counts, and a second migration run producing zero changes with identical counts. + +**Step 2: Run to verify failure** + +Run: `pytest tests/test_project_backfill.py -v` + +Expected: FAIL because the migration module does not exist. + +**Step 3: Implement minimal migration** + +Create a migration that reads session metadata from source JSONL files, updates only `project GLOB '[0-9]' OR project GLOB '[0-9][0-9]'`, uses 5–10K transactions, checkpoints every three batches, stores a migration audit record, and offers a CLI with an explicit live-DB opt-in. + +**Step 4: Run to verify pass** + +Run: `pytest tests/test_project_backfill.py -v` + +Expected: PASS. + +### Task 3: Verify and execute the production pass + +**Files:** +- Create: `REPORT.md` +- Create: `docs.local/tasks/d2-rollback-.tsv` (untracked rollback artifact) + +**Step 1:** Run the focused and full pytest suites. + +**Step 2:** Stop enrichment writers and claim the shared bulk-DB lock in `docs.local/.../collab.md`. + +**Step 3:** Re-count numeric projects; export the rollback TSV; run `PRAGMA wal_checkpoint(FULL)`; execute the migration in batches; checkpoint every three batches and at completion; release the lock. + +**Step 4:** Independently query the three-way split, remaining numeric projects, and distinct projects before/after. Write `REPORT.md` with command evidence and commit only repository changes (not the live DB artifact). diff --git a/scripts/d2_project_backfill.py b/scripts/d2_project_backfill.py new file mode 100644 index 00000000..bc50e551 --- /dev/null +++ b/scripts/d2_project_backfill.py @@ -0,0 +1,115 @@ +"""One-off repair for chunks whose project was derived from a date directory.""" + +from __future__ import annotations + +import argparse +import csv +import sqlite3 +from dataclasses import dataclass +from pathlib import Path + +from brainlayer.watcher_bridge import _extract_project_from_session_file + +NUMERIC_PROJECT_PREDICATE = "(project GLOB '[0-9]' OR project GLOB '[0-9][0-9]')" +_BATCH_SIZE_MIN = 5_000 +_BATCH_SIZE_MAX = 10_000 + + +@dataclass(frozen=True) +class ProjectBackfillResult: + rows_updated: int + rows_rederived: int + rows_set_null: int + rows_left_untouched: int + + +def _export_rollback(conn: sqlite3.Connection, rollback_path: Path) -> None: + if rollback_path.exists(): + raise FileExistsError(f"rollback artifact already exists: {rollback_path}") + rollback_path.parent.mkdir(parents=True, exist_ok=True) + with rollback_path.open("x", encoding="utf-8", newline="") as handle: + writer = csv.writer(handle, delimiter="\t", lineterminator="\n") + writer.writerow(("id", "project")) + writer.writerows( + conn.execute(f"SELECT id, project FROM chunks WHERE {NUMERIC_PROJECT_PREDICATE} ORDER BY rowid") + ) + + +def backfill_numeric_projects( + db_path: str | Path, + *, + rollback_path: str | Path | None = None, + batch_size: int = _BATCH_SIZE_MIN, +) -> ProjectBackfillResult: + """Update only one- and two-digit projects from recorded session metadata.""" + if not _BATCH_SIZE_MIN <= batch_size <= _BATCH_SIZE_MAX: + raise ValueError("batch_size must be between 5,000 and 10,000") + + conn = sqlite3.connect(Path(db_path), timeout=30) + source_projects: dict[str, str | None] = {} + try: + rows_left_untouched = int( + conn.execute(f"SELECT COUNT(*) FROM chunks WHERE NOT {NUMERIC_PROJECT_PREDICATE}").fetchone()[0] + ) + if rollback_path is not None: + _export_rollback(conn, Path(rollback_path)) + + conn.execute("PRAGMA wal_checkpoint(FULL)") + last_rowid = 0 + batches = 0 + rows_updated = rows_rederived = rows_set_null = 0 + while True: + rows = conn.execute( + f""" + SELECT rowid, id, source_file + FROM chunks + WHERE rowid > ? AND {NUMERIC_PROJECT_PREDICATE} + ORDER BY rowid + LIMIT ? + """, + (last_rowid, batch_size), + ).fetchall() + if not rows: + break + conn.execute("BEGIN IMMEDIATE") + try: + for rowid, chunk_id, source_file in rows: + last_rowid = rowid + if source_file not in source_projects: + source_projects[source_file] = _extract_project_from_session_file(source_file) + project = source_projects[source_file] + conn.execute( + f"UPDATE chunks SET project = ? WHERE id = ? AND {NUMERIC_PROJECT_PREDICATE}", + (project, chunk_id), + ) + rows_updated += 1 + if project is None: + rows_set_null += 1 + else: + rows_rederived += 1 + conn.commit() + except Exception: + conn.rollback() + raise + batches += 1 + if batches % 3 == 0: + conn.execute("PRAGMA wal_checkpoint(FULL)") + + conn.execute("PRAGMA wal_checkpoint(FULL)") + return ProjectBackfillResult(rows_updated, rows_rederived, rows_set_null, rows_left_untouched) + finally: + conn.close() + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Re-derive numeric BrainLayer projects from session metadata") + parser.add_argument("--db", required=True, type=Path) + parser.add_argument("--rollback-path", type=Path) + parser.add_argument("--batch-size", type=int, default=_BATCH_SIZE_MIN) + args = parser.parse_args(argv) + print(backfill_numeric_projects(args.db, rollback_path=args.rollback_path, batch_size=args.batch_size)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/brainlayer/watcher_bridge.py b/src/brainlayer/watcher_bridge.py index 982a40da..b9cd0bd7 100644 --- a/src/brainlayer/watcher_bridge.py +++ b/src/brainlayer/watcher_bridge.py @@ -223,23 +223,62 @@ def _normalize_project_name(raw: str) -> str: return name -def _extract_project_from_source(source_file: str) -> str | None: - """Extract the project root from a watcher source path. +def _extract_workspace_project(entry: dict[str, Any] | None) -> str | None: + """Return a project name from explicit session workspace metadata only.""" + if not isinstance(entry, dict): + return None + + candidates = [entry] + for key in ("payload", "metadata"): + value = entry.get(key) + if isinstance(value, dict): + candidates.append(value) + for candidate in candidates: + for key in ("cwd", "workspace", "workdir"): + value = candidate.get(key) + if not isinstance(value, str) or not value.strip(): + continue + name = _normalize_project_name(Path(value).name) + if not name.isdigit(): + return name + return None + + +def _extract_project_from_source(source_file: str, entry: dict[str, Any] | None = None) -> str | None: + """Extract a project only from workspace metadata or Claude's project path. - For Claude Code transcripts the canonical project directory is the segment - immediately under `.../projects/`, even when the JSONL lives under nested - session folders like `subagents/` or `tool-results/`. + Codex, Cursor, and Gemini paths are date-partitioned and therefore cannot + safely provide a project name. Claude Code encodes the workspace explicitly + in its `projects/` path, which remains a trusted fallback. """ p = Path(source_file) parts = p.parts if "projects" in parts: project_index = parts.index("projects") + 1 if project_index < len(parts): - return _normalize_project_name(parts[project_index]) + project = _normalize_project_name(parts[project_index]) + if not project.isdigit(): + return project + return _extract_workspace_project(entry) - parent_name = p.parent.name - if parent_name: - return _normalize_project_name(parent_name) + +def _extract_project_from_session_file(source_file: str) -> str | None: + """Derive a project from the source file's durable session metadata.""" + project = _extract_project_from_source(source_file) + if project is not None: + return project + try: + with Path(source_file).open(encoding="utf-8") as handle: + for line in handle: + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + project = _extract_project_from_source(source_file, entry) + if project is not None: + return project + except OSError: + return None return None @@ -264,6 +303,7 @@ def create_flush_callback(db_path: Path | None = None, *, arbitrated: bool | Non arbitrated = os.environ.get("BRAINLAYER_ARBITRATED") == "1" store = None if arbitrated else VectorStore(db_path or get_db_path()) liveness_schema_ready = False + source_projects: dict[str, str | None] = {} def ensure_direct_liveness_schema() -> None: if liveness_schema_ready or store is None: @@ -338,7 +378,9 @@ def enqueue_chunk( for entry in entries: source_file = entry.get("_source_file", "unknown") source_files_seen.add(source_file) - project = _extract_project_from_source(source_file) + if source_file not in source_projects: + source_projects[source_file] = _extract_project_from_session_file(source_file) + project = source_projects[source_file] claude_conversation_id = _extract_claude_conversation_id(source_file) # Layer 1: Pre-classify filter diff --git a/tests/test_project_backfill.py b/tests/test_project_backfill.py new file mode 100644 index 00000000..49ad795c --- /dev/null +++ b/tests/test_project_backfill.py @@ -0,0 +1,81 @@ +import json +import sqlite3 + +import pytest + +from scripts.d2_project_backfill import backfill_numeric_projects + + +def _create_chunks_db(path, recoverable_source, missing_source, untouched_source): + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE chunks (id TEXT PRIMARY KEY, source_file TEXT NOT NULL, project TEXT)" + ) + conn.execute( + "INSERT INTO chunks VALUES ('recoverable', ?, '30')", + (str(recoverable_source),), + ) + conn.execute( + "INSERT INTO chunks VALUES ('underivable', ?, '7')", + (str(missing_source),), + ) + conn.execute( + "INSERT INTO chunks VALUES ('untouched', ?, 'already-real')", + (str(untouched_source),), + ) + conn.commit() + conn.close() + + +def test_backfill_exports_rollback_derives_projects_and_is_idempotent(tmp_path): + db_path = tmp_path / "brainlayer.db" + source = tmp_path / "sessions" / "recoverable.jsonl" + missing_source = tmp_path / "sessions" / "missing.jsonl" + untouched_source = tmp_path / "sessions" / "untouched.jsonl" + _create_chunks_db(db_path, source, missing_source, untouched_source) + source.parent.mkdir() + source.write_text( + json.dumps({"type": "session_meta", "payload": {"cwd": "/Users/test/Gits/brainlayer"}}) + "\n", + encoding="utf-8", + ) + rollback_path = tmp_path / "rollback.tsv" + + first = backfill_numeric_projects( + db_path, + rollback_path=rollback_path, + batch_size=5_000, + ) + + conn = sqlite3.connect(db_path) + rows_after_first = dict(conn.execute("SELECT id, project FROM chunks")) + conn.close() + assert rows_after_first == { + "recoverable": "brainlayer", + "underivable": None, + "untouched": "already-real", + } + assert first.rows_rederived == 1 + assert first.rows_set_null == 1 + assert first.rows_left_untouched == 1 + assert rollback_path.read_text(encoding="utf-8").splitlines() == [ + "id\tproject", + "recoverable\t30", + "underivable\t7", + ] + + second = backfill_numeric_projects( + db_path, + batch_size=5_000, + ) + + conn = sqlite3.connect(db_path) + rows_after_second = dict(conn.execute("SELECT id, project FROM chunks")) + conn.close() + assert second.rows_updated == 0 + assert rows_after_second == rows_after_first + + +@pytest.mark.parametrize("batch_size", (4_999, 10_001)) +def test_backfill_requires_production_safe_batch_sizes(tmp_path, batch_size): + with pytest.raises(ValueError, match="5,000 and 10,000"): + backfill_numeric_projects(tmp_path / "brainlayer.db", batch_size=batch_size) diff --git a/tests/test_watcher_bridge.py b/tests/test_watcher_bridge.py index 43c9cee7..16f92ad2 100644 --- a/tests/test_watcher_bridge.py +++ b/tests/test_watcher_bridge.py @@ -151,6 +151,35 @@ def test_extract_from_brainlayer_grill_top_level_file(self): path = "/Users/etanheyman/.claude/projects/-Users-etanheyman-Gits-brainlayer-grill/abc123.jsonl" assert _extract_project_from_source(path) == "brainlayer-grill" + def test_claude_projects_path_wins_over_session_cwd(self): + path = "/Users/etanheyman/.claude/projects/-Users-etanheyman-Gits-brainlayer-grill/abc123.jsonl" + entry = {"cwd": "/Users/etanheyman/Gits/grill"} + + assert _extract_project_from_source(path, entry) == "brainlayer-grill" + + def test_extracts_codex_date_partitioned_path_from_session_cwd(self): + path = "/Users/etanheyman/.codex/sessions/2026/07/30/rollout-abc.jsonl" + entry = {"payload": {"cwd": "/Users/etanheyman/Gits/brainlayer"}} + + assert _extract_project_from_source(path, entry) == "brainlayer" + + def test_extracts_cursor_path_from_session_cwd(self): + path = "/Users/etanheyman/.cursor/sessions/2026/07/30/agent.jsonl" + entry = {"cwd": "/Users/etanheyman/Gits/cursor-project"} + + assert _extract_project_from_source(path, entry) == "cursor-project" + + def test_extracts_gemini_path_from_session_cwd(self): + path = "/Users/etanheyman/.gemini/sessions/2026/07/30/session.jsonl" + entry = {"payload": {"workspace": "/Users/etanheyman/Gits/gemini-project"}} + + assert _extract_project_from_source(path, entry) == "gemini-project" + + def test_returns_none_for_path_without_workspace_signal(self): + path = "/Users/etanheyman/.codex/sessions/2026/07/30/rollout-abc.jsonl" + + assert _extract_project_from_source(path) is None + def test_extract_claude_conversation_id_from_top_level_jsonl(self): path = ( "/Users/etanheyman/.claude/projects/-Users-etanheyman-Gits-brainlayer/" @@ -170,6 +199,32 @@ def test_extract_claude_conversation_id_from_nested_subagent_jsonl(self): class TestFlushCallback: + def test_codex_session_metadata_derives_project_for_normalized_entries(self, tmp_path, monkeypatch): + db_path = tmp_path / "test.db" + queue_dir = tmp_path / "queue" + VectorStore(db_path).close() + monkeypatch.setenv("BRAINLAYER_QUEUE_DIR", str(queue_dir)) + source_file = tmp_path / ".codex" / "sessions" / "2026" / "07" / "31" / "rollout.jsonl" + source_file.parent.mkdir(parents=True) + source_file.write_text( + json.dumps( + { + "type": "session_meta", + "timestamp": "2026-07-31T12:26:57.941Z", + "payload": {"cwd": "/Users/etanheyman/Gits/t3code"}, + } + ) + + "\n", + encoding="utf-8", + ) + entry = _make_jsonl_entry(text=_LONG_TEXT, entry_type="assistant") + entry["_source_file"] = str(source_file) + + create_flush_callback(db_path, arbitrated=True)([entry]) + + queued = json.loads(next(queue_dir.glob("watcher-*.jsonl")).read_text(encoding="utf-8")) + assert queued["project"] == "t3code" + def test_flush_scrubs_secrets_before_hash_and_persistence(self, tmp_path, monkeypatch): db_path = tmp_path / "test.db" queue_dir = tmp_path / "queue" From 06b73497cbc020da2e81517ff0569509a6cc543b Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 1 Aug 2026 18:02:04 +0300 Subject: [PATCH 2/5] fix: harden project derivation backfill --- ...2026-08-01-d2-project-derivation-design.md | 77 ----------- scripts/d2_project_backfill.py | 128 +++++++++++------- src/brainlayer/watcher_bridge.py | 12 +- tests/test_project_backfill.py | 94 ++++++++++++- tests/test_watcher_bridge.py | 38 ++++++ 5 files changed, 212 insertions(+), 137 deletions(-) delete mode 100644 docs/plans/2026-08-01-d2-project-derivation-design.md diff --git a/docs/plans/2026-08-01-d2-project-derivation-design.md b/docs/plans/2026-08-01-d2-project-derivation-design.md deleted file mode 100644 index 86669499..00000000 --- a/docs/plans/2026-08-01-d2-project-derivation-design.md +++ /dev/null @@ -1,77 +0,0 @@ -# D2 Project Derivation Implementation Plan - -> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. - -**Goal:** Derive watcher chunk projects only from trustworthy workspace signals and safely repair the existing numeric-project corruption. - -**Architecture:** Keep source-path parsing only for Claude's explicit `projects/` workspace encoding. For every other provider, obtain the project from session-record workspace metadata, normalize the workspace basename, and return `None` when no such signal is present. A standalone, resumable migration will update only bare one- or two-digit projects, export a narrow rollback TSV first, batch updates, and record its own audit/progress state. - -**Tech Stack:** Python 3, SQLite/APSW, pytest. - ---- - -### Task 1: Define derivation regressions - -**Files:** -- Modify: `tests/test_watcher_bridge.py` -- Modify: `src/brainlayer/watcher_bridge.py` - -**Step 1: Write failing tests** - -Add tests for a Codex date-partitioned source with a metadata `cwd`, a Claude `projects/` source, Cursor and Gemini sources with a metadata `cwd`, and a source with no derivable workspace. Assert no provider falls back to a numeric directory. - -**Step 2: Run to verify failure** - -Run: `pytest tests/test_watcher_bridge.py::TestProjectExtraction -v` - -Expected: FAIL because extraction has no metadata input and returns the date segment. - -**Step 3: Implement minimal derivation** - -Change `_extract_project_from_source` to accept an optional entry/metadata workspace signal, preserve the Claude `projects/` parser, and reject pure numeric candidates in all paths. Pass each watcher entry to the helper. - -**Step 4: Run to verify pass** - -Run: `pytest tests/test_watcher_bridge.py::TestProjectExtraction -v` - -Expected: PASS. - -### Task 2: Implement the narrow, idempotent backfill - -**Files:** -- Create: `src/brainlayer/project_backfill.py` -- Create: `tests/test_project_backfill.py` - -**Step 1: Write failing migration tests** - -Create a fixture DB containing numeric projects with recoverable and unrecoverable source sessions plus a non-numeric project. Test TSV rollback export, real-project/NULL/untouched counts, and a second migration run producing zero changes with identical counts. - -**Step 2: Run to verify failure** - -Run: `pytest tests/test_project_backfill.py -v` - -Expected: FAIL because the migration module does not exist. - -**Step 3: Implement minimal migration** - -Create a migration that reads session metadata from source JSONL files, updates only `project GLOB '[0-9]' OR project GLOB '[0-9][0-9]'`, uses 5–10K transactions, checkpoints every three batches, stores a migration audit record, and offers a CLI with an explicit live-DB opt-in. - -**Step 4: Run to verify pass** - -Run: `pytest tests/test_project_backfill.py -v` - -Expected: PASS. - -### Task 3: Verify and execute the production pass - -**Files:** -- Create: `REPORT.md` -- Create: `docs.local/tasks/d2-rollback-.tsv` (untracked rollback artifact) - -**Step 1:** Run the focused and full pytest suites. - -**Step 2:** Stop enrichment writers and claim the shared bulk-DB lock in `docs.local/.../collab.md`. - -**Step 3:** Re-count numeric projects; export the rollback TSV; run `PRAGMA wal_checkpoint(FULL)`; execute the migration in batches; checkpoint every three batches and at completion; release the lock. - -**Step 4:** Independently query the three-way split, remaining numeric projects, and distinct projects before/after. Write `REPORT.md` with command evidence and commit only repository changes (not the live DB artifact). diff --git a/scripts/d2_project_backfill.py b/scripts/d2_project_backfill.py index bc50e551..f44f83db 100644 --- a/scripts/d2_project_backfill.py +++ b/scripts/d2_project_backfill.py @@ -5,9 +5,12 @@ import argparse import csv import sqlite3 +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path +from typing import Iterator +from brainlayer.vector_store import VectorStore from brainlayer.watcher_bridge import _extract_project_from_session_file NUMERIC_PROJECT_PREDICATE = "(project GLOB '[0-9]' OR project GLOB '[0-9][0-9]')" @@ -35,6 +38,19 @@ def _export_rollback(conn: sqlite3.Connection, rollback_path: Path) -> None: ) +@contextmanager +def _operation_writer_lock(db_path: Path) -> Iterator[None]: + """Acquire BrainLayer's standard pidfile without opening a second DB writer.""" + store = VectorStore.__new__(VectorStore) + store.db_path = db_path + store._writer_pidfile_acquired = False + store._acquire_writer_pidfile() + try: + yield + finally: + store._release_writer_pidfile() + + def backfill_numeric_projects( db_path: str | Path, *, @@ -44,67 +60,75 @@ def backfill_numeric_projects( """Update only one- and two-digit projects from recorded session metadata.""" if not _BATCH_SIZE_MIN <= batch_size <= _BATCH_SIZE_MAX: raise ValueError("batch_size must be between 5,000 and 10,000") - - conn = sqlite3.connect(Path(db_path), timeout=30) - source_projects: dict[str, str | None] = {} - try: - rows_left_untouched = int( - conn.execute(f"SELECT COUNT(*) FROM chunks WHERE NOT {NUMERIC_PROJECT_PREDICATE}").fetchone()[0] - ) - if rollback_path is not None: + if rollback_path is None: + raise ValueError("rollback_path is required before backfilling numeric projects") + + db_path = Path(db_path) + with _operation_writer_lock(db_path): + conn = sqlite3.connect(db_path, timeout=30) + source_projects: dict[str, str | None] = {} + try: + rows_left_untouched = int( + conn.execute(f"SELECT COUNT(*) FROM chunks WHERE NOT {NUMERIC_PROJECT_PREDICATE}").fetchone()[0] + ) _export_rollback(conn, Path(rollback_path)) - conn.execute("PRAGMA wal_checkpoint(FULL)") - last_rowid = 0 - batches = 0 - rows_updated = rows_rederived = rows_set_null = 0 - while True: - rows = conn.execute( - f""" - SELECT rowid, id, source_file - FROM chunks - WHERE rowid > ? AND {NUMERIC_PROJECT_PREDICATE} - ORDER BY rowid - LIMIT ? - """, - (last_rowid, batch_size), - ).fetchall() - if not rows: - break - conn.execute("BEGIN IMMEDIATE") - try: - for rowid, chunk_id, source_file in rows: - last_rowid = rowid + conn.execute("PRAGMA wal_checkpoint(FULL)") + last_rowid = 0 + batches = 0 + rows_updated = rows_rederived = rows_set_null = 0 + while True: + rows = conn.execute( + f""" + SELECT rowid, id, source_file + FROM chunks + WHERE rowid > ? AND {NUMERIC_PROJECT_PREDICATE} + ORDER BY rowid + LIMIT ? + """, + (last_rowid, batch_size), + ).fetchall() + if not rows: + break + + projects_by_rowid: dict[int, str | None] = {} + for rowid, _chunk_id, source_file in rows: if source_file not in source_projects: source_projects[source_file] = _extract_project_from_session_file(source_file) - project = source_projects[source_file] - conn.execute( - f"UPDATE chunks SET project = ? WHERE id = ? AND {NUMERIC_PROJECT_PREDICATE}", - (project, chunk_id), - ) - rows_updated += 1 - if project is None: - rows_set_null += 1 - else: - rows_rederived += 1 - conn.commit() - except Exception: - conn.rollback() - raise - batches += 1 - if batches % 3 == 0: - conn.execute("PRAGMA wal_checkpoint(FULL)") - - conn.execute("PRAGMA wal_checkpoint(FULL)") - return ProjectBackfillResult(rows_updated, rows_rederived, rows_set_null, rows_left_untouched) - finally: - conn.close() + projects_by_rowid[rowid] = source_projects[source_file] + last_rowid = rows[-1][0] + + conn.execute("BEGIN IMMEDIATE") + try: + for rowid, chunk_id, _source_file in rows: + project = projects_by_rowid[rowid] + conn.execute( + f"UPDATE chunks SET project = ? WHERE id = ? AND {NUMERIC_PROJECT_PREDICATE}", + (project, chunk_id), + ) + rows_updated += 1 + if project is None: + rows_set_null += 1 + else: + rows_rederived += 1 + conn.commit() + except Exception: + conn.rollback() + raise + batches += 1 + if batches % 3 == 0: + conn.execute("PRAGMA wal_checkpoint(FULL)") + + conn.execute("PRAGMA wal_checkpoint(FULL)") + return ProjectBackfillResult(rows_updated, rows_rederived, rows_set_null, rows_left_untouched) + finally: + conn.close() def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Re-derive numeric BrainLayer projects from session metadata") parser.add_argument("--db", required=True, type=Path) - parser.add_argument("--rollback-path", type=Path) + parser.add_argument("--rollback-path", required=True, type=Path) parser.add_argument("--batch-size", type=int, default=_BATCH_SIZE_MIN) args = parser.parse_args(argv) print(backfill_numeric_projects(args.db, rollback_path=args.rollback_path, batch_size=args.batch_size)) diff --git a/src/brainlayer/watcher_bridge.py b/src/brainlayer/watcher_bridge.py index b9cd0bd7..9cd6a2a2 100644 --- a/src/brainlayer/watcher_bridge.py +++ b/src/brainlayer/watcher_bridge.py @@ -239,7 +239,7 @@ def _extract_workspace_project(entry: dict[str, Any] | None) -> str | None: if not isinstance(value, str) or not value.strip(): continue name = _normalize_project_name(Path(value).name) - if not name.isdigit(): + if name and not name.isdigit(): return name return None @@ -303,7 +303,7 @@ def create_flush_callback(db_path: Path | None = None, *, arbitrated: bool | Non arbitrated = os.environ.get("BRAINLAYER_ARBITRATED") == "1" store = None if arbitrated else VectorStore(db_path or get_db_path()) liveness_schema_ready = False - source_projects: dict[str, str | None] = {} + source_projects: dict[str, str] = {} def ensure_direct_liveness_schema() -> None: if liveness_schema_ready or store is None: @@ -378,9 +378,11 @@ def enqueue_chunk( for entry in entries: source_file = entry.get("_source_file", "unknown") source_files_seen.add(source_file) - if source_file not in source_projects: - source_projects[source_file] = _extract_project_from_session_file(source_file) - project = source_projects[source_file] + project = source_projects.get(source_file) + if project is None: + project = _extract_project_from_session_file(source_file) + if project is not None: + source_projects[source_file] = project claude_conversation_id = _extract_claude_conversation_id(source_file) # Layer 1: Pre-classify filter diff --git a/tests/test_project_backfill.py b/tests/test_project_backfill.py index 49ad795c..9d8651d7 100644 --- a/tests/test_project_backfill.py +++ b/tests/test_project_backfill.py @@ -1,16 +1,19 @@ import json +import os import sqlite3 +import subprocess +import sys +from pathlib import Path import pytest +from brainlayer.vector_store import WriterInUseError from scripts.d2_project_backfill import backfill_numeric_projects def _create_chunks_db(path, recoverable_source, missing_source, untouched_source): conn = sqlite3.connect(path) - conn.execute( - "CREATE TABLE chunks (id TEXT PRIMARY KEY, source_file TEXT NOT NULL, project TEXT)" - ) + conn.execute("CREATE TABLE chunks (id TEXT PRIMARY KEY, source_file TEXT NOT NULL, project TEXT)") conn.execute( "INSERT INTO chunks VALUES ('recoverable', ?, '30')", (str(recoverable_source),), @@ -65,6 +68,7 @@ def test_backfill_exports_rollback_derives_projects_and_is_idempotent(tmp_path): second = backfill_numeric_projects( db_path, + rollback_path=tmp_path / "second-rollback.tsv", batch_size=5_000, ) @@ -75,6 +79,90 @@ def test_backfill_exports_rollback_derives_projects_and_is_idempotent(tmp_path): assert rows_after_second == rows_after_first +def test_backfill_requires_rollback_artifact_before_mutating_rows(tmp_path): + db_path = tmp_path / "brainlayer.db" + source = tmp_path / "sessions" / "recoverable.jsonl" + missing_source = tmp_path / "sessions" / "missing.jsonl" + untouched_source = tmp_path / "sessions" / "untouched.jsonl" + _create_chunks_db(db_path, source, missing_source, untouched_source) + + with pytest.raises(ValueError, match="rollback_path is required"): + backfill_numeric_projects(db_path) + + conn = sqlite3.connect(db_path) + assert dict(conn.execute("SELECT id, project FROM chunks"))["recoverable"] == "30" + conn.close() + + +def test_backfill_does_not_mutate_when_rollback_artifact_cannot_be_created(tmp_path): + db_path = tmp_path / "brainlayer.db" + source = tmp_path / "sessions" / "recoverable.jsonl" + missing_source = tmp_path / "sessions" / "missing.jsonl" + untouched_source = tmp_path / "sessions" / "untouched.jsonl" + _create_chunks_db(db_path, source, missing_source, untouched_source) + rollback_path = tmp_path / "rollback.tsv" + rollback_path.write_text("existing artifact\n", encoding="utf-8") + + with pytest.raises(FileExistsError, match="rollback artifact already exists"): + backfill_numeric_projects(db_path, rollback_path=rollback_path) + + conn = sqlite3.connect(db_path) + assert dict(conn.execute("SELECT id, project FROM chunks"))["recoverable"] == "30" + conn.close() + + +def test_backfill_refuses_while_standard_writer_lock_is_held(tmp_path, monkeypatch): + db_path = tmp_path / "brainlayer.db" + source = tmp_path / "sessions" / "recoverable.jsonl" + missing_source = tmp_path / "sessions" / "missing.jsonl" + untouched_source = tmp_path / "sessions" / "untouched.jsonl" + _create_chunks_db(db_path, source, missing_source, untouched_source) + pidfile_dir = tmp_path / "pidfiles" + monkeypatch.setenv("BRAINLAYER_WRITER_PIDFILE_DIR", str(pidfile_dir)) + repo_root = Path(__file__).resolve().parents[1] + holder = subprocess.Popen( + [ + sys.executable, + "-c", + """ +import sys +from pathlib import Path +from brainlayer.vector_store import VectorStore + +store = VectorStore.__new__(VectorStore) +store.db_path = Path(sys.argv[1]) +store._writer_pidfile_acquired = False +store._acquire_writer_pidfile() +print("ready", flush=True) +sys.stdin.readline() +store._release_writer_pidfile() +""", + str(db_path), + ], + env={ + **os.environ, + "PYTHONPATH": f"{repo_root / 'src'}:{os.environ.get('PYTHONPATH', '')}", + }, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "ready" + with pytest.raises(WriterInUseError, match="another writer is using"): + backfill_numeric_projects(db_path, rollback_path=tmp_path / "rollback.tsv") + finally: + assert holder.stdin is not None + holder.stdin.write("release\n") + holder.stdin.close() + holder.wait(timeout=5) + + conn = sqlite3.connect(db_path) + assert dict(conn.execute("SELECT id, project FROM chunks"))["recoverable"] == "30" + conn.close() + + @pytest.mark.parametrize("batch_size", (4_999, 10_001)) def test_backfill_requires_production_safe_batch_sizes(tmp_path, batch_size): with pytest.raises(ValueError, match="5,000 and 10,000"): diff --git a/tests/test_watcher_bridge.py b/tests/test_watcher_bridge.py index 16f92ad2..fdd14ada 100644 --- a/tests/test_watcher_bridge.py +++ b/tests/test_watcher_bridge.py @@ -175,6 +175,12 @@ def test_extracts_gemini_path_from_session_cwd(self): assert _extract_project_from_source(path, entry) == "gemini-project" + def test_returns_none_for_root_workspace(self): + path = "/Users/etanheyman/.codex/sessions/2026/07/30/rollout-abc.jsonl" + entry = {"payload": {"cwd": "/"}} + + assert _extract_project_from_source(path, entry) is None + def test_returns_none_for_path_without_workspace_signal(self): path = "/Users/etanheyman/.codex/sessions/2026/07/30/rollout-abc.jsonl" @@ -225,6 +231,38 @@ def test_codex_session_metadata_derives_project_for_normalized_entries(self, tmp queued = json.loads(next(queue_dir.glob("watcher-*.jsonl")).read_text(encoding="utf-8")) assert queued["project"] == "t3code" + def test_retries_source_file_after_metadata_is_appended(self, tmp_path, monkeypatch): + db_path = tmp_path / "test.db" + queue_dir = tmp_path / "queue" + VectorStore(db_path).close() + monkeypatch.setenv("BRAINLAYER_QUEUE_DIR", str(queue_dir)) + source_file = tmp_path / ".codex" / "sessions" / "2026" / "07" / "31" / "rollout.jsonl" + source_file.parent.mkdir(parents=True) + source_file.write_text("", encoding="utf-8") + flush = create_flush_callback(db_path, arbitrated=True) + + first = _make_jsonl_entry(text=_LONG_TEXT, entry_type="assistant") + first["_source_file"] = str(source_file) + flush([first]) + + with source_file.open("a", encoding="utf-8") as handle: + handle.write( + json.dumps( + { + "type": "session_meta", + "timestamp": "2026-07-31T12:26:57.941Z", + "payload": {"cwd": "/Users/etanheyman/Gits/t3code"}, + } + ) + + "\n" + ) + second = _make_jsonl_entry(text=_LONG_TEXT + " after metadata", entry_type="assistant") + second["_source_file"] = str(source_file) + flush([second]) + + queued = [json.loads(path.read_text(encoding="utf-8")) for path in queue_dir.glob("watcher-*.jsonl")] + assert {item["project"] for item in queued} == {None, "t3code"} + def test_flush_scrubs_secrets_before_hash_and_persistence(self, tmp_path, monkeypatch): db_path = tmp_path / "test.db" queue_dir = tmp_path / "queue" From 9f3cf43171382e9514e999bbecda6dc30bf5244a Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 1 Aug 2026 18:07:23 +0300 Subject: [PATCH 3/5] fix: address project backfill review findings --- scripts/d2_project_backfill.py | 17 ++++++---- src/brainlayer/watcher_bridge.py | 24 ++++++++++++-- tests/test_project_backfill.py | 56 ++++++++++++++++++++++++++++---- tests/test_watcher_bridge.py | 17 ++++++++++ 4 files changed, 98 insertions(+), 16 deletions(-) diff --git a/scripts/d2_project_backfill.py b/scripts/d2_project_backfill.py index f44f83db..93622d21 100644 --- a/scripts/d2_project_backfill.py +++ b/scripts/d2_project_backfill.py @@ -69,7 +69,9 @@ def backfill_numeric_projects( source_projects: dict[str, str | None] = {} try: rows_left_untouched = int( - conn.execute(f"SELECT COUNT(*) FROM chunks WHERE NOT {NUMERIC_PROJECT_PREDICATE}").fetchone()[0] + conn.execute( + f"SELECT COUNT(*) FROM chunks WHERE COALESCE({NUMERIC_PROJECT_PREDICATE}, 0) = 0" + ).fetchone()[0] ) _export_rollback(conn, Path(rollback_path)) @@ -102,15 +104,16 @@ def backfill_numeric_projects( try: for rowid, chunk_id, _source_file in rows: project = projects_by_rowid[rowid] - conn.execute( + update = conn.execute( f"UPDATE chunks SET project = ? WHERE id = ? AND {NUMERIC_PROJECT_PREDICATE}", (project, chunk_id), ) - rows_updated += 1 - if project is None: - rows_set_null += 1 - else: - rows_rederived += 1 + if update.rowcount == 1: + rows_updated += 1 + if project is None: + rows_set_null += 1 + else: + rows_rederived += 1 conn.commit() except Exception: conn.rollback() diff --git a/src/brainlayer/watcher_bridge.py b/src/brainlayer/watcher_bridge.py index 9cd6a2a2..e887c0a2 100644 --- a/src/brainlayer/watcher_bridge.py +++ b/src/brainlayer/watcher_bridge.py @@ -282,6 +282,15 @@ def _extract_project_from_session_file(source_file: str) -> str | None: return None +def _source_file_fingerprint(source_file: str) -> tuple[int, int] | None: + """Return a version token for retrying an unresolved session file.""" + try: + stat = Path(source_file).stat() + except OSError: + return None + return stat.st_size, stat.st_mtime_ns + + def _content_class_for_visibility(base_content_class: str, visibility: str) -> str: if visibility == "default": return base_content_class @@ -304,6 +313,7 @@ def create_flush_callback(db_path: Path | None = None, *, arbitrated: bool | Non store = None if arbitrated else VectorStore(db_path or get_db_path()) liveness_schema_ready = False source_projects: dict[str, str] = {} + unresolved_source_versions: dict[str, tuple[int, int] | None] = {} def ensure_direct_liveness_schema() -> None: if liveness_schema_ready or store is None: @@ -380,9 +390,17 @@ def enqueue_chunk( source_files_seen.add(source_file) project = source_projects.get(source_file) if project is None: - project = _extract_project_from_session_file(source_file) - if project is not None: - source_projects[source_file] = project + source_version = _source_file_fingerprint(source_file) + if ( + source_file not in unresolved_source_versions + or unresolved_source_versions[source_file] != source_version + ): + project = _extract_project_from_session_file(source_file) + if project is not None: + source_projects[source_file] = project + unresolved_source_versions.pop(source_file, None) + else: + unresolved_source_versions[source_file] = source_version claude_conversation_id = _extract_claude_conversation_id(source_file) # Layer 1: Pre-classify filter diff --git a/tests/test_project_backfill.py b/tests/test_project_backfill.py index 9d8651d7..f1692afa 100644 --- a/tests/test_project_backfill.py +++ b/tests/test_project_backfill.py @@ -7,6 +7,7 @@ import pytest +import scripts.d2_project_backfill as project_backfill from brainlayer.vector_store import WriterInUseError from scripts.d2_project_backfill import backfill_numeric_projects @@ -26,6 +27,10 @@ def _create_chunks_db(path, recoverable_source, missing_source, untouched_source "INSERT INTO chunks VALUES ('untouched', ?, 'already-real')", (str(untouched_source),), ) + conn.execute( + "INSERT INTO chunks VALUES ('already-null', ?, NULL)", + (str(untouched_source),), + ) conn.commit() conn.close() @@ -53,13 +58,14 @@ def test_backfill_exports_rollback_derives_projects_and_is_idempotent(tmp_path): rows_after_first = dict(conn.execute("SELECT id, project FROM chunks")) conn.close() assert rows_after_first == { + "already-null": None, "recoverable": "brainlayer", "underivable": None, "untouched": "already-real", } assert first.rows_rederived == 1 assert first.rows_set_null == 1 - assert first.rows_left_untouched == 1 + assert first.rows_left_untouched == 2 assert rollback_path.read_text(encoding="utf-8").splitlines() == [ "id\tproject", "recoverable\t30", @@ -145,24 +151,62 @@ def test_backfill_refuses_while_standard_writer_lock_is_held(tmp_path, monkeypat }, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, ) try: assert holder.stdout is not None - assert holder.stdout.readline().strip() == "ready" + readiness = holder.stdout.readline().strip() + if readiness != "ready": + assert holder.stderr is not None + stderr = holder.stderr.read() if holder.poll() is not None else "holder did not exit" + pytest.fail(f"writer lock holder did not become ready: {stderr}") with pytest.raises(WriterInUseError, match="another writer is using"): backfill_numeric_projects(db_path, rollback_path=tmp_path / "rollback.tsv") finally: - assert holder.stdin is not None - holder.stdin.write("release\n") - holder.stdin.close() - holder.wait(timeout=5) + if holder.poll() is None and holder.stdin is not None: + try: + holder.stdin.write("release\n") + holder.stdin.close() + holder.wait(timeout=5) + except (OSError, subprocess.TimeoutExpired): + holder.terminate() + holder.wait(timeout=5) conn = sqlite3.connect(db_path) assert dict(conn.execute("SELECT id, project FROM chunks"))["recoverable"] == "30" conn.close() +def test_backfill_counts_only_rows_changed_by_numeric_guard(tmp_path, monkeypatch): + db_path = tmp_path / "brainlayer.db" + source = tmp_path / "sessions" / "recoverable.jsonl" + missing_source = tmp_path / "sessions" / "missing.jsonl" + untouched_source = tmp_path / "sessions" / "untouched.jsonl" + _create_chunks_db(db_path, source, missing_source, untouched_source) + source.parent.mkdir() + source.write_text( + json.dumps({"type": "session_meta", "payload": {"cwd": "/Users/test/Gits/brainlayer"}}) + "\n", + encoding="utf-8", + ) + original_extract = project_backfill._extract_project_from_session_file + + def change_project_before_write(source_file): + if source_file == str(source): + conn = sqlite3.connect(db_path) + conn.execute("UPDATE chunks SET project = 'already-real' WHERE id = 'recoverable'") + conn.commit() + conn.close() + return original_extract(source_file) + + monkeypatch.setattr(project_backfill, "_extract_project_from_session_file", change_project_before_write) + result = backfill_numeric_projects(db_path, rollback_path=tmp_path / "rollback.tsv") + + assert result.rows_updated == 1 + assert result.rows_rederived == 0 + assert result.rows_set_null == 1 + + @pytest.mark.parametrize("batch_size", (4_999, 10_001)) def test_backfill_requires_production_safe_batch_sizes(tmp_path, batch_size): with pytest.raises(ValueError, match="5,000 and 10,000"): diff --git a/tests/test_watcher_bridge.py b/tests/test_watcher_bridge.py index fdd14ada..3efb3e6d 100644 --- a/tests/test_watcher_bridge.py +++ b/tests/test_watcher_bridge.py @@ -232,6 +232,8 @@ def test_codex_session_metadata_derives_project_for_normalized_entries(self, tmp assert queued["project"] == "t3code" def test_retries_source_file_after_metadata_is_appended(self, tmp_path, monkeypatch): + import brainlayer.watcher_bridge as bridge + db_path = tmp_path / "test.db" queue_dir = tmp_path / "queue" VectorStore(db_path).close() @@ -240,11 +242,25 @@ def test_retries_source_file_after_metadata_is_appended(self, tmp_path, monkeypa source_file.parent.mkdir(parents=True) source_file.write_text("", encoding="utf-8") flush = create_flush_callback(db_path, arbitrated=True) + original_extract = bridge._extract_project_from_session_file + extraction_calls = 0 + + def count_extractions(path): + nonlocal extraction_calls + extraction_calls += 1 + return original_extract(path) + + monkeypatch.setattr(bridge, "_extract_project_from_session_file", count_extractions) first = _make_jsonl_entry(text=_LONG_TEXT, entry_type="assistant") first["_source_file"] = str(source_file) flush([first]) + unchanged = _make_jsonl_entry(text=_LONG_TEXT + " before metadata", entry_type="assistant") + unchanged["_source_file"] = str(source_file) + flush([unchanged]) + assert extraction_calls == 1 + with source_file.open("a", encoding="utf-8") as handle: handle.write( json.dumps( @@ -259,6 +275,7 @@ def test_retries_source_file_after_metadata_is_appended(self, tmp_path, monkeypa second = _make_jsonl_entry(text=_LONG_TEXT + " after metadata", entry_type="assistant") second["_source_file"] = str(source_file) flush([second]) + assert extraction_calls == 2 queued = [json.loads(path.read_text(encoding="utf-8")) for path in queue_dir.glob("watcher-*.jsonl")] assert {item["project"] for item in queued} == {None, "t3code"} From 4a7c6c3d02203d61ac587e4e424d06b325542c36 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 1 Aug 2026 18:08:36 +0300 Subject: [PATCH 4/5] fix: invalidate replaced watcher sources --- src/brainlayer/watcher_bridge.py | 21 +++++++++++++++++++++ tests/test_watcher_bridge.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/src/brainlayer/watcher_bridge.py b/src/brainlayer/watcher_bridge.py index e887c0a2..e67136a3 100644 --- a/src/brainlayer/watcher_bridge.py +++ b/src/brainlayer/watcher_bridge.py @@ -291,6 +291,15 @@ def _source_file_fingerprint(source_file: str) -> tuple[int, int] | None: return stat.st_size, stat.st_mtime_ns +def _source_file_identity(source_file: str) -> tuple[int, int, int] | None: + """Return enough state to detect replacement or rewind without invalidating on append.""" + try: + stat = Path(source_file).stat() + except OSError: + return None + return stat.st_dev, stat.st_ino, stat.st_size + + def _content_class_for_visibility(base_content_class: str, visibility: str) -> str: if visibility == "default": return base_content_class @@ -313,6 +322,7 @@ def create_flush_callback(db_path: Path | None = None, *, arbitrated: bool | Non store = None if arbitrated else VectorStore(db_path or get_db_path()) liveness_schema_ready = False source_projects: dict[str, str] = {} + resolved_source_identities: dict[str, tuple[int, int, int]] = {} unresolved_source_versions: dict[str, tuple[int, int] | None] = {} def ensure_direct_liveness_schema() -> None: @@ -389,6 +399,15 @@ def enqueue_chunk( source_file = entry.get("_source_file", "unknown") source_files_seen.add(source_file) project = source_projects.get(source_file) + source_identity = _source_file_identity(source_file) + cached_identity = resolved_source_identities.get(source_file) + if project is not None and source_identity is not None and cached_identity is not None: + if source_identity[:2] != cached_identity[:2] or source_identity[2] < cached_identity[2]: + source_projects.pop(source_file, None) + resolved_source_identities.pop(source_file, None) + project = None + else: + resolved_source_identities[source_file] = source_identity if project is None: source_version = _source_file_fingerprint(source_file) if ( @@ -398,6 +417,8 @@ def enqueue_chunk( project = _extract_project_from_session_file(source_file) if project is not None: source_projects[source_file] = project + if source_identity is not None: + resolved_source_identities[source_file] = source_identity unresolved_source_versions.pop(source_file, None) else: unresolved_source_versions[source_file] = source_version diff --git a/tests/test_watcher_bridge.py b/tests/test_watcher_bridge.py index 3efb3e6d..0bc0c7e6 100644 --- a/tests/test_watcher_bridge.py +++ b/tests/test_watcher_bridge.py @@ -280,6 +280,36 @@ def count_extractions(path): queued = [json.loads(path.read_text(encoding="utf-8")) for path in queue_dir.glob("watcher-*.jsonl")] assert {item["project"] for item in queued} == {None, "t3code"} + def test_rederives_project_when_source_file_is_replaced(self, tmp_path, monkeypatch): + db_path = tmp_path / "test.db" + queue_dir = tmp_path / "queue" + VectorStore(db_path).close() + monkeypatch.setenv("BRAINLAYER_QUEUE_DIR", str(queue_dir)) + source_file = tmp_path / ".codex" / "sessions" / "2026" / "07" / "31" / "rollout.jsonl" + source_file.parent.mkdir(parents=True) + source_file.write_text( + json.dumps({"type": "session_meta", "payload": {"cwd": "/Users/etanheyman/Gits/t3code"}}) + "\n", + encoding="utf-8", + ) + flush = create_flush_callback(db_path, arbitrated=True) + + first = _make_jsonl_entry(text=_LONG_TEXT, entry_type="assistant") + first["_source_file"] = str(source_file) + flush([first]) + + replacement = tmp_path / "replacement.jsonl" + replacement.write_text( + json.dumps({"type": "session_meta", "payload": {"cwd": "/Users/etanheyman/Gits/brainlayer"}}) + "\n", + encoding="utf-8", + ) + replacement.replace(source_file) + second = _make_jsonl_entry(text=_LONG_TEXT + " after replacement", entry_type="assistant") + second["_source_file"] = str(source_file) + flush([second]) + + queued = [json.loads(path.read_text(encoding="utf-8")) for path in queue_dir.glob("watcher-*.jsonl")] + assert {item["project"] for item in queued} == {"t3code", "brainlayer"} + def test_flush_scrubs_secrets_before_hash_and_persistence(self, tmp_path, monkeypatch): db_path = tmp_path / "test.db" queue_dir = tmp_path / "queue" From 57afe1a6500419c7a9e9e88cb9e5736ce88010a9 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sat, 1 Aug 2026 18:14:08 +0300 Subject: [PATCH 5/5] fix: constrain MCP SDK to supported major --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 3dac39be..9b87bd77 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ dependencies = [ "sentence-transformers>=2.2.0", # MCP server - "mcp>=1.0.0", + "mcp>=1.0.0,<2.0.0", # CLI "typer>=0.9.0",