fix(watcher): derive projects from workspace metadata

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

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

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