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", diff --git a/scripts/d2_project_backfill.py b/scripts/d2_project_backfill.py new file mode 100644 index 00000000..93622d21 --- /dev/null +++ b/scripts/d2_project_backfill.py @@ -0,0 +1,142 @@ +"""One-off repair for chunks whose project was derived from a date directory.""" + +from __future__ import annotations + +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]')" +_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") + ) + + +@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, + *, + 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") + 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 COALESCE({NUMERIC_PROJECT_PREDICATE}, 0) = 0" + ).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 + + 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) + 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] + update = conn.execute( + f"UPDATE chunks SET project = ? WHERE id = ? AND {NUMERIC_PROJECT_PREDICATE}", + (project, chunk_id), + ) + 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() + 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", 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)) + 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..e67136a3 100644 --- a/src/brainlayer/watcher_bridge.py +++ b/src/brainlayer/watcher_bridge.py @@ -223,26 +223,83 @@ 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 name and 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 +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 _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 @@ -264,6 +321,9 @@ 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] = {} + resolved_source_identities: dict[str, tuple[int, int, int]] = {} + unresolved_source_versions: dict[str, tuple[int, int] | None] = {} def ensure_direct_liveness_schema() -> None: if liveness_schema_ready or store is None: @@ -338,7 +398,30 @@ 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) + 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 ( + 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 + 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 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..f1692afa --- /dev/null +++ b/tests/test_project_backfill.py @@ -0,0 +1,213 @@ +import json +import os +import sqlite3 +import subprocess +import sys +from pathlib import Path + +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 + + +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.execute( + "INSERT INTO chunks VALUES ('already-null', ?, NULL)", + (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 == { + "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 == 2 + assert rollback_path.read_text(encoding="utf-8").splitlines() == [ + "id\tproject", + "recoverable\t30", + "underivable\t7", + ] + + second = backfill_numeric_projects( + db_path, + rollback_path=tmp_path / "second-rollback.tsv", + 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 + + +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, + stderr=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + 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: + 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"): + 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..0bc0c7e6 100644 --- a/tests/test_watcher_bridge.py +++ b/tests/test_watcher_bridge.py @@ -151,6 +151,41 @@ 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_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" + + 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 +205,111 @@ 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_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() + 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) + 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( + { + "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]) + 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"} + + 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"