diff --git a/AGENTS.md b/AGENTS.md index 90c3bc5..9e3dfb6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,7 +33,8 @@ Semantic search over agent session transcripts. Independent project, originally - **Provider backfill controls (SESF-6)** — multi-harness ingestion (`codex`, `opencode`, `antigravity_cli`, `antigravity_desktop`) shares one embedding budget. Tune via `SESSIONFLOW_EMBED_BATCH_SIZE`, `SESSIONFLOW_EMBED_COOLDOWN_MS` (floor 200ms — MLX Metal SIGSEGVs lower), `SESSIONFLOW_BACKFILL_MODE` (`recent`|`incremental`|`full`), `SESSIONFLOW_BACKFILL_MAX_TURNS_PER_RUN`, `SESSIONFLOW_BACKFILL_MAX_FILES_PER_RUN`, `SESSIONFLOW_BACKFILL_RECENT_DAYS`, `SESSIONFLOW_BACKFILL_PAUSED`. Pause/resume + per-provider control via `python cleanup.py backfill {status|pause|resume|enqueue} [--provider ]`. Queue state is durable across restarts. - **Claude Desktop is `claude_code_cli` (SESF-59)** — Desktop Code-tab sessions, former CoWork projects included, write normal transcripts to `~/.claude/projects/` and are indexed as `claude_code_cli`. The `claude-code-sessions/**/local_*.json` files are only metadata pointing at those transcripts (`cliSessionId`), so the old `claude_desktop_cowork` probe was retired. Legacy Cowork agent-mode transcripts under `local-agent-mode-sessions/` are a separate provider, `claude_cowork_legacy` (SESF-60). - **Legacy Cowork agent mode (SESF-60)** — `claude_cowork_legacy` ingests `local-agent-mode-sessions/*/*/local_*/.claude/projects/**/*.jsonl` with the inherited CLI parser. Subagent transcripts are included; `audit.jsonl` is skipped. The transcript `cwd` is a VM path (`/sessions/`), so `project_root` comes from the sibling `local_.json`. It uses `userSelectedFolders[0]`, else a non-sandbox `cwd`, else `unknown`. No other metadata key is read. This is a one-time source and is not in the startup backfill list. Enqueue it with `cleanup.py backfill enqueue --provider claude_cowork_legacy --mode full`. Desktop may prune or move the folder (Unified Claude, 2026-11-10). To re-ingest from a backup, set `SESSIONFLOW_COWORK_LEGACY_ROOT` to the copy. -- **Moved project folders (SESF-73)** — project filters match `project_root` exactly, so turns tagged with a folder's old path are missed. `SESSIONFLOW_PROJECT_PATH_MAP` rewrites the root at ingest. Its format is `old=>new` pairs separated by `;`. Both sides are absolute paths. The longest prefix wins, and a prefix only matches whole path components. Only `claude_cowork_legacy` reads it so far. launchd does not pass shell env, so set it in the LaunchAgent plist (or `launchctl setenv`) before a re-ingest. To fix rows already indexed, run `cleanup.py retag-project --from --to [--provider P]`. It is a dry run by default, and `--apply` refuses without `--yes`. It calls `rag_engine.update_row_metadata`, which rewrites Milvus and FTS but keeps the stored vector. It also reads FTS, so a re-run picks up any turn whose FTS rewrite failed. +- **Project scope (SESF-64)** — `project_scope.resolve_project_scope` turns the `project_root` argument (or the `X-Project-Root` header) into a list of indexed roots. A path covers its subfolders, and a path inside `.worktrees/` or `.claude/worktrees/` resolves to its repository. An unindexed path falls back to its nearest indexed parent, and a bare name matches root basenames case-insensitively. A miss searches every project. Search and stats output opens with a `Scope:` line that names the roots and the reason. Known roots come from `rag_engine.known_project_roots` (FTS `DISTINCT`, cached `KNOWN_ROOTS_TTL_S`). Codex worktrees under `~/.codex/worktrees/` sit outside the repo path, so only a name lookup reaches them. +- **Moved project folders (SESF-73)** — project filters cover a root and its subfolders, but turns tagged with a folder's old path are missed. `SESSIONFLOW_PROJECT_PATH_MAP` rewrites the root at ingest. Its format is `old=>new` pairs separated by `;`. Both sides are absolute paths. The longest prefix wins, and a prefix only matches whole path components. Only `claude_cowork_legacy` reads it so far. launchd does not pass shell env, so set it in the LaunchAgent plist (or `launchctl setenv`) before a re-ingest. To fix rows already indexed, run `cleanup.py retag-project --from --to [--provider P]`. It is a dry run by default, and `--apply` refuses without `--yes`. It calls `rag_engine.update_row_metadata`, which rewrites Milvus and FTS but keeps the stored vector. It also reads FTS, so a re-run picks up any turn whose FTS rewrite failed. - **Hosted embeddings deferred** — SESF-6 keeps embedding fully local (MLX). No hosted/OpenAI setup steps, credentials, or collections exist. Future hosted path would require a separate identity/collection to avoid vector mixing. - **Optional LaunchAgent for multi-harness startup** — `./sessionflow-server.sh install-agent` writes `~/Library/LaunchAgents/cc.lbruton.sessionflow.plist` so the server starts at login before any harness hook races. OPTIONAL — `start/stop/status/restart` behavior is unchanged when not installed. `agent-status` / `uninstall-agent` manage it. `setup.sh` honors `SESSIONFLOW_INSTALL_AGENT=1` for non-interactive installs. - **Optional hourly backfill LaunchAgent (SESF-7)** — `./sessionflow-server.sh install-backfill-agent` writes `cc.lbruton.sessionflow-backfill.plist` with a `StartInterval` (defaults to 3600s; override via `SESSIONFLOW_BACKFILL_INTERVAL_SECONDS`). It runs `python cleanup.py backfill enqueue --provider claude_code_cli --mode incremental`, decoupling catch-up indexing from MCP session activity. MCP-client SessionStart hooks should connect to the already-running server rather than racing to start it. diff --git a/CLAUDE.md b/CLAUDE.md index 467b6ad..f069ed0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,7 +33,8 @@ Semantic search over Claude Code session transcripts. Independent project, origi - **Provider backfill controls (SESF-6)** — multi-harness ingestion (`codex`, `opencode`, `antigravity_cli`, `antigravity_desktop`) shares one embedding budget. Tune via `SESSIONFLOW_EMBED_BATCH_SIZE`, `SESSIONFLOW_EMBED_COOLDOWN_MS` (floor 200ms — MLX Metal SIGSEGVs lower), `SESSIONFLOW_BACKFILL_MODE` (`recent`|`incremental`|`full`), `SESSIONFLOW_BACKFILL_MAX_TURNS_PER_RUN`, `SESSIONFLOW_BACKFILL_MAX_FILES_PER_RUN`, `SESSIONFLOW_BACKFILL_RECENT_DAYS`, `SESSIONFLOW_BACKFILL_PAUSED`. Pause/resume + per-provider control via `python cleanup.py backfill {status|pause|resume|enqueue} [--provider ]`. Queue state is durable across restarts. - **Claude Desktop is `claude_code_cli` (SESF-59)** — Desktop Code-tab sessions, former CoWork projects included, write normal transcripts to `~/.claude/projects/` and are indexed as `claude_code_cli`. The `claude-code-sessions/**/local_*.json` files are only metadata pointing at those transcripts (`cliSessionId`), so the old `claude_desktop_cowork` probe was retired. Legacy Cowork agent-mode transcripts under `local-agent-mode-sessions/` are a separate provider, `claude_cowork_legacy` (SESF-60). - **Legacy Cowork agent mode (SESF-60)** — `claude_cowork_legacy` ingests `local-agent-mode-sessions/*/*/local_*/.claude/projects/**/*.jsonl` with the inherited CLI parser. Subagent transcripts are included; `audit.jsonl` is skipped. The transcript `cwd` is a VM path (`/sessions/`), so `project_root` comes from the sibling `local_.json`. It uses `userSelectedFolders[0]`, else a non-sandbox `cwd`, else `unknown`. No other metadata key is read. This is a one-time source and is not in the startup backfill list. Enqueue it with `cleanup.py backfill enqueue --provider claude_cowork_legacy --mode full`. Desktop may prune or move the folder (Unified Claude, 2026-11-10). To re-ingest from a backup, set `SESSIONFLOW_COWORK_LEGACY_ROOT` to the copy. -- **Moved project folders (SESF-73)** — project filters match `project_root` exactly, so turns tagged with a folder's old path are missed. `SESSIONFLOW_PROJECT_PATH_MAP` rewrites the root at ingest. Its format is `old=>new` pairs separated by `;`. Both sides are absolute paths. The longest prefix wins, and a prefix only matches whole path components. Only `claude_cowork_legacy` reads it so far. launchd does not pass shell env, so set it in the LaunchAgent plist (or `launchctl setenv`) before a re-ingest. To fix rows already indexed, run `cleanup.py retag-project --from --to [--provider P]`. It is a dry run by default, and `--apply` refuses without `--yes`. It calls `rag_engine.update_row_metadata`, which rewrites Milvus and FTS but keeps the stored vector. It also reads FTS, so a re-run picks up any turn whose FTS rewrite failed. +- **Project scope (SESF-64)** — `project_scope.resolve_project_scope` turns the `project_root` argument (or the `X-Project-Root` header) into a list of indexed roots. A path covers its subfolders, and a path inside `.worktrees/` or `.claude/worktrees/` resolves to its repository. An unindexed path falls back to its nearest indexed parent, and a bare name matches root basenames case-insensitively. A miss searches every project. Search and stats output opens with a `Scope:` line that names the roots and the reason. Known roots come from `rag_engine.known_project_roots` (FTS `DISTINCT`, cached `KNOWN_ROOTS_TTL_S`). Codex worktrees under `~/.codex/worktrees/` sit outside the repo path, so only a name lookup reaches them. +- **Moved project folders (SESF-73)** — project filters cover a root and its subfolders, but turns tagged with a folder's old path are missed. `SESSIONFLOW_PROJECT_PATH_MAP` rewrites the root at ingest. Its format is `old=>new` pairs separated by `;`. Both sides are absolute paths. The longest prefix wins, and a prefix only matches whole path components. Only `claude_cowork_legacy` reads it so far. launchd does not pass shell env, so set it in the LaunchAgent plist (or `launchctl setenv`) before a re-ingest. To fix rows already indexed, run `cleanup.py retag-project --from --to [--provider P]`. It is a dry run by default, and `--apply` refuses without `--yes`. It calls `rag_engine.update_row_metadata`, which rewrites Milvus and FTS but keeps the stored vector. It also reads FTS, so a re-run picks up any turn whose FTS rewrite failed. - **Hosted embeddings deferred** — SESF-6 keeps embedding fully local (MLX). No hosted/OpenAI setup steps, credentials, or collections exist. Future hosted path would require a separate identity/collection to avoid vector mixing. - **Optional LaunchAgent for multi-harness startup** — `./sessionflow-server.sh install-agent` writes `~/Library/LaunchAgents/cc.lbruton.sessionflow.plist` so the server starts at login before any harness hook races. OPTIONAL — `start/stop/status/restart` behavior is unchanged when not installed. `agent-status` / `uninstall-agent` manage it. `setup.sh` honors `SESSIONFLOW_INSTALL_AGENT=1` for non-interactive installs. - **Optional hourly backfill LaunchAgent (SESF-7)** — `./sessionflow-server.sh install-backfill-agent` writes `cc.lbruton.sessionflow-backfill.plist` with a `StartInterval` (defaults to 3600s; override via `SESSIONFLOW_BACKFILL_INTERVAL_SECONDS`). It runs `python cleanup.py backfill enqueue --provider claude_code_cli --mode incremental`, decoupling catch-up indexing from MCP session activity. MCP-client SessionStart hooks should connect to the already-running server rather than racing to start it. diff --git a/fts_hybrid.py b/fts_hybrid.py index 3cbf293..d42c337 100644 --- a/fts_hybrid.py +++ b/fts_hybrid.py @@ -23,7 +23,7 @@ import sqlite3 import threading from pathlib import Path -from typing import Dict, List, Optional, Set +from typing import Dict, List, Optional, Sequence, Set, Union logger = logging.getLogger("fts-hybrid") @@ -241,28 +241,37 @@ def insert(self, conn: sqlite3.Connection, records: List[Dict]): ) conn.commit() - def count_rows(self, conn: sqlite3.Connection, project_root: Optional[str] = None) -> int: + def count_rows(self, conn: sqlite3.Connection, + project_root: Optional[Union[str, Sequence[str]]] = None) -> int: """Return the number of rows in the FTS table. - Runs `SELECT count(*) FROM {self.table_name}`, adding a - `WHERE project_root = ?` filter when `project_root` is supplied (the - turns_fts table stores project_root), so callers can read either the - global row count or a single project's count. - Args: conn: the per-thread sqlite connection for this FTS index. - project_root: when provided, count only rows whose project_root matches. + project_root: when provided, count only rows whose project_root + equals it, or any of its values when it is a list. Returns: int: row count, optionally filtered by project_root. """ sql = f"SELECT count(*) FROM {self.table_name}" params: tuple = () - if project_root is not None: + if isinstance(project_root, str): sql += " WHERE project_root = ?" params = (project_root,) + elif project_root: + sql += f" WHERE project_root IN ({', '.join('?' for _ in project_root)})" + params = tuple(project_root) return int(conn.execute(sql, params).fetchone()[0]) + def distinct_values(self, conn: sqlite3.Connection, column: str) -> List[str]: + """Return the distinct non-null values of a metadata column.""" + if column not in self.metadata_columns: + raise ValueError(f"Unknown FTS column: {column!r}") + rows = conn.execute( + f"SELECT DISTINCT {column} FROM {self.table_name} WHERE {column} IS NOT NULL" + ).fetchall() + return [r[0] for r in rows] + def delete(self, conn: sqlite3.Connection, column: str, value): """Delete rows where column == value.""" conn.execute(f"DELETE FROM {self.table_name} WHERE {column} = ?", (value,)) @@ -289,7 +298,8 @@ def search(self, query: str, n: int = 15, Args: query: Search text n: Max results - filters: Dict of column -> value for exact-match filtering + filters: Dict of column -> value for exact-match filtering. A value + may be ``(op, operand)`` for a comparison or ``("in", values)``. db_path: Milvus DB path (used to derive FTS path) Returns: @@ -313,7 +323,10 @@ def search(self, query: str, n: int = 15, for key, val in filters.items(): if val is None: continue - if isinstance(val, tuple) and len(val) == 2: + if isinstance(val, tuple) and len(val) == 2 and val[0] == "in": + where_parts.append(f"{key} IN ({', '.join('?' for _ in val[1])})") + params.extend(val[1]) + elif isinstance(val, tuple) and len(val) == 2: op, operand = val col = key.rsplit("_", 1)[0] if key.endswith(("_gte", "_lte", "_gt", "_lt")) else key where_parts.append(f"{col} {op} ?") diff --git a/project_scope.py b/project_scope.py new file mode 100644 index 0000000..b0aba2a --- /dev/null +++ b/project_scope.py @@ -0,0 +1,97 @@ +"""Resolve a search's project argument to the indexed ``project_root`` values it covers.""" + +from __future__ import annotations + +import posixpath +from dataclasses import dataclass +from typing import Iterable, List, Optional, Tuple + +# Worktree folders inside a repository: ours, and the Claude desktop app's. +_WORKTREE_MARKERS = ("/.worktrees/", "/.claude/worktrees/") +_LABEL_ROOTS = 3 + + +@dataclass(frozen=True) +class ProjectScope: + """The project roots a search covers. + + Attributes: + roots: Sorted ``project_root`` values to match, or None for every project. + note: Why the scope differs from the argument, or an empty string. + """ + + roots: Optional[Tuple[str, ...]] + note: str = "" + + def filter_value(self) -> Optional[List[str]]: + """Return the roots as a list for the engine filters, or None for every project.""" + return list(self.roots) if self.roots else None + + def label(self) -> str: + """Return the roots for display, truncated after three.""" + if not self.roots: + return "all projects" + shown = ", ".join(self.roots[:_LABEL_ROOTS]) + extra = len(self.roots) - _LABEL_ROOTS + return f"{shown} (+{extra} more)" if extra > 0 else shown + + +def _with_subfolders(base: str, known: List[str]) -> List[str]: + if base == "/": + return [r for r in known if r == "/"] + return [r for r in known if r == base or r.startswith(base + "/")] + + +def _nearest_parent(path: str, known: List[str]) -> Optional[str]: + parents = [r for r in known if r != "/" and path.startswith(r + "/")] + return max(parents, key=len) if parents else None + + +def _basename(root: str) -> str: + return root.rstrip("/").rsplit("/", 1)[-1] + + +def resolve_project_scope(arg: Optional[str], header_root: Optional[str], + known_roots: Iterable[str]) -> ProjectScope: + """Resolve a project argument against the indexed roots. + + ``'*'`` covers every project. An empty argument falls back to ``header_root``. + A path covers its indexed root and subfolders; a path inside a worktree folder + resolves to its repository first, and an unindexed path to its nearest indexed + parent. A bare name matches root basenames case-insensitively. A miss covers + every project and says why in ``note``. + + Args: + arg: The caller's ``project_root`` argument. + header_root: The current project from the request header. + known_roots: Every ``project_root`` value in the index. + + Returns: + The resolved scope. + """ + if arg == "*": + return ProjectScope(roots=None) + target = (arg or "").strip() or (header_root or "").strip() + if not target: + return ProjectScope(roots=None, note="no current project") + known = sorted(set(known_roots)) + + if target.startswith("/"): + # normpath keeps exactly two leading slashes, so collapse them. + path = "/" + posixpath.normpath(target).lstrip("/") + for marker in _WORKTREE_MARKERS: + path = path.split(marker, 1)[0] + matched = _with_subfolders(path, known) + if matched: + return ProjectScope(roots=tuple(matched)) + parent = _nearest_parent(path, known) + if parent: + return ProjectScope(roots=(parent,), note=f"nearest indexed parent of {target}") + return ProjectScope(roots=None, note=f"no indexed turns under {target}") + + name = target.lower() + matched = sorted({sub for r in known if _basename(r).lower() == name + for sub in _with_subfolders(r, known)}) + if matched: + return ProjectScope(roots=tuple(matched)) + return ProjectScope(roots=None, note=f'no project named "{target}"') diff --git a/rag_engine.py b/rag_engine.py index 0935f93..ac91351 100644 --- a/rag_engine.py +++ b/rag_engine.py @@ -31,7 +31,7 @@ from pymilvus.exceptions import MilvusException from contextlib import contextmanager from concurrent.futures import ThreadPoolExecutor -from typing import Iterator, List, Dict, NamedTuple, Optional +from typing import Iterator, List, Dict, NamedTuple, Optional, Sequence, Union import asyncio import logging import sys @@ -439,6 +439,11 @@ def _embed_texts_local(texts: List[str], is_query: bool = False) -> List[List[fl "project_root", "logical_session_id", "provider", "source_kind", "source_class", "source_id", "source_path", "issue_ids", ]) +# A project filter: one root, or several (a root with its subfolders, or a name match). +ProjectRoots = Union[str, Sequence[str]] +# db_path -> (monotonic time, roots); see known_project_roots. +_known_roots_cache: Dict[Optional[str], tuple] = {} +KNOWN_ROOTS_TTL_S = 300.0 _write_lock: Optional[asyncio.Lock] = None _embed_semaphore: Optional[asyncio.Semaphore] = None # Dedicated single-worker executor so every MLX/Metal call runs on the same OS @@ -1099,8 +1104,26 @@ def _row_to_result(entity: Dict, defaults: Dict, distance: Optional[float] = Non } +def _project_root_filter(project_root: Optional[ProjectRoots]) -> Optional[str]: + """Milvus clause for one root (``==``) or several (``in``); None when unscoped.""" + if not project_root: + return None + if isinstance(project_root, str): + project_root = [project_root] + if len(project_root) == 1: + return f'project_root == "{_escape_filter_scalar(project_root[0])}"' + values = ", ".join(f'"{_escape_filter_scalar(r)}"' for r in project_root) + return f"project_root in [{values}]" + + +def _fts_project_filter(project_root: ProjectRoots): + if isinstance(project_root, str): + return project_root + return ("in", list(project_root)) + + def _build_milvus_filter(session_id: Optional[str], git_branch: Optional[str], - project_root: Optional[str], provider: Optional[str], + project_root: Optional[ProjectRoots], provider: Optional[str], source_kind: Optional[str], date_from: Optional[str], date_to: Optional[str], issue_id: Optional[str] = None) -> Optional[str]: @@ -1116,7 +1139,7 @@ def _build_milvus_filter(session_id: Optional[str], git_branch: Optional[str], if git_branch: filters.append(f'git_branch == "{_escape_filter_scalar(git_branch)}"') if project_root: - filters.append(f'project_root == "{_escape_filter_scalar(project_root)}"') + filters.append(_project_root_filter(project_root)) if provider: filters.append(f'provider == "{provider}"') if source_kind: @@ -1147,7 +1170,7 @@ def _is_identifier_query(query: Optional[str]) -> bool: def search(query: Optional[str], n: int = 5, session_id: Optional[str] = None, - git_branch: Optional[str] = None, project_root: Optional[str] = None, + git_branch: Optional[str] = None, project_root: Optional[ProjectRoots] = None, sort_by: str = "hybrid", date_from: Optional[str] = None, date_to: Optional[str] = None, provider: Optional[str] = None, source_kind: Optional[str] = None, @@ -1160,8 +1183,8 @@ def search(query: Optional[str], n: int = 5, session_id: Optional[str] = None, ``relevance`` (pure RRF order), ``recency`` (timestamp-descending re-rank), or ``hybrid`` (blended semantic + recency score; the default). - project_root: when set, restricts results to that project. When None, - searches across all projects (cross-project search). + project_root: when set, restricts results to that root, or to any of a list + of roots. When None, searches across all projects (cross-project search). date_from/date_to: ISO 8601 date strings (e.g. '2026-04-02') to restrict results to a time range. Timestamps are VARCHAR and sort lexicographically. issue_id: when set, restricts results to turns tagged with that issue id @@ -1246,7 +1269,7 @@ def search(query: Optional[str], n: int = 5, session_id: Optional[str] = None, if git_branch: fts_filters["git_branch"] = git_branch if project_root: - fts_filters["project_root"] = project_root + fts_filters["project_root"] = _fts_project_filter(project_root) if provider: fts_filters["provider"] = provider if source_kind: @@ -1304,7 +1327,7 @@ def search(query: Optional[str], n: int = 5, session_id: Optional[str] = None, def _recent_listing(n: int, session_id: Optional[str] = None, git_branch: Optional[str] = None, - project_root: Optional[str] = None, + project_root: Optional[ProjectRoots] = None, provider: Optional[str] = None, source_kind: Optional[str] = None, date_from: Optional[str] = None, @@ -2014,8 +2037,9 @@ def get_turns(session_id: str, turn_index: int, context: int = 2, return formatted -def get_stats(project_root: Optional[str] = None, db_path: Optional[str] = None) -> Dict: - """Get index statistics. Optionally filter to a specific project.""" +def get_stats(project_root: Optional[ProjectRoots] = None, + db_path: Optional[str] = None) -> Dict: + """Get index statistics. Optionally filter to one project root or a list of roots.""" with milvus_client(db_path) as client: if not client.has_collection(COLLECTION_NAME): # Keep the redaction surface on every return path so @@ -2031,7 +2055,7 @@ def get_stats(project_root: Optional[str] = None, db_path: Optional[str] = None) # Query for breakdowns (capped by Milvus offset limit) all_results = _query_all( ["session_id", "chunk_type", "git_branch", "project_root", "provider"], - filter_expr=f'project_root == "{_escape_filter_scalar(project_root)}"' if project_root else None, + filter_expr=_project_root_filter(project_root), db_path=db_path, ) @@ -2087,11 +2111,7 @@ def _count_milvus_turns(db_path=None, project_root=None) -> int: Returns: int: the number of turn rows (project-scoped when ``project_root`` is set). """ - filter_expr = ( - f'project_root == "{_escape_filter_scalar(project_root)}"' - if project_root - else None - ) + filter_expr = _project_root_filter(project_root) with milvus_client_for_migration(db_path) as client: if not client.has_collection(COLLECTION_NAME): return 0 @@ -2105,7 +2125,7 @@ def _count_milvus_turns(db_path=None, project_root=None) -> int: def fts_lag_status( db_path: Optional[str] = None, - project_root: Optional[str] = None, + project_root: Optional[ProjectRoots] = None, milvus_turn_count: Optional[int] = None, ) -> Dict[str, object]: """Return static FTS-vs-Milvus lag data for observability. @@ -2152,6 +2172,52 @@ def fts_lag_status( } +def _fts_roots_complete(conn, db_path: Optional[str]) -> bool: + """Return True when the FTS sidecar holds every indexed turn. + + A set backfill sentinel or an FTS row count below the Milvus turn count + means FTS is rebuilding or missed inserts. A failed Milvus count keeps the + FTS answer. + """ + if fts_backfill_required(): + return False + try: + return _fts.count_rows(conn) >= _count_milvus_turns(db_path=db_path) + except Exception as e: + logger.warning("Milvus turn count failed, keeping FTS roots: %s", _scrub_exception(e)) + return True + + +def known_project_roots(db_path: Optional[str] = None) -> List[str]: + """Return every distinct ``project_root`` in the index. + + Reads the FTS sidecar, which is local, and reads Milvus instead when FTS + fails or is incomplete. The result is cached per ``db_path`` for + ``KNOWN_ROOTS_TTL_S``. + """ + cached = _known_roots_cache.get(db_path) + if cached and time.monotonic() - cached[0] < KNOWN_ROOTS_TTL_S: + return cached[1] + roots = None + try: + conn = _fts.connection(db_path) + try: + if _fts_roots_complete(conn, db_path): + roots = _fts.distinct_values(conn, "project_root") + finally: + _fts.close_ephemeral(conn) + except Exception as e: + logger.warning("FTS project_root scan failed, reading Milvus: %s", _scrub_exception(e)) + if roots is None: + found = set() + for batch in _query_batches(["project_root"], db_path=db_path): + found.update(row.get("project_root") or "" for row in batch) + roots = list(found) + roots = sorted(r for r in roots if r) + _known_roots_cache[db_path] = (time.monotonic(), roots) + return roots + + def _query_batches(output_fields: list, batch_size: int = 1000, filter_expr: Optional[str] = None, db_path: Optional[str] = None) -> Iterator[list]: @@ -2941,7 +3007,8 @@ def list_sessions(project_root: Optional[str] = None, # --- Async wrappers --- async def search_async(query: str, n: int = 5, session_id: Optional[str] = None, - git_branch: Optional[str] = None, project_root: Optional[str] = None, + git_branch: Optional[str] = None, + project_root: Optional[ProjectRoots] = None, sort_by: str = "hybrid", date_from: Optional[str] = None, date_to: Optional[str] = None, provider: Optional[str] = None, source_kind: Optional[str] = None, diff --git a/tests/test_known_roots.py b/tests/test_known_roots.py new file mode 100644 index 0000000..408d611 --- /dev/null +++ b/tests/test_known_roots.py @@ -0,0 +1,86 @@ +"""known_project_roots reads Milvus whenever the FTS sidecar may be incomplete.""" + +from __future__ import annotations + +import importlib + +import pytest + +rag_engine = importlib.import_module("rag_engine") + +FTS_ROOTS = ["/a", "/a/b"] +MILVUS_ROOTS = ["/a", "/a/b", "/c"] + + +class _FakeFts: + def __init__(self, roots=None, rows=2, raises=False): + self.roots, self.rows, self.raises = roots or FTS_ROOTS, rows, raises + + def connection(self, db_path=None): + return object() + + def close_ephemeral(self, conn): + pass + + def distinct_values(self, conn, column): + if self.raises: + raise RuntimeError("fts down") + return list(self.roots) + + def count_rows(self, conn, project_root=None): + return self.rows + + +@pytest.fixture +def setup(monkeypatch): + calls = {"scan": 0} + + def _scan(fields, db_path=None, **kw): + calls["scan"] += 1 + yield [{"project_root": r} for r in MILVUS_ROOTS] + + def _install(fts, sentinel=False, milvus_count=2): + monkeypatch.setattr(rag_engine, "_fts", fts) + monkeypatch.setattr(rag_engine, "fts_backfill_required", lambda: sentinel) + monkeypatch.setattr(rag_engine, "_count_milvus_turns", lambda **kw: milvus_count) + monkeypatch.setattr(rag_engine, "_query_batches", _scan) + return calls + + rag_engine._known_roots_cache.clear() + yield _install + rag_engine._known_roots_cache.clear() + + +def test_sentinel_set_reads_milvus(setup): + calls = setup(_FakeFts(), sentinel=True) + assert rag_engine.known_project_roots("db") == MILVUS_ROOTS + assert calls["scan"] == 1 + + +def test_fts_row_count_below_milvus_reads_milvus(setup): + calls = setup(_FakeFts(rows=2), milvus_count=5) + assert rag_engine.known_project_roots("db") == MILVUS_ROOTS + assert calls["scan"] == 1 + + +def test_equal_counts_and_no_sentinel_use_fts_only(setup): + calls = setup(_FakeFts(rows=2), milvus_count=2) + assert rag_engine.known_project_roots("db") == FTS_ROOTS + assert calls["scan"] == 0 + + +def test_fts_read_failure_reads_milvus(setup): + calls = setup(_FakeFts(raises=True)) + assert rag_engine.known_project_roots("db") == MILVUS_ROOTS + assert calls["scan"] == 1 + + +def test_milvus_count_failure_keeps_fts_list(setup, monkeypatch): + calls = setup(_FakeFts()) + + def _boom(**kw): + raise RuntimeError("milvus down") + + monkeypatch.setattr(rag_engine, "_count_milvus_turns", _boom) + assert rag_engine.known_project_roots("db") == FTS_ROOTS + assert calls["scan"] == 0 diff --git a/tests/test_project_scope.py b/tests/test_project_scope.py new file mode 100644 index 0000000..349aa48 --- /dev/null +++ b/tests/test_project_scope.py @@ -0,0 +1,330 @@ +"""SESF-64 red-phase tests: project scope resolution. + +Project filters matched ``project_root`` exactly, so a short name such as +``CoWork``, a subfolder root or a worktree path matched nothing. The scope +argument now resolves against the indexed roots: a path keeps its subfolders, +a worktree resolves to its repository, an unknown path falls back to the +nearest indexed parent, and a bare name matches root basenames. A miss widens +to every project, and every result set opens with a ``Scope:`` line. +""" + +from __future__ import annotations + +import asyncio +import importlib +from contextlib import contextmanager + +import pytest + +from fts_hybrid import FTSIndex +from project_scope import ProjectScope, resolve_project_scope + +rag_engine = importlib.import_module("rag_engine") +tools = importlib.import_module("tools") + +KNOWN = [ + "/", + "/Users/lbruton/Claude/Projects/CoWork", + "/Users/lbruton/Claude/Projects/CoWork/DocVault-CoWork", + "/Users/lbruton/Claude/Projects/Finances", + "/Volumes/DATA/GitHub/SessionFlow", + "/Volumes/DATA/GitHub/SessionFlow/.worktrees/fix-sesf-64", + "/Volumes/DATA/GitHub/StakTrakr", + "/Volumes/DATA/GitHub/Archive/StakTrakr", + "unknown", +] + + +def _resolve(arg, header=None): + return resolve_project_scope(arg, header, KNOWN) + + +# --------------------------------------------------------------------------- +# resolve_project_scope +# --------------------------------------------------------------------------- + +def test_star_is_all_projects(): + scope = _resolve("*", header="/Volumes/DATA/GitHub/SessionFlow") + assert scope.roots is None + assert scope.label() == "all projects" + + +def test_missing_arg_uses_header_root(): + scope = _resolve(None, header="/Users/lbruton/Claude/Projects/Finances") + assert scope.roots == ("/Users/lbruton/Claude/Projects/Finances",) + + +def test_no_arg_and_no_header_is_all_projects_with_note(): + scope = _resolve(None, header=None) + assert scope.roots is None + assert scope.note + + +def test_known_path_includes_subfolders(): + scope = _resolve("/Users/lbruton/Claude/Projects/CoWork") + assert scope.roots == ( + "/Users/lbruton/Claude/Projects/CoWork", + "/Users/lbruton/Claude/Projects/CoWork/DocVault-CoWork", + ) + + +def test_trailing_slash_is_ignored(): + scope = _resolve("/Users/lbruton/Claude/Projects/Finances/") + assert scope.roots == ("/Users/lbruton/Claude/Projects/Finances",) + + +def test_parent_segment_is_collapsed(): + scope = _resolve("/Users/lbruton/Claude/Projects/CoWork/../Finances") + assert scope.roots == ("/Users/lbruton/Claude/Projects/Finances",) + + +def test_dot_and_doubled_slashes_resolve_like_the_clean_path(): + clean = _resolve("/Users/lbruton/Claude/Projects/Finances") + assert _resolve("/Users/lbruton/Claude/./Projects//Finances/.").roots == clean.roots + + +def test_leading_double_slash_resolves_like_a_single_slash(): + scope = _resolve("//Users/lbruton/Claude/Projects/Finances") + assert scope.roots == ("/Users/lbruton/Claude/Projects/Finances",) + + +def test_parent_segment_out_of_a_worktree_keeps_no_marker(): + scope = _resolve("/Volumes/DATA/GitHub/SessionFlow/.worktrees/x/../../..") + assert "/Volumes/DATA/GitHub/StakTrakr" in scope.roots + assert "/" not in scope.roots + + +def test_worktree_path_resolves_to_repository(): + scope = _resolve(None, header="/Volumes/DATA/GitHub/SessionFlow/.worktrees/fix-sesf-64") + assert scope.roots == ( + "/Volumes/DATA/GitHub/SessionFlow", + "/Volumes/DATA/GitHub/SessionFlow/.worktrees/fix-sesf-64", + ) + + +def test_desktop_worktree_path_resolves_to_repository(): + scope = _resolve(None, header="/Volumes/DATA/GitHub/SessionFlow/.claude/worktrees/nice-curie") + assert scope.roots[0] == "/Volumes/DATA/GitHub/SessionFlow" + + +def test_unknown_path_with_indexed_subfolders_matches_them(): + scope = _resolve("/Users/lbruton/Claude/Projects") + assert scope.roots == ( + "/Users/lbruton/Claude/Projects/CoWork", + "/Users/lbruton/Claude/Projects/CoWork/DocVault-CoWork", + "/Users/lbruton/Claude/Projects/Finances", + ) + + +def test_unknown_path_falls_back_to_nearest_indexed_parent(): + scope = _resolve("/Users/lbruton/Claude/Projects/CoWork/notes/2026") + assert scope.roots == ("/Users/lbruton/Claude/Projects/CoWork",) + assert "parent" in scope.note + + +def test_slash_root_is_not_a_parent_of_everything(): + scope = _resolve("/Volumes/Other/Project") + assert scope.roots is None + assert "/Volumes/Other/Project" in scope.note + + +def test_slash_root_matches_only_itself(): + assert _resolve("/").roots == ("/",) + + +def test_bare_name_matches_basename_case_insensitively(): + scope = _resolve("cowork") + assert scope.roots == ( + "/Users/lbruton/Claude/Projects/CoWork", + "/Users/lbruton/Claude/Projects/CoWork/DocVault-CoWork", + ) + + +def test_ambiguous_name_matches_every_root_with_that_basename(): + scope = _resolve("StakTrakr") + assert scope.roots == ( + "/Volumes/DATA/GitHub/Archive/StakTrakr", + "/Volumes/DATA/GitHub/StakTrakr", + ) + + +def test_unknown_name_falls_back_to_all_projects_with_note(): + scope = _resolve("NoSuchProject") + assert scope.roots is None + assert "NoSuchProject" in scope.note + + +def test_name_unknown_matches_the_unknown_bucket(): + assert _resolve("unknown").roots == ("unknown",) + + +def test_label_lists_up_to_three_roots(): + scope = ProjectScope(roots=("/a", "/a/b")) + assert scope.label() == "/a, /a/b" + + +def test_label_truncates_long_root_lists(): + scope = ProjectScope(roots=("/a", "/a/b", "/a/c", "/a/d", "/a/e")) + assert scope.label() == "/a, /a/b, /a/c (+2 more)" + + +# --------------------------------------------------------------------------- +# Milvus filter +# --------------------------------------------------------------------------- + +def test_milvus_filter_single_root_is_equality(): + expr = rag_engine._build_milvus_filter(None, None, ["/a"], None, None, None, None) + assert expr == 'project_root == "/a"' + + +def test_milvus_filter_several_roots_uses_in_and_escapes(): + expr = rag_engine._build_milvus_filter( + None, None, ["/a", '/b"q'], None, None, None, None) + assert expr == 'project_root in ["/a", "/b\\"q"]' + + +def test_milvus_filter_string_root_is_unchanged(): + expr = rag_engine._build_milvus_filter(None, None, "/a", None, None, None, None) + assert expr == 'project_root == "/a"' + + +# --------------------------------------------------------------------------- +# FTS filter and distinct roots +# --------------------------------------------------------------------------- + +def _fts_with_rows(tmp_path): + fts = FTSIndex("turns_fts", ["session_id", "project_root"]) + db_path = str(tmp_path / "milvus.db") + conn = fts.connection(db_path) + try: + fts.insert(conn, [ + {"doc_id": "d1", "content": "deploy notes", "project_root": "/a"}, + {"doc_id": "d2", "content": "deploy notes", "project_root": "/a/b"}, + {"doc_id": "d3", "content": "deploy notes", "project_root": '/c"q'}, + ]) + finally: + fts.close_ephemeral(conn) + return fts, db_path + + +def test_fts_in_filter_matches_any_listed_root(tmp_path): + fts, db_path = _fts_with_rows(tmp_path) + rows = fts.search("deploy", n=10, filters={"project_root": ("in", ["/a/b", '/c"q'])}, + db_path=db_path) + assert sorted(r["doc_id"] for r in rows) == ["d2", "d3"] + + +def test_fts_count_rows_accepts_several_roots(tmp_path): + fts, db_path = _fts_with_rows(tmp_path) + conn = fts.connection(db_path) + try: + assert fts.count_rows(conn, project_root=["/a", "/a/b"]) == 2 + assert fts.count_rows(conn, project_root="/a") == 1 + finally: + fts.close_ephemeral(conn) + + +def test_fts_distinct_values(tmp_path): + fts, db_path = _fts_with_rows(tmp_path) + conn = fts.connection(db_path) + try: + assert sorted(fts.distinct_values(conn, "project_root")) == ["/a", "/a/b", '/c"q'] + finally: + fts.close_ephemeral(conn) + + +def test_known_project_roots_reads_fts_and_caches(tmp_path, monkeypatch): + fts, db_path = _fts_with_rows(tmp_path) + monkeypatch.setattr(rag_engine, "_fts", fts) + rag_engine._known_roots_cache.clear() + assert sorted(rag_engine.known_project_roots(db_path)) == ["/a", "/a/b", '/c"q'] + + def _boom(*a, **kw): + raise AssertionError("cache miss") + + monkeypatch.setattr(fts, "distinct_values", _boom) + assert sorted(rag_engine.known_project_roots(db_path)) == ["/a", "/a/b", '/c"q'] + rag_engine._known_roots_cache.clear() + + +# --------------------------------------------------------------------------- +# search sends the root list to both engines +# --------------------------------------------------------------------------- + +def _patch_search(monkeypatch) -> dict: + seen: dict = {} + + class _Client: + def search(self, *a, **kw): + seen["milvus_filter"] = kw.get("filter") + return [[]] + + @contextmanager + def _fake_client(db_path=None): + yield _Client() + + def _fts_search(query, n=15, filters=None, db_path=None): + seen["fts_filters"] = filters + return [] + + monkeypatch.setattr(rag_engine, "embed_texts", lambda *a, **kw: [[0.0] * 768]) + monkeypatch.setattr(rag_engine, "milvus_client", _fake_client) + monkeypatch.setattr(rag_engine._fts, "search", _fts_search) + return seen + + +def test_search_passes_root_list_to_both_engines(monkeypatch): + seen = _patch_search(monkeypatch) + rag_engine.search("deploy notes", n=5, project_root=["/a", "/a/b"]) + assert seen["milvus_filter"] == 'project_root in ["/a", "/a/b"]' + assert seen["fts_filters"] == {"project_root": ("in", ["/a", "/a/b"])} + + +def test_search_single_string_root_keeps_equality(monkeypatch): + seen = _patch_search(monkeypatch) + rag_engine.search("deploy notes", n=5, project_root="/a") + assert seen["fts_filters"] == {"project_root": "/a"} + + +# --------------------------------------------------------------------------- +# tools: scope resolution and the Scope line +# --------------------------------------------------------------------------- + +def test_resolve_scope_uses_known_roots(monkeypatch): + monkeypatch.setattr(rag_engine, "known_project_roots", lambda db_path=None: KNOWN) + scope = asyncio.run(tools.resolve_scope("cowork", "/Volumes/DATA/GitHub/SessionFlow", "db")) + assert scope.roots[0] == "/Users/lbruton/Claude/Projects/CoWork" + + +def test_resolve_scope_star_skips_the_root_lookup(monkeypatch): + def _boom(db_path=None): + raise AssertionError("lookup not needed") + + monkeypatch.setattr(rag_engine, "known_project_roots", _boom) + scope = asyncio.run(tools.resolve_scope("*", "/x", "db")) + assert scope.roots is None + + +def test_format_results_opens_with_scope_line(): + row = {"content": "body", "session_id": "s", "turn_index": 1, "chunk_type": "turn"} + scope = ProjectScope(roots=None, note='no project named "Nope"') + rendered = tools.format_results([row], scope=scope) + assert rendered.startswith('Scope: all projects (no project named "Nope")\n') + + +def test_format_results_empty_still_shows_scope(): + rendered = tools.format_results([], scope=ProjectScope(roots=("/a",))) + assert rendered == "Scope: /a\n\nNo results found." + + +def test_format_results_without_scope_is_unchanged(): + assert tools.format_results([]) == "No results found." + + +@pytest.mark.parametrize("roots, expected", [ + (None, None), + (("/a",), ["/a"]), + (("/a", "/a/b"), ["/a", "/a/b"]), +]) +def test_scope_filter_value(roots, expected): + assert ProjectScope(roots=roots).filter_value() == expected diff --git a/tools.py b/tools.py index 7caadec..ff32851 100644 --- a/tools.py +++ b/tools.py @@ -12,6 +12,7 @@ import rag_engine import sanitize +from project_scope import ProjectScope, resolve_project_scope from provider_adapters import ( LEGAL_PROVIDERS, LEGAL_SORT_BY, @@ -42,6 +43,22 @@ def get_db_path() -> str: return os.getenv("SESSIONFLOW_MILVUS_URI", str(Path.home() / ".sessionflow" / "milvus.db")) +async def resolve_scope(arg: str | None, header_root: str | None, db_path: str) -> ProjectScope: + """Resolve a tool's project argument against the indexed roots. + + The root lookup reads SQLite, so it runs off the event loop. ``'*'`` needs no lookup. + """ + if arg == "*": + return ProjectScope(roots=None) + known = await asyncio.to_thread(rag_engine.known_project_roots, db_path) + return resolve_project_scope(arg, header_root, known) + + +def _scope_line(scope: ProjectScope) -> str: + line = f"Scope: {scope.label()}" + return f"{line} ({scope.note})" if scope.note else line + + def _validate_enum_arg(name: str, value, legal_values) -> "types.TextContent | None": """Return an error TextContent if `value` is non-None and outside `legal_values`. @@ -76,12 +93,12 @@ def _format_signals(r: dict) -> str: return " · ".join(parts) if parts else "Listing" -def format_results(results: list[dict]) -> str: - """Format search results as markdown.""" +def format_results(results: list[dict], scope: ProjectScope | None = None) -> str: + """Format search results as markdown, opening with the scope line when ``scope`` is set.""" + output = [_scope_line(scope), ""] if scope is not None else [] if not results: - return "No results found." + return "\n".join(output + ["No results found."]) - output = [] for i, r in enumerate(results, 1): # Header with metadata session_id = r.get("session_id", "") @@ -152,9 +169,10 @@ def format_turns(results: list[dict]) -> str: return "\n".join(output) -def format_stats(stats: dict, db_path: str) -> str: - """Format index statistics.""" - lines = [ +def format_stats(stats: dict, db_path: str, scope: ProjectScope | None = None) -> str: + """Format index statistics, opening with the scope line when ``scope`` is set.""" + lines = [_scope_line(scope), ""] if scope is not None else [] + lines += [ f"**Total Turns:** {stats['total_turns']}", f"**Sessions:** {stats['sessions']}", ] @@ -283,7 +301,7 @@ def build_search_all_sessions_schema() -> dict: }, "project_root": { "type": "string", - "description": "Project path to search, or '*' for every project. Default: the current project.", + "description": "Project to search: a path (its subfolders and worktrees are included), a folder name such as 'CoWork', or '*' for every project. Default: the current project. A name or path with no indexed turns searches every project, and the Scope line says so.", }, "provider": { "type": "string", @@ -789,29 +807,21 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: return [types.TextContent( type="text", text="issue_id must be a valid issue token like SESF-25")] + scope = await resolve_scope(None, current_project, db) # search_async runs on the mlx-embed thread; a blocking search() # here would run MLX on the event loop and race backfill. results = await rag_engine.search_async( arguments.get("query") or "", arguments.get("n", 5), session_id=session_id, - project_root=current_project, + project_root=scope.filter_value(), sort_by=sort_by_arg, issue_id=issue_id_arg, db_path=db, ) - return [types.TextContent(type="text", text=format_results(results))] + return [types.TextContent(type="text", text=format_results(results, scope))] elif name == "search_all_sessions": - # project_root scoping: default=current project, "*"=all projects - pr_arg = arguments.get("project_root") - if pr_arg == "*": - pr = None # cross-project search - elif pr_arg: - pr = pr_arg # explicit project - else: - pr = current_project # default: current project - provider_arg = arguments.get("provider") source_kind_arg = arguments.get("source_kind") sort_by_arg = arguments.get("sort_by", "hybrid") @@ -829,11 +839,12 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: type="text", text="issue_id must be a valid issue token like SESF-25")] + scope = await resolve_scope(arguments.get("project_root"), current_project, db) results = await rag_engine.search_async( arguments.get("query") or "", arguments.get("n", 10), git_branch=arguments.get("git_branch"), - project_root=pr, + project_root=scope.filter_value(), sort_by=sort_by_arg, date_from=arguments.get("date_from"), date_to=arguments.get("date_to"), @@ -843,8 +854,12 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: db_path=db, ) if arguments.get("format") == "json": - return [types.TextContent(type="text", text=_json_text({"results": results}))] - return [types.TextContent(type="text", text=format_results(results))] + payload = { + "scope": {"roots": scope.filter_value(), "note": scope.note}, + "results": results, + } + return [types.TextContent(type="text", text=_json_text(payload))] + return [types.TextContent(type="text", text=format_results(results, scope))] elif name == "get_turns": results = rag_engine.get_turns( @@ -856,11 +871,12 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: return [types.TextContent(type="text", text=format_turns(results))] elif name == "get_session_stats": + scope = await resolve_scope(None, current_project, db) stats = rag_engine.get_stats( - project_root=current_project, + project_root=scope.filter_value(), db_path=db, ) - return [types.TextContent(type="text", text=format_stats(stats, db))] + return [types.TextContent(type="text", text=format_stats(stats, db, scope))] elif name == "get_issue_timeline": if not is_valid_issue_token(arguments.get("issue_id")):