diff --git a/src/cli/commands/execute.py b/src/cli/commands/execute.py index 771649f..a51cc97 100644 --- a/src/cli/commands/execute.py +++ b/src/cli/commands/execute.py @@ -48,14 +48,22 @@ def _maybe_refresh_index(cwd: Path, cfg) -> None: if building_marker.exists(): logger.info("index.skip_async_build_in_progress") return + workers = getattr(cfg, "index_build_workers", 0) + batch_size = getattr(cfg, "index_build_batch_size", 1000) try: from state.file_index import IndexBuilder, _last_indexed_sha if not db_path.exists(): - IndexBuilder.build_full(cwd, db_path) + IndexBuilder.build_full( + cwd, db_path, workers=workers, batch_size=batch_size + ) else: IndexBuilder.build_incremental( - cwd, db_path, since_sha=_last_indexed_sha(db_path) + cwd, + db_path, + since_sha=_last_indexed_sha(db_path), + workers=workers, + batch_size=batch_size, ) except Exception as exc: # noqa: BLE001 - never block on index failure logger.warning("index.refresh_failed", err=str(exc)) diff --git a/src/cli/commands/init.py b/src/cli/commands/init.py index 17e3131..a0070e2 100644 --- a/src/cli/commands/init.py +++ b/src/cli/commands/init.py @@ -203,6 +203,10 @@ def init(platform: str, force: bool, inline: bool, rebuild_index: bool) -> None: str(cwd), "--db", str(db_path), + "--workers", + str(cfg.index_build_workers), + "--batch-size", + str(cfg.index_build_batch_size), ] subprocess.Popen( # noqa: S603 - executable is sys.executable cmd, @@ -221,7 +225,12 @@ def init(platform: str, force: bool, inline: bool, rebuild_index: bool) -> None: ) else: with console.status("Building file/symbol index..."): - stats = IndexBuilder.build_full(cwd, db_path) + stats = IndexBuilder.build_full( + cwd, + db_path, + workers=cfg.index_build_workers, + batch_size=cfg.index_build_batch_size, + ) index_summary = ( f"{stats.file_count} files, {stats.symbol_count} symbols " f"({stats.duration_ms} ms)" diff --git a/src/cli/commands/plan.py b/src/cli/commands/plan.py index 55f3986..b333da5 100644 --- a/src/cli/commands/plan.py +++ b/src/cli/commands/plan.py @@ -51,14 +51,22 @@ def _maybe_refresh_index(cwd: Path, cfg) -> None: if building_marker.exists(): logger.info("index.skip_async_build_in_progress") return + workers = getattr(cfg, "index_build_workers", 0) + batch_size = getattr(cfg, "index_build_batch_size", 1000) try: from state.file_index import IndexBuilder, _last_indexed_sha if not db_path.exists(): - IndexBuilder.build_full(cwd, db_path) + IndexBuilder.build_full( + cwd, db_path, workers=workers, batch_size=batch_size + ) else: IndexBuilder.build_incremental( - cwd, db_path, since_sha=_last_indexed_sha(db_path) + cwd, + db_path, + since_sha=_last_indexed_sha(db_path), + workers=workers, + batch_size=batch_size, ) except Exception as exc: # noqa: BLE001 - never block on index failure logger.warning("index.refresh_failed", err=str(exc)) diff --git a/src/cli/commands/resume.py b/src/cli/commands/resume.py index d6e8198..a99589d 100644 --- a/src/cli/commands/resume.py +++ b/src/cli/commands/resume.py @@ -35,14 +35,22 @@ def _maybe_refresh_index(cwd: Path, cfg) -> None: if building_marker.exists(): logger.info("index.skip_async_build_in_progress") return + workers = getattr(cfg, "index_build_workers", 0) + batch_size = getattr(cfg, "index_build_batch_size", 1000) try: from state.file_index import IndexBuilder, _last_indexed_sha if not db_path.exists(): - IndexBuilder.build_full(cwd, db_path) + IndexBuilder.build_full( + cwd, db_path, workers=workers, batch_size=batch_size + ) else: IndexBuilder.build_incremental( - cwd, db_path, since_sha=_last_indexed_sha(db_path) + cwd, + db_path, + since_sha=_last_indexed_sha(db_path), + workers=workers, + batch_size=batch_size, ) except Exception as exc: # noqa: BLE001 - never block on index failure logger.warning("index.refresh_failed", err=str(exc)) diff --git a/src/config/schema.py b/src/config/schema.py index f672472..e721268 100644 --- a/src/config/schema.py +++ b/src/config/schema.py @@ -1030,13 +1030,26 @@ class AutodevConfig(BaseModel): index_full_rebuild_threshold_files: int = Field( default=5000, ge=100, le=100_000 ) - # On huge repos (``RepoCapacity.is_huge``) the initial full build can - # take minutes. When True (default), ``autodev init`` spawns the - # builder in a background subprocess and returns immediately; the - # ``.autodev/index.db.building`` marker file signals to the per-trigger - # incremental hook that it should skip until the build completes. - # Set False to force synchronous initial build even on huge repos. - index_huge_repo_async_init: bool = True + # Number of worker processes used by ``IndexBuilder.build_full`` to + # parse files in parallel (the parse stage is CPU- and GIL-bound, so + # processes, not threads). ``0`` (default) = ``os.cpu_count() or 1``; + # ``1`` forces the serial in-process parse. Workers feed a single + # bulk-loading sqlite writer in the parent. + index_build_workers: int = Field(default=0, ge=0) + # Files per write transaction during a full bulk build. Bounds the WAL + # so a huge repo doesn't accumulate one giant single-transaction blob + # (the pre-parallel builder produced a ~600 MB WAL that only flushed at + # the end). Lower it on memory-constrained hosts. + index_build_batch_size: int = Field(default=1000, ge=1) + # On huge repos (``RepoCapacity.is_huge``) the initial full build used + # to take minutes, so it was spawned in a background subprocess. Now + # that the build is parallel + bulk-loaded it is fast enough to run + # synchronously, so the default is False (sync). Set True to restore + # the opt-in async escape hatch: ``autodev init`` spawns the builder in + # a background subprocess and returns immediately; the + # ``.autodev/index.db.building`` marker signals the per-trigger + # incremental hook to skip until the build completes. + index_huge_repo_async_init: bool = False # v0.30.0 Bug 5: cross-task infrastructure-failure circuit breaker. # Counts adapter failures whose ``subtype`` is in diff --git a/src/state/file_index.py b/src/state/file_index.py index 28b8123..f627ed4 100644 --- a/src/state/file_index.py +++ b/src/state/file_index.py @@ -39,12 +39,13 @@ import hashlib import json import logging +import multiprocessing as mp import os import re import sqlite3 import subprocess import time -from collections.abc import Callable +from collections.abc import Callable, Iterable from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path @@ -435,6 +436,156 @@ def _content_hash(text: bytes) -> str: return hashlib.sha256(text).hexdigest()[:16] +# --------------------------------------------------------------------------- +# Parallel parse (CPU-bound, GIL-bound → processes, not threads) +# --------------------------------------------------------------------------- +# +# The parse stage (read bytes → hash → stat → decode → extractor.extract()) +# is CPU- and GIL-bound, so we fan it out across worker *processes*. The +# write stage stays single-threaded in the parent (sqlite single-writer). +# Serial and parallel paths feed the SAME writer (``_bulk_write`` / +# ``_index_parsed_serial``) → parity by construction. +# +# macOS defaults to the ``spawn`` start method, so the worker function and +# its arguments must be top-level + picklable. We reuse the ``forkserver`` +# context (mirrors ``qa/sandbox.py``). Workers re-import this module; the +# extractors are module-level singletons that re-init idempotently, so the +# worker just calls ``lookup_extractor(suffix)``. Workers do NO sqlite I/O. + + +@dataclass(frozen=True) +class _ParsedFile: + """Result of parsing one file in a worker process (picklable). + + ``symbols`` is a tuple of ``(name, kind, signature, line, col)`` tuples + so the dataclass is trivially picklable. ``rel`` is the repo-relative + POSIX path used as the ``files.path`` primary key. + """ + + rel: str + content_hash: str + mtime_ns: int + size_bytes: int + lang: str + symbols: tuple[tuple[str, str, str, int, int], ...] + + +# Set by the pool initializer in each worker (picklable string arg). Workers +# need the repo root to compute the repo-relative path. +_WORKER_CWD: str | None = None + + +def _init_worker(cwd: str) -> None: + """Pool initializer: stash the repo root in a module global.""" + global _WORKER_CWD + _WORKER_CWD = cwd + + +def _parse_one(abs_path: Path, cwd: Path) -> _ParsedFile | None: + """Parse a single file into a :class:`_ParsedFile` (no DB access). + + Best-effort parity with :func:`_index_one_file`: read/decode/extract + failures yield an empty symbol list; a path outside *cwd* or an + unreadable/vanished file yields ``None`` (skipped). + """ + try: + rel = abs_path.relative_to(cwd).as_posix() + except ValueError: + return None + try: + raw = abs_path.read_bytes() + except OSError: + return None + try: + st = abs_path.stat() + except OSError: + return None + chash = _content_hash(raw) + extractor = lookup_extractor(abs_path.suffix.lower()) + try: + text = raw.decode("utf-8") + except UnicodeDecodeError: + text = raw.decode("utf-8", errors="replace") + try: + extracted = extractor.extract(text) + except Exception as exc: # noqa: BLE001 — never let one file kill the build + _log.debug("file_index.extract_failed path=%s err=%s", rel, str(exc)) + extracted = [] + symbols = tuple( + ( + sym["name"], + sym["kind"], + sym["signature"], + int(sym["line"]), + int(sym["col"]), + ) + for sym in extracted + ) + return _ParsedFile( + rel=rel, + content_hash=chash, + mtime_ns=st.st_mtime_ns, + size_bytes=st.st_size, + lang=extractor.lang_tag, + symbols=symbols, + ) + + +def _parse_file_worker(abs_str: str) -> _ParsedFile | None: + """Top-level, picklable worker entrypoint for the process pool. + + Uses the pool-initialized ``_WORKER_CWD`` to resolve repo-relative + paths. Returns ``None`` for files that should be skipped. + """ + if _WORKER_CWD is None: # pragma: no cover — initializer always runs first + return None + return _parse_one(Path(abs_str), Path(_WORKER_CWD)) + + +def _iter_parsed( + files: list[Path], + cwd: Path, + workers: int, +) -> Iterator[_ParsedFile]: + """Yield :class:`_ParsedFile` for each file, parallel when ``workers>1``. + + Parallel path: a ``forkserver`` ``Pool`` runs :func:`_parse_file_worker` + via ``imap_unordered``. On any pool-construction failure we fall back to + the serial path so a build never hard-fails on a multiprocessing quirk. + Serial path (``workers<=1`` or fallback): parse in-process. ``None`` + results (skipped files) are filtered out in both paths. + """ + if workers > 1 and files: + try: + ctx = mp.get_context("forkserver") + with ctx.Pool( + processes=workers, + initializer=_init_worker, + initargs=(str(cwd),), + ) as pool: + for parsed in pool.imap_unordered( + _parse_file_worker, + [str(p) for p in files], + chunksize=64, + ): + if parsed is not None: + yield parsed + return + except Exception as exc: # noqa: BLE001 — fall back to serial + _log.warning( + "file_index.pool_init_failed workers=%d err=%s; " + "falling back to serial parse", + workers, + str(exc), + ) + + # Serial fallback (workers<=1, empty file list, or pool failure). + for abs_path in files: + parsed = _parse_one(abs_path, cwd) + if parsed is not None: + yield parsed + + def _index_one_file( conn: sqlite3.Connection, cwd: Path, @@ -518,6 +669,161 @@ def _delete_file_row(conn: sqlite3.Connection, rel: str) -> None: conn.execute("DELETE FROM files_fts WHERE path=?", (rel,)) +# --------------------------------------------------------------------------- +# Bulk-loading writer (Step 2) — single-threaded, batched, FTS rebuild +# --------------------------------------------------------------------------- + + +def _drop_symbols_fts_triggers(conn: sqlite3.Connection) -> None: + """Drop the per-row external-content FTS sync triggers. + + During a bulk load we populate ``symbols_fts`` once via the + ``('rebuild')`` command instead of firing ``symbols_ai`` on every + INSERT (orders of magnitude faster on huge repos). + """ + conn.execute("DROP TRIGGER IF EXISTS symbols_ai") + conn.execute("DROP TRIGGER IF EXISTS symbols_ad") + + +# The AI/AD trigger bodies, factored out so the bulk path can re-create the +# exact same triggers it dropped (keeping ``build_incremental`` per-file +# inserts populating the external-content FTS). +_SYMBOLS_FTS_TRIGGERS_DDL = """ +CREATE TRIGGER IF NOT EXISTS symbols_ai AFTER INSERT ON symbols BEGIN + INSERT INTO symbols_fts(rowid, name, file_path, signature) + VALUES (new.id, new.name, new.file_path, new.signature); +END; +CREATE TRIGGER IF NOT EXISTS symbols_ad AFTER DELETE ON symbols BEGIN + INSERT INTO symbols_fts(symbols_fts, rowid, name, file_path, signature) + VALUES('delete', old.id, old.name, old.file_path, old.signature); +END; +""" + + +def _recreate_symbols_fts_triggers(conn: sqlite3.Connection) -> None: + """Re-create the ``symbols_ai`` / ``symbols_ad`` sync triggers.""" + conn.executescript(_SYMBOLS_FTS_TRIGGERS_DDL) + + +def _bulk_write( + conn: sqlite3.Connection, + parsed_iter: Iterable[_ParsedFile], + indexed_at: int, + batch_size: int, + progress_cb: Callable[[int, int], None] | None, + total: int, + start: float, +) -> tuple[int, int]: + """Bulk-load parsed files into a freshly-wiped index. + + Assumes the db was just created (``build_full`` wipes first), so there + are no pre-existing rows to skip/delete. The ``symbols_fts`` + external-content sync triggers are dropped for the duration; FTS is + repopulated once at the end via ``INSERT INTO + symbols_fts(symbols_fts) VALUES('rebuild')``, then the triggers are + re-created so ``build_incremental`` per-file inserts keep working. + + Commits every *batch_size* files to bound the WAL (no 600 MB + single-transaction blob). Returns ``(file_count, symbol_count)``. + + Note: ``files_fts`` is a *standalone* trigram FTS (no triggers), so we + insert its ``path`` rows directly alongside the ``files`` rows. + """ + _drop_symbols_fts_triggers(conn) + + file_count = 0 + symbol_count = 0 + files_batch: list[tuple] = [] + files_fts_batch: list[tuple] = [] + symbols_batch: list[tuple] = [] + + def _flush() -> None: + if files_batch: + conn.executemany( + "INSERT INTO files(path, content_hash, mtime_ns, size_bytes, " + "lang, indexed_at) VALUES (?, ?, ?, ?, ?, ?)", + files_batch, + ) + conn.executemany( + "INSERT INTO files_fts(path) VALUES (?)", files_fts_batch + ) + if symbols_batch: + conn.executemany( + "INSERT INTO symbols(file_path, name, kind, signature, line, " + "col) VALUES (?, ?, ?, ?, ?, ?)", + symbols_batch, + ) + files_batch.clear() + files_fts_batch.clear() + symbols_batch.clear() + + conn.execute("BEGIN") + in_txn = True + try: + for parsed in parsed_iter: + files_batch.append( + ( + parsed.rel, + parsed.content_hash, + parsed.mtime_ns, + parsed.size_bytes, + parsed.lang, + indexed_at, + ) + ) + files_fts_batch.append((parsed.rel,)) + for sym in parsed.symbols: + symbols_batch.append( + (parsed.rel, sym[0], sym[1], sym[2], sym[3], sym[4]) + ) + file_count += 1 + symbol_count += len(parsed.symbols) + + if progress_cb is not None and ( + file_count % 1000 == 0 or file_count == total + ): + try: + progress_cb(file_count, total) + except Exception: # noqa: BLE001 + pass + if file_count % 5000 == 0: + _log.info( + "file_index.build_progress files_done=%d files_total=%d " + "elapsed_ms=%d", + file_count, + total, + int((time.monotonic() - start) * 1000), + ) + + # Commit per batch to bound the WAL. + if len(files_batch) >= batch_size: + _flush() + conn.execute("COMMIT") + conn.execute("BEGIN") + # Final partial batch. + _flush() + conn.execute("COMMIT") + in_txn = False + + # One-shot external-content FTS rebuild → identical to the + # trigger-populated index. + conn.execute("INSERT INTO symbols_fts(symbols_fts) VALUES('rebuild')") + # Restore the per-row sync triggers for build_incremental. + _recreate_symbols_fts_triggers(conn) + except Exception: + if in_txn: + conn.execute("ROLLBACK") + # Best-effort: leave the schema with triggers re-created so a + # subsequent incremental isn't left without FTS sync. + try: + _recreate_symbols_fts_triggers(conn) + except sqlite3.DatabaseError: + pass + raise + + return file_count, symbol_count + + def _set_meta(conn: sqlite3.Connection, **kv: str | int) -> None: for k, v in kv.items(): conn.execute( @@ -596,9 +902,16 @@ def build_full( db_path: Path, languages: list[str] | None = None, progress_cb: Callable[[int, int], None] | None = None, + workers: int = 0, + batch_size: int = 1000, ) -> IndexStats: """Rebuild the index from scratch over every tracked file. + The parse stage (read + decode + symbol extraction) is fanned out + across worker *processes* and feeds a single bulk-loading sqlite + writer in the parent. Serial (``workers<=1``) and parallel paths + share the same writer, so the produced index is identical. + Args: cwd: Repo root. db_path: Where to write the sqlite db. @@ -608,6 +921,11 @@ def build_full( progress_cb: Optional ``(files_done, files_total)`` callback. Called once per ~1000 files to give the operator liveness on huge repos. + workers: Number of parse worker processes. ``0`` (default) + means ``os.cpu_count() or 1``. ``1`` forces the serial + in-process parse path. + batch_size: Number of files per write transaction. Bounds the + WAL so a huge repo doesn't accumulate one giant commit. Returns: :class:`IndexStats` summarizing the run. @@ -660,57 +978,49 @@ def build_full( ) indexed_at = int(time.time()) - file_count = 0 - symbol_count = 0 + resolved_workers = workers if workers > 0 else ( + os.cpu_count() or 1 + ) files = list(iter_repo_files(cwd)) total = len(files) - conn.execute("BEGIN") - try: - for idx, abs_path in enumerate(files, start=1): - if languages is not None: - ext = lookup_extractor(abs_path.suffix.lower()) - if ext.lang_tag not in languages: - continue - changed, syms = _index_one_file( - conn, cwd, abs_path, indexed_at - ) - if changed: - file_count += 1 - symbol_count += syms - if progress_cb is not None and ( - idx % 1000 == 0 or idx == total - ): - try: - progress_cb(idx, total) - except Exception: # noqa: BLE001 - pass - if idx % 5000 == 0: - _log.info( - "file_index.build_progress files_done=%d files_total=%d elapsed_ms=%d", - idx, - total, - int((time.monotonic() - start) * 1000), - ) - sha = _current_git_sha(cwd) - duration_ms = int((time.monotonic() - start) * 1000) - final_file_count = conn.execute( - "SELECT COUNT(*) FROM files" - ).fetchone()[0] - final_symbol_count = conn.execute( - "SELECT COUNT(*) FROM symbols" - ).fetchone()[0] - _set_meta( - conn, - last_indexed_sha=sha or "", - last_indexed_at=indexed_at, - file_count=final_file_count, - symbol_count=final_symbol_count, - build_duration_ms=duration_ms, + + # Parse stage (parallel via forkserver, else serial). The + # ``files_fts`` rows + symbols are written serially in the + # parent by ``_bulk_write``, which shares the parse output + # with the serial path → parity by construction. + parsed_iter = _iter_parsed(files, cwd, resolved_workers) + if languages is not None: + allow = set(languages) + parsed_iter = ( + p for p in parsed_iter if p.lang in allow ) - conn.execute("COMMIT") - except Exception: - conn.execute("ROLLBACK") - raise + + file_count, symbol_count = _bulk_write( + conn, + parsed_iter, + indexed_at, + batch_size, + progress_cb, + total, + start, + ) + + sha = _current_git_sha(cwd) + duration_ms = int((time.monotonic() - start) * 1000) + final_file_count = conn.execute( + "SELECT COUNT(*) FROM files" + ).fetchone()[0] + final_symbol_count = conn.execute( + "SELECT COUNT(*) FROM symbols" + ).fetchone()[0] + _set_meta( + conn, + last_indexed_sha=sha or "", + last_indexed_at=indexed_at, + file_count=final_file_count, + symbol_count=final_symbol_count, + build_duration_ms=duration_ms, + ) finally: conn.close() @@ -743,6 +1053,8 @@ def build_incremental( db_path: Path, since_sha: str | None, full_rebuild_threshold: int = _DEFAULT_FULL_REBUILD_THRESHOLD, + workers: int = 0, + batch_size: int = 1000, ) -> IndexStats: """Incrementally update the index. @@ -758,6 +1070,10 @@ def build_incremental( mtime-vs-row-mtime check across the full file inventory (still cheap because :func:`_index_one_file` skips unchanged rows). + ``workers`` / ``batch_size`` are forwarded only to the + :meth:`build_full` delegations (the targeted-changed path stays + serial — change-sets are small). + Idempotent: holds ``.autodev/index.db.lock`` for the duration; on lock-held returns a no-op :class:`IndexStats`. """ @@ -765,7 +1081,9 @@ def build_incremental( start = time.monotonic() if not db_path.exists(): - return IndexBuilder.build_full(cwd, db_path) + return IndexBuilder.build_full( + cwd, db_path, workers=workers, batch_size=batch_size + ) # Schema-version check (cheap migration trigger). try: @@ -785,7 +1103,9 @@ def build_incremental( stored_version, INDEX_SCHEMA_VERSION, ) - return IndexBuilder.build_full(cwd, db_path) + return IndexBuilder.build_full( + cwd, db_path, workers=workers, batch_size=batch_size + ) # If git diff is available + since_sha is set, peek at the # changed-set BEFORE acquiring the lock so we can route to @@ -802,7 +1122,9 @@ def build_incremental( len(peeked_changed), full_rebuild_threshold, ) - return IndexBuilder.build_full(cwd, db_path) + return IndexBuilder.build_full( + cwd, db_path, workers=workers, batch_size=batch_size + ) with _builder_lock(db_path) as held: if not held: @@ -1279,3 +1601,45 @@ def _to_fts_query(text: str) -> str: "SymbolHit", "_last_indexed_sha", ] + + +def _main(argv: list[str] | None = None) -> int: + """CLI entrypoint for the opt-in async (subprocess) full build. + + ``autodev init`` spawns ``python -m state.file_index build-full + --cwd --db [--workers N] [--batch-size N]`` for huge-repo + async builds. Prior to this entrypoint the module had no ``__main__`` + block, so the subprocess loaded the module and exited doing nothing — + the async path never indexed. This restores it. + """ + import argparse + + ap = argparse.ArgumentParser(prog="state.file_index") + sub = ap.add_subparsers(dest="command", required=True) + bf = sub.add_parser("build-full", help="Full rebuild of the index.") + bf.add_argument("--cwd", type=Path, required=True) + bf.add_argument("--db", type=Path, required=True) + bf.add_argument("--workers", type=int, default=0) + bf.add_argument("--batch-size", type=int, default=1000) + args = ap.parse_args(argv) + + if args.command == "build-full": + logging.basicConfig(level=logging.INFO) + stats = IndexBuilder.build_full( + args.cwd, + args.db, + workers=args.workers, + batch_size=args.batch_size, + ) + _log.info( + "file_index.build_full_done files=%d symbols=%d duration_ms=%d", + stats.file_count, + stats.symbol_count, + stats.duration_ms, + ) + return 0 + return 2 + + +if __name__ == "__main__": + raise SystemExit(_main()) diff --git a/tests/test_cli_init_with_index.py b/tests/test_cli_init_with_index.py index bcb96f8..ff621cd 100644 --- a/tests/test_cli_init_with_index.py +++ b/tests/test_cli_init_with_index.py @@ -144,12 +144,24 @@ def test_init_async_for_huge_repo(tmp_path: Path) -> None: return_value=_FakeCapacity(is_huge=True) ) + # The async escape hatch is opt-in (default flipped to sync in the + # parallel-build work). Enable it explicitly to exercise the spawn path. + from config.defaults import default_config as _real_default_config + + def _async_enabled_config(): + cfg = _real_default_config() + cfg.index_huge_repo_async_init = True + return cfg + with runner.isolated_filesystem(temp_dir=tmp_path) as cwd: with mock.patch.dict( "sys.modules", {"state.file_index": fake_index_module} ), mock.patch( "runtime.repo_probe.probe_repo", fake_probe_module.probe_repo, + ), mock.patch( + "cli.commands.init.default_config", + _async_enabled_config, ), mock.patch( "subprocess.Popen" ) as mock_popen: diff --git a/tests/test_cli_status.py b/tests/test_cli_status.py index 5f54e8f..2fa2371 100644 --- a/tests/test_cli_status.py +++ b/tests/test_cli_status.py @@ -320,7 +320,7 @@ def test_status_blocked_lists_dump_paths(tmp_path: Path) -> None: plan = _make_blocked_plan() out = StringIO() - console = Console(file=out, force_terminal=False) + console = Console(file=out, force_terminal=False, width=200) _render_blocked_section(console, plan, cwd=tmp_path) rendered = out.getvalue() assert "Archived Rejected Plans" in rendered diff --git a/tests/test_state_file_index_parallel.py b/tests/test_state_file_index_parallel.py new file mode 100644 index 0000000..3e7e4e0 --- /dev/null +++ b/tests/test_state_file_index_parallel.py @@ -0,0 +1,347 @@ +"""Parity + correctness tests for the parallelized index build. + +Covers the two-step parallelization of :meth:`IndexBuilder.build_full`: + + * Step 1 / Step 2 parity: ``workers=1`` and ``workers=4`` must produce + identical ``files`` + ``symbols`` rows (modulo autoincrement ``id``) + and identical :class:`IndexQuery` results. + * FTS correctness after the external-content ``('rebuild')`` path. + * Edge cases: binary/unparseable file, empty repo, a file deleted + before the worker parses it (worker returns ``None``). + * Async-entry regression: ``python -m state.file_index build-full`` must + produce a NON-EMPTY index (it was previously a silent no-op). + +Fixture pattern mirrors ``tests/test_state_file_index_query.py``. +""" + +from __future__ import annotations + +import os +import sqlite3 +import subprocess +import sys +from pathlib import Path + +from state.file_index import IndexBuilder, IndexQuery + + +def _git_init(repo: Path) -> None: + subprocess.run(["git", "init", "-q"], cwd=str(repo), check=True) + subprocess.run( + ["git", "config", "user.email", "t@t"], cwd=str(repo), check=True + ) + subprocess.run( + ["git", "config", "user.name", "t"], cwd=str(repo), check=True + ) + + +def _git_commit(repo: Path) -> None: + subprocess.run(["git", "add", "-A"], cwd=str(repo), check=True) + subprocess.run( + ["git", "commit", "-qm", "init"], cwd=str(repo), check=True + ) + + +def _write_sources(repo: Path) -> None: + """Write a small multi-language source tree under *repo*.""" + (repo / ".gitignore").write_text(".autodev/\n") + (repo / "src").mkdir() + (repo / "src" / "parse_plan.py").write_text( + "def parse_plan_markdown(md: str) -> dict:\n" + " return {}\n" + "\n" + "class PlanParseError(Exception):\n" + " pass\n" + ) + (repo / "src" / "validate_files.py").write_text( + "def validate_files_exist(plan, cwd):\n" + " return None\n" + ) + (repo / "src" / "common.py").write_text( + "def helper():\n" + " return 1\n" + "\n" + "MAX_RETRIES = 3\n" + ) + (repo / "src" / "widget.cpp").write_text( + "namespace ns {\n" + "int compute_widget(int a) { return a; }\n" + "}\n" + ) + (repo / "src" / "app.ts").write_text( + "export function renderApp(props: any) { return props; }\n" + "export class AppController {}\n" + ) + (repo / "README.md").write_text("# fixture repo\n") + + +def _make_repo(tmp_path: Path, name: str) -> Path: + repo = tmp_path / name + repo.mkdir() + _git_init(repo) + _write_sources(repo) + _git_commit(repo) + return repo + + +def _dump_files(db: Path) -> list[tuple]: + """Return ``files`` rows ordered by path (id-independent).""" + conn = sqlite3.connect(str(db)) + try: + rows = conn.execute( + "SELECT path, content_hash, mtime_ns, size_bytes, lang " + "FROM files ORDER BY path" + ).fetchall() + finally: + conn.close() + return rows + + +def _dump_symbols(db: Path) -> list[tuple]: + """Return ``symbols`` rows ordered deterministically, sans ``id``.""" + conn = sqlite3.connect(str(db)) + try: + rows = conn.execute( + "SELECT file_path, name, kind, signature, line, col " + "FROM symbols ORDER BY file_path, name, kind, line, col" + ).fetchall() + finally: + conn.close() + return rows + + +# --------------------------------------------------------------------------- +# Parity: serial (workers=1) vs parallel (workers=4) +# --------------------------------------------------------------------------- + + +def test_serial_vs_parallel_files_and_symbols_identical(tmp_path: Path) -> None: + """workers=1 and workers=4 produce identical files + symbols rows.""" + repo = _make_repo(tmp_path, "r") + db_serial = tmp_path / "serial.db" + db_par = tmp_path / "parallel.db" + + IndexBuilder.build_full(repo, db_serial, workers=1) + IndexBuilder.build_full(repo, db_par, workers=4) + + assert _dump_files(db_serial) == _dump_files(db_par) + assert _dump_symbols(db_serial) == _dump_symbols(db_par) + # Sanity: the fixture actually produced rows. + assert len(_dump_files(db_serial)) >= 6 + assert len(_dump_symbols(db_serial)) >= 3 + + +def test_serial_vs_parallel_query_results_identical(tmp_path: Path) -> None: + """IndexQuery results identical across workers=1 and workers=4.""" + repo = _make_repo(tmp_path, "r") + db_serial = tmp_path / "serial.db" + db_par = tmp_path / "parallel.db" + + IndexBuilder.build_full(repo, db_serial, workers=1) + IndexBuilder.build_full(repo, db_par, workers=4) + + qs = IndexQuery(db_serial) + qp = IndexQuery(db_par) + try: + for term in ("parse_plan_markdown", "parse", "validate", "widget", "render"): + assert qs.search_symbols(term) == qp.search_symbols(term), term + for pat in ("parse_plan", "src", "validate", ".py", "app"): + assert qs.search_files(pat) == qp.search_files(pat), pat + for spec in ( + "refactor parsePlanMarkdown and validateFilesExist", + "render the app controller widget", + ): + ds = qs.get_candidates_for_spec(spec) + dp = qp.get_candidates_for_spec(spec) + assert ds.symbol_hits == dp.symbol_hits, spec + assert ds.file_hits == dp.file_hits, spec + finally: + qs.close() + qp.close() + + +def test_fts_hits_correct_after_rebuild(tmp_path: Path) -> None: + """symbols_fts / files_fts MATCH return expected fixture hits. + + Exercises the external-content FTS ``('rebuild')`` path (Step 2) plus + the standalone trigram ``files_fts``. + """ + repo = _make_repo(tmp_path, "r") + db = tmp_path / "idx.db" + IndexBuilder.build_full(repo, db, workers=4) + + conn = sqlite3.connect(str(db)) + try: + # External-content symbols_fts MATCH must resolve via rowid join. + rows = conn.execute( + "SELECT s.name FROM symbols s JOIN symbols_fts f ON s.id = f.rowid " + "WHERE symbols_fts MATCH 'parse_plan_markdown'" + ).fetchall() + assert any(r[0] == "parse_plan_markdown" for r in rows) + + # files_fts trigram MATCH (standalone). + frows = conn.execute( + "SELECT path FROM files_fts WHERE files_fts MATCH 'parse_plan'" + ).fetchall() + assert any("parse_plan.py" in r[0] for r in frows) + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + + +def test_binary_unparseable_file_indexed_with_no_symbols(tmp_path: Path) -> None: + """A binary file appears in ``files`` (lang=other) with no symbols.""" + repo = tmp_path / "r" + repo.mkdir() + _git_init(repo) + (repo / ".gitignore").write_text(".autodev/\n") + (repo / "good.py").write_text("def f():\n return 1\n") + (repo / "blob.bin").write_bytes(bytes(range(256)) * 4) + _git_commit(repo) + + db_serial = tmp_path / "serial.db" + db_par = tmp_path / "parallel.db" + IndexBuilder.build_full(repo, db_serial, workers=1) + IndexBuilder.build_full(repo, db_par, workers=4) + + assert _dump_files(db_serial) == _dump_files(db_par) + assert _dump_symbols(db_serial) == _dump_symbols(db_par) + + files = {r[0] for r in _dump_files(db_par)} + assert "blob.bin" in files + # No symbols attributed to the binary file. + bin_syms = [s for s in _dump_symbols(db_par) if s[0] == "blob.bin"] + assert bin_syms == [] + + +def test_empty_repo(tmp_path: Path) -> None: + """An empty repo builds a valid, empty index for both worker counts.""" + repo = tmp_path / "empty" + repo.mkdir() + _git_init(repo) + # A single committed file we then ignore: keep repo non-degenerate for + # git, but exercise the "no source files of interest" shape by only + # committing a gitignore (git ls-files yields just .gitignore). + (repo / ".gitignore").write_text(".autodev/\n") + _git_commit(repo) + + db_serial = tmp_path / "serial.db" + db_par = tmp_path / "parallel.db" + stats_s = IndexBuilder.build_full(repo, db_serial, workers=1) + stats_p = IndexBuilder.build_full(repo, db_par, workers=4) + + assert _dump_files(db_serial) == _dump_files(db_par) + assert _dump_symbols(db_serial) == _dump_symbols(db_par) + assert stats_s.symbol_count == stats_p.symbol_count == 0 + # IndexQuery opens cleanly and returns nothing. + q = IndexQuery(db_par) + try: + assert q.search_symbols("anything") == [] + finally: + q.close() + + +def test_truly_empty_repo_no_tracked_files(tmp_path: Path) -> None: + """A git repo with zero tracked files builds an empty index.""" + repo = tmp_path / "void" + repo.mkdir() + _git_init(repo) + + db = tmp_path / "void.db" + stats = IndexBuilder.build_full(repo, db, workers=4) + assert stats.file_count == 0 + assert stats.symbol_count == 0 + assert _dump_files(db) == [] + + +def test_file_deleted_before_parse_returns_none(tmp_path: Path) -> None: + """A path that git tracks but vanishes before parse is skipped cleanly. + + Simulates the worker returning ``None`` for a removed file: the index + build must not crash and must omit the missing file. + """ + repo = tmp_path / "r" + repo.mkdir() + _git_init(repo) + (repo / ".gitignore").write_text(".autodev/\n") + (repo / "keep.py").write_text("def keep():\n return 1\n") + (repo / "ghost.py").write_text("def ghost():\n return 2\n") + _git_commit(repo) + + # Delete ghost.py AFTER commit so git ls-files still lists it but the + # worker cannot read it. + (repo / "ghost.py").unlink() + + db = tmp_path / "idx.db" + stats = IndexBuilder.build_full(repo, db, workers=4) + + files = {r[0] for r in _dump_files(db)} + assert "keep.py" in files + assert "ghost.py" not in files + assert stats.file_count >= 1 + + +def test_default_workers_zero_builds_index(tmp_path: Path) -> None: + """workers=0 (auto = cpu_count) builds a valid, non-empty index.""" + repo = _make_repo(tmp_path, "r") + db = tmp_path / "idx.db" + stats = IndexBuilder.build_full(repo, db) # default workers=0 + assert stats.file_count >= 6 + assert stats.symbol_count >= 3 + + +def test_backward_compatible_signature(tmp_path: Path) -> None: + """Existing callers using build_full(cwd, db) keep working unchanged.""" + repo = _make_repo(tmp_path, "r") + db = tmp_path / "idx.db" + stats = IndexBuilder.build_full(repo, db) + assert stats.file_count >= 6 + + +# --------------------------------------------------------------------------- +# Async-entry regression: the __main__ build-full entrypoint +# --------------------------------------------------------------------------- + + +def test_async_entry_module_main_produces_nonempty_index(tmp_path: Path) -> None: + """`python -m state.file_index build-full` must build a NON-EMPTY index. + + Guards the bug where the module had no ``__main__`` block, so the + async build subprocess loaded the module and exited doing nothing. + """ + repo = _make_repo(tmp_path, "r") + db = tmp_path / "async.db" + + src_dir = Path(__file__).resolve().parents[1] / "src" + env = dict(os.environ) + env["PYTHONPATH"] = str(src_dir) + os.pathsep + env.get("PYTHONPATH", "") + + result = subprocess.run( + [ + sys.executable, + "-m", + "state.file_index", + "build-full", + "--cwd", + str(repo), + "--db", + str(db), + "--workers", + "2", + ], + cwd=str(repo), + env=env, + capture_output=True, + text=True, + timeout=120, + ) + assert result.returncode == 0, result.stderr + assert db.exists(), result.stderr + files = _dump_files(db) + symbols = _dump_symbols(db) + assert len(files) >= 6, f"empty index: {result.stderr}" + assert len(symbols) >= 3, f"no symbols: {result.stderr}"