From 0068bf72973968e0d390d0a32a07c8650b3ae984 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:10:00 -0700 Subject: [PATCH 1/9] store: do not advance the semantics stamp on a zero-parse run --- indexer/store.py | 40 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/indexer/store.py b/indexer/store.py index 84e110a..d4e05c2 100644 --- a/indexer/store.py +++ b/indexer/store.py @@ -112,7 +112,11 @@ def index_repo( set is empty** -- an empty seen-set would otherwise strip ``branch`` from every row in the repo; conservatively skipping is safer than wiping. 5. CAS-stamp the ``repo_branches`` row for ``(repo_id, branch)`` against the - step-2 baseline (raises :class:`StaleIndexError` on mismatch). + step-2 baseline (raises :class:`StaleIndexError` on mismatch). A run whose + seen-set was EMPTY advances ``last_indexed_commit`` but leaves + ``index_semantics_version`` at the step-2 baseline -- it wrote nothing, so + it indexed nothing at the current semantics version. See + :func:`_stamp_repo_branch`. ``items`` may be a lazy generator; it is consumed inside the open transaction so memory stays bounded. ``chunk_writer`` defaults to ``None``, @@ -269,6 +273,7 @@ def index_repo( head_sha=head_sha, baseline_commit=baseline_commit, baseline_version=baseline_version, + seen_any=bool(seen_paths), ) return IndexCounts(files=file_count, symbols=symbol_count, swept=swept, edges=edge_count) @@ -339,12 +344,43 @@ def _stamp_repo_branch( head_sha: str, baseline_commit: str | None, baseline_version: int | None, + seen_any: bool = True, ) -> None: """Compare-and-set the ``repo_branches`` stamp against the statement-2 baseline. Raises :class:`StaleIndexError` if the row no longer matches the baseline, which propagates out of ``index_repo``'s ``conn.begin()`` and rolls the whole ``(repo, branch)`` transaction back rather than regressing the index. + + **``seen_any=False`` holds the semantics version at ``baseline_version``.** + A run that parsed zero indexable files (the transient case + ``_sweep_membership``'s empty-seen-set guard exists for) has written nothing, + so it has not indexed anything at the CURRENT semantics version and must not + claim to have. Without this the following is silent and terminal: + + 1. ``INDEX_SEMANTICS_VERSION`` goes 4 -> 5; branch ``b`` is stored at + ``(sha1, 4)``, so the skip seam forces a re-index. + 2. That re-index parses zero files. Nothing is written -- but the stamp + advances to ``(sha2, 5)``. + 3. The next run sees ``baseline_version == 5 == current``, opens the + file-level delta gate, and finds every row carrying ``b`` unchanged on + ``(path, content_sha)`` -- so it skips them all. + 4. ``b`` serves v4-extracted rows under a v5 stamp, permanently. + + ``last_indexed_commit`` still advances to ``head_sha``: the commit IS what + this run looked at. Leaving the version behind is what makes the branch + mismatch (and therefore re-index) on its next run -- self-healing, in the + safe direction. The statement shape, the CAS predicate, and + :class:`StaleIndexError` are untouched. + + **Known, deliberate divergence:** ``index_repo``'s statement 1 writes the + DEPRECATED ``repos.index_semantics_version`` unconditionally on + ``is_default``, with no seen-set awareness. So a zero-parse default-branch + run leaves ``repos`` at the current version while ``repo_branches`` sits at + the old one. That is cosmetic -- no decision anywhere reads + ``repos.index_semantics_version`` (the three legacy columns are documented + deprecated in ``app/db/models.py``) -- and extending this fix to the legacy + stamp is scope this change deliberately does not take. """ result = conn.execute( update(RepoBranch) @@ -356,7 +392,7 @@ def _stamp_repo_branch( ) .values( last_indexed_commit=head_sha, - index_semantics_version=INDEX_SEMANTICS_VERSION, + index_semantics_version=(INDEX_SEMANTICS_VERSION if seen_any else baseline_version), last_indexed_at=func.now(), ) ) From 1f69da2d66b43ce457d7e51a961b8f0ff76eaca0 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:12:35 -0700 Subject: [PATCH 2/9] store: read helpers + delta classification behind the semantics gate --- indexer/store.py | 155 +++++++++++++-- tests/unit/test_store_delta.py | 352 +++++++++++++++++++++++++++++++++ 2 files changed, 496 insertions(+), 11 deletions(-) create mode 100644 tests/unit/test_store_delta.py diff --git a/indexer/store.py b/indexer/store.py index d4e05c2..d164894 100644 --- a/indexer/store.py +++ b/indexer/store.py @@ -69,6 +69,81 @@ class ReconcileCounts: # computed -- this seam never calls an embedder itself. ChunkWriter = Callable[[Connection, int, int, ParsedFile], None] +# The two projection reads behind the file-level delta path. Returned as +# ``(carried, present)`` -- see read_repo_content_shas. +ContentShaSets = tuple[set[tuple[str, str]], set[tuple[str, str]]] + + +def read_repo_content_shas(conn: Connection, *, repo_id: int, branch: str) -> ContentShaSets: + """Project one repo's stored ``(path, content_sha)`` pairs, twice. + + Returns ``(carried, present)``: + + * ``carried`` -- the pairs on rows whose ``branches`` array already contains + ``branch``. A parsed file in this set is UNCHANGED for this branch. + * ``present`` -- every pair stored for this repo, on any branch. A parsed + file in ``present - carried`` already exists as a row this branch does not + yet carry, i.e. the membership-only class. + + **The projection is ``path, content_sha`` and nothing else, deliberately.** + Two expensive mistakes are available here and both must stay closed: + + * selecting ``content`` pulls the entire corpus into the worker and defeats + the whole point of the delta path; + * selecting ``branches`` forces a heap fetch per row on a table whose rows + carry that content, so the second read stops being an Index Only Scan. + + **They are also two separate statements on purpose.** Do NOT collapse them + into ``SELECT path, content_sha, branches @> ... AS carries FROM files WHERE + repo_id = :id``: that drops the branch predicate entirely (so + ``ix_files_branches_gin`` is never consulted) *and* projects ``branches``. + The containment form ``branches @> ARRAY[:branch]`` is what the GIN index can + serve; ``_sweep_membership``'s ``:branch = ANY(branches)`` cannot be. + + **Keyed on ``(path, content_sha)``, NEVER on path alone.** + ``uq_files_repo_path_sha`` permits several rows for one path with different + content (the divergent-branch case), so a path-keyed dict silently drops rows + and which one survives depends on row order. + """ + carried = { + (row.path, row.content_sha) + for row in conn.execute( + text( + "SELECT path, content_sha FROM files " + "WHERE repo_id = :repo_id AND branches @> CAST(:branch_arr AS text[])" + ), + {"repo_id": repo_id, "branch_arr": [branch]}, + ) + } + present = { + (row.path, row.content_sha) + for row in conn.execute( + text("SELECT path, content_sha FROM files WHERE repo_id = :repo_id"), + {"repo_id": repo_id}, + ) + } + return carried, present + + +def read_indexed_shas(conn: Connection, *, name: str, branch: str) -> ContentShaSets: + """Name-keyed wrapper around :func:`read_repo_content_shas` for ``indexer.job``. + + Resolves ``repos.name -> id`` itself and returns two empty sets for a repo + that has never been indexed (which degrades to "everything is changed/new" -- + safe in the correct direction). This is the ADVISORY copy of the read: the + authoritative one runs inside ``index_repo``'s transaction. Both go through + the same helper so there is exactly one pair of queries and one keying rule. + + Called on its own short-lived connection that is closed BEFORE embedding + starts -- never on a connection held across network I/O. + """ + repo_id = conn.execute( + text("SELECT id FROM repos WHERE name = :name"), {"name": name} + ).scalar_one_or_none() + if repo_id is None: + return set(), set() + return read_repo_content_shas(conn, repo_id=int(repo_id), branch=branch) + def index_repo( conn: Connection, @@ -95,14 +170,24 @@ def index_repo( 2. Upsert/read the ``repo_branches`` row for ``(repo_id, branch)`` under its row lock, capturing ``(baseline_commit, baseline_version)`` -- the CAS baseline for step 5, mirroring the same ``RETURNING`` trick as step 1. - 3. Per file: an array-union upsert on ``uq_files_repo_path_sha`` -- a file - whose content already exists under another branch gets THIS branch - unioned into its ``branches`` array (one row, shared content); a file - whose content differs from every existing version gets its own row. Then - delete-and-reinsert its ``symbols`` and ``reference_edges`` (neither has a - natural key), then call ``chunk_writer`` (if given) so chunk writes - commit/roll back with the rest of that file's row. Each processed file's - ``(path, content_sha)`` is collected into this branch's seen-set. + 3. Two projection reads (:func:`read_repo_content_shas`), issued ONLY when + the delta gate is open -- see below -- then, per file, a three-way + classification on ``(pf.path, content_sha(pf.content))``: + + * **unchanged** (the pair is already on a row carrying this branch): no + statement at all. + * **changed/new** (everything else): an array-union upsert on + ``uq_files_repo_path_sha`` -- a file whose content already exists under + another branch gets THIS branch unioned into its ``branches`` array (one + row, shared content); a file whose content differs from every existing + version gets its own row. Then delete-and-reinsert its ``symbols`` and + ``reference_edges`` (neither has a natural key), then call + ``chunk_writer`` (if given) so chunk writes commit/roll back with the + rest of that file's row. + + **Every** parsed file -- classified or written -- is collected into this + branch's seen-set, so step 4 and its empty-seen-set guard are correct by + construction and untouched by the delta path. 4. Membership sweep, keyed on THIS branch's seen-set (never on ``commit``, which is ambiguous under dedup): strip ``branch`` from any row's ``branches`` array that is not in the seen-set, then delete any row left @@ -123,6 +208,36 @@ def index_repo( which makes this byte-identical to the core (semantic-off) path; when given, it must write PRECOMPUTED chunks -- embeddings are computed outside this transaction, so no network call ever happens here. + + **The delta gate.** The classification above is taken only when + ``baseline_version == INDEX_SEMANTICS_VERSION`` -- statement 2's ``RETURNING`` + value, already in hand, no extra query. A ``NULL`` or older version means + every file takes the full write path and statements 3a/3b are never issued. + + Why the unchanged path is sound, inductively: + + * A row carrying this branch was necessarily written by this branch's last + COMPLETED run, and that run ran at ``baseline_version == + INDEX_SEMANTICS_VERSION``. In it the row was either written full-path (so + it is current), or skipped as unchanged (current, by induction). + * Base case: the first run after ANY version transition has + ``baseline_version != INDEX_SEMANTICS_VERSION``, so it is full-path for + every parsed file. A zero-parse run cannot manufacture a spurious base + case -- it does not advance the version stamp (see + :func:`_stamp_repo_branch`). + + Two columns are deliberately NOT re-derived for a skipped file: + + * ``lang`` and ``size`` are pure functions of ``(path, content)`` via + ``indexer/parse.py`` + ``indexer/languages.py``, both watched by + ``tests/unit/test_semantics_version_tripwire.py`` -- so a change to either + derivation is MEANT to force a version bump, which closes this gate. That + tripwire is a local-developer guard rather than a CI one, so treat this as + a strong convention backed by review, not a machine-enforced invariant. + * ``files.commit`` goes staler. No production read path exists (every + ``commit:`` filter resolves from ``repo_branches.last_indexed_commit``, + and the column is documented write-only and ambiguous under dedup in + ``app/db/models.py``). This makes it staler; it makes nothing wrong. """ file_count = 0 symbol_count = 0 @@ -163,8 +278,29 @@ def index_repo( ) baseline_commit, baseline_version = conn.execute(branch_stmt).one() + # Statements 3a/3b, issued only behind the delta gate. Everything the + # classification below needs is now in hand; nothing else is read. + delta_on = baseline_version == INDEX_SEMANTICS_VERSION + carried: set[tuple[str, str]] = set() + present: set[tuple[str, str]] = set() + if delta_on: + carried, present = read_repo_content_shas(conn, repo_id=repo_id, branch=branch) + for pf, ex in items: sha = content_sha(pf.content) + # Seen-set membership is recorded for EVERY parsed file, whatever its + # class -- that is what keeps the sweep (and its empty-seen-set + # guard) correct without any delta awareness of its own. + file_count += 1 + seen_paths.append(pf.path) + seen_shas.append(sha) + + if delta_on and (pf.path, sha) in carried: + # Unchanged: this exact content is already stored on a row this + # branch already carries. No file upsert, no symbol/edge + # delete-reinsert, no chunk_writer call. + continue + file_stmt = ( pg_insert(File) .values( @@ -198,9 +334,6 @@ def index_repo( .returning(File.id) ) file_id = conn.execute(file_stmt).scalar_one() - file_count += 1 - seen_paths.append(pf.path) - seen_shas.append(sha) conn.execute(delete(Symbol).where(Symbol.file_id == file_id)) if ex.symbols: diff --git a/tests/unit/test_store_delta.py b/tests/unit/test_store_delta.py new file mode 100644 index 0000000..14bacdb --- /dev/null +++ b/tests/unit/test_store_delta.py @@ -0,0 +1,352 @@ +"""Unit tests for ``index_repo``'s file-level delta path, at the STATEMENT level. + +A hand-rolled fake ``Connection`` (the ``_FakeConn`` idiom from +``tests/unit/test_store_chunk_writer.py``, extended to answer the two projection +reads and the provenance gate) stands in for Postgres, so these are true unit +tests: what they pin is the exact *statement inventory* each classification +produces, and the order the transaction issues them in. Row-identity proof -- +that skipping really does leave the stored serials untouched -- is +``tests/integration/test_store_delta.py``'s job, against real SQL. + +The fake records one short label per executed statement (``repos-insert``, +``read-carried``, ``file-upsert``, ...) rather than the statement objects, so an +assertion reads as the inventory it is checking. +""" + +from __future__ import annotations + +import contextlib +from typing import Any, NamedTuple + +import pytest +from sqlalchemy import Delete, Insert, Update + +from app.db.models import INDEX_SEMANTICS_VERSION +from indexer.hashing import content_sha +from indexer.languages import ( + ExtractedEdge, + ExtractedSymbol, + FileExtraction, + IndexCounts, + ParsedFile, +) +from indexer.store import index_repo + + +class _ShaRow(NamedTuple): + """The exact projection statements 3a/3b return -- path and content_sha, nothing else.""" + + path: str + content_sha: str + + +class _IdRow(NamedTuple): + """The batched membership ``UPDATE ... RETURNING id, path, content_sha`` shape.""" + + id: int + path: str + content_sha: str + + +class _FakeResult: + def __init__( + self, + *, + scalar: Any = None, + rowcount: int = 0, + row: Any = None, + rows: list[Any] | None = None, + ) -> None: + self._scalar = scalar + self._row = row + self._rows = rows or [] + self.rowcount = rowcount + + def scalar_one(self) -> Any: + return self._scalar + + def scalar_one_or_none(self) -> Any: + return self._scalar + + def one(self) -> Any: + return self._row + + def all(self) -> list[Any]: + return self._rows + + def __iter__(self) -> Any: + return iter(self._rows) + + +class _FakeConn: + """Answers every statement ``index_repo`` can issue, and labels it. + + ``baseline`` is what statement 2's ``RETURNING`` yields -- the pair that + opens or closes the delta gate. ``carried``/``present`` script statements 3a + and 3b (see ``indexer.store.read_repo_content_shas``); ``provenance`` + scripts statement 4's ``NOT EXISTS`` gate. + """ + + def __init__( + self, + *, + baseline: tuple[str | None, int | None] = (None, None), + carried: set[tuple[str, str]] | None = None, + present: set[tuple[str, str]] | None = None, + provenance: bool = True, + stamp_rowcount: int = 1, + ) -> None: + self._next_file_id = 1 + self._baseline = baseline + self._carried = sorted(carried or set()) + self._present = sorted(present or set()) + self._provenance = provenance + self._stamp_rowcount = stamp_rowcount + self.kinds: list[str] = [] + self.stamp_values: dict[str, Any] = {} + self.membership_params: dict[str, Any] = {} + + def begin(self) -> Any: + return contextlib.nullcontext() + + def _text_execute(self, sql: str, params: Any) -> _FakeResult: + # Ordered most-specific-first: statement 3b's text is a PREFIX of 3a's. + if "SELECT path, content_sha FROM files" in sql and "branches @>" in sql: + self.kinds.append("read-carried") + return _FakeResult(rows=[_ShaRow(p, s) for p, s in self._carried]) + if "SELECT path, content_sha FROM files" in sql: + self.kinds.append("read-present") + return _FakeResult(rows=[_ShaRow(p, s) for p, s in self._present]) + if "SELECT NOT EXISTS" in sql: + self.kinds.append("provenance-gate") + return _FakeResult(scalar=self._provenance) + if "UPDATE files" in sql and "array_agg(DISTINCT e)" in sql: + self.kinds.append("membership-union") + self.membership_params = dict(params or {}) + rows = [] + for path, sha in zip( + (params or {}).get("paths", []), (params or {}).get("shas", []), strict=True + ): + rows.append(_IdRow(self._next_file_id, path, sha)) + self._next_file_id += 1 + return _FakeResult(rows=rows, rowcount=len(rows)) + if "UPDATE files SET branches = array_remove" in sql: + self.kinds.append("sweep-update") + return _FakeResult(rowcount=0) + if "DELETE FROM files" in sql: + self.kinds.append("sweep-delete") + return _FakeResult(rowcount=0) + raise AssertionError(f"unexpected text() statement: {sql!r}") + + def execute(self, stmt: Any, params: Any = None) -> _FakeResult: + sql = getattr(stmt, "text", None) + if sql is not None: + return self._text_execute(sql, params) + + table = stmt.table.name + if isinstance(stmt, Insert) and table == "repos": + self.kinds.append("repos-insert") + return _FakeResult(scalar=1) + if isinstance(stmt, Insert) and table == "repo_branches": + self.kinds.append("repo-branches-insert") + return _FakeResult(row=self._baseline) + if isinstance(stmt, Insert) and table == "files": + self.kinds.append("file-upsert") + file_id = self._next_file_id + self._next_file_id += 1 + return _FakeResult(scalar=file_id) + if isinstance(stmt, Delete) and table == "symbols": + self.kinds.append("symbols-delete") + return _FakeResult() + if isinstance(stmt, Insert) and table == "symbols": + self.kinds.append("symbols-insert") + return _FakeResult() + if isinstance(stmt, Delete) and table == "reference_edges": + self.kinds.append("edges-delete") + return _FakeResult() + if isinstance(stmt, Insert) and table == "reference_edges": + self.kinds.append("edges-insert") + return _FakeResult() + if isinstance(stmt, Update) and table == "repo_branches": + self.kinds.append("stamp") + self.stamp_values = dict(stmt.compile().params) + return _FakeResult(rowcount=self._stamp_rowcount) + raise AssertionError(f"unexpected statement against {table!r}: {stmt}") + + +def _pf(path: str, content: str) -> ParsedFile: + return ParsedFile(path=path, lang="python", size=len(content.encode()), content=content) + + +def _item( + path: str, content: str, *, symbols: bool = True, edges: bool = False +) -> tuple[ParsedFile, FileExtraction]: + symbol = ExtractedSymbol("f", "function", 1, 2) + return ( + _pf(path, content), + FileExtraction( + symbols=[symbol] if symbols else [], + edges=[ExtractedEdge(kind="call", target="t", line=2, enclosing=symbol)] + if edges + else [], + ), + ) + + +def _key(path: str, content: str) -> tuple[str, str]: + return (path, content_sha(content)) + + +def _index(conn: _FakeConn, items: Any, **kwargs: Any) -> IndexCounts: + return index_repo( + conn, + name="acme/widgets", + branch="main", + is_default=True, + head_sha="sha_new", + items=items, + **kwargs, + ) + + +# --- The gate: closed unless the stored version equals the current one -------- + + +@pytest.mark.unit +@pytest.mark.parametrize( + "baseline", + [(None, None), ("sha_old", None), ("sha_old", INDEX_SEMANTICS_VERSION - 1)], + ids=["never-indexed", "null-version", "older-version"], +) +def test_version_mismatch_never_issues_the_projection_reads( + baseline: tuple[str | None, int | None], +) -> None: + """T1: a NULL or stale stored version means full path for every file, and the + two projection reads (and the provenance gate) are never issued at all.""" + conn = _FakeConn(baseline=baseline, carried={_key("a.py", "x = 1\n")}) + counts = _index(conn, [_item("a.py", "x = 1\n")]) + + assert "read-carried" not in conn.kinds + assert "read-present" not in conn.kinds + assert "provenance-gate" not in conn.kinds + assert "file-upsert" in conn.kinds + assert counts == IndexCounts(files=1, symbols=1, swept=0, edges=0) + + +@pytest.mark.unit +def test_delta_on_all_unchanged_writes_nothing() -> None: + """T2 (AC1): every parsed file already carried by this branch at this content + means zero files/symbols/reference_edges statements -- but the sweep still + runs against the FULL seen-set, and the stamp is still last.""" + items = [_item("a.py", "x = 1\n"), _item("b.py", "y = 2\n")] + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("a.py", "x = 1\n"), _key("b.py", "y = 2\n")}, + present={_key("a.py", "x = 1\n"), _key("b.py", "y = 2\n")}, + ) + counts = _index(conn, items) + + assert conn.kinds == [ + "repos-insert", + "repo-branches-insert", + "read-carried", + "read-present", + "sweep-update", + "sweep-delete", + "stamp", + ] + # files still counts files SEEN this run (the seen-set size the sweep uses), + # so the sweep and its empty-seen-set guard need no delta awareness. + assert counts == IndexCounts(files=2, symbols=0, swept=0, edges=0) + + +@pytest.mark.unit +def test_all_unchanged_run_calls_no_chunk_writer() -> None: + """T2 (AC1), chunk half: an unchanged file's chunk rows are never rewritten.""" + calls: list[str] = [] + items = [_item("a.py", "x = 1\n")] + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("a.py", "x = 1\n")}, + present={_key("a.py", "x = 1\n")}, + ) + _index( + conn, + items, + chunk_writer=lambda _c, _r, _f, pf: calls.append(pf.path), + ) + assert calls == [] + + +@pytest.mark.unit +def test_changed_content_at_a_known_path_takes_the_full_path() -> None: + """A path whose content moved is NOT in ``carried`` under its new sha, so it + takes the full write path -- the (path, content_sha) keying, not path alone.""" + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("a.py", "x = 1\n")}, + present={_key("a.py", "x = 1\n")}, + ) + counts = _index(conn, [_item("a.py", "x = 999\n", edges=True)]) + + assert conn.kinds.count("file-upsert") == 1 + assert conn.kinds.count("symbols-delete") == 1 + assert conn.kinds.count("edges-delete") == 1 + assert counts == IndexCounts(files=1, symbols=1, swept=0, edges=1) + + +# --- Transaction shape and the untouched guards ------------------------------ + + +@pytest.mark.unit +def test_transaction_shape_is_pinned() -> None: + """T6: the epic's non-negotiable rule. repos first, repo_branches second, the + projection reads third, the CAS stamp LAST -- whatever the classification.""" + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("a.py", "x = 1\n")}, + present={_key("a.py", "x = 1\n"), _key("b.py", "y = 2\n")}, + ) + _index(conn, [_item("a.py", "x = 1\n"), _item("c.py", "z = 3\n")]) + + assert conn.kinds[0] == "repos-insert" + assert conn.kinds[1] == "repo-branches-insert" + assert conn.kinds[2] == "read-carried" + assert conn.kinds[3] == "read-present" + assert conn.kinds[-1] == "stamp" + + +@pytest.mark.unit +def test_empty_items_still_skips_the_sweep_with_delta_on() -> None: + """T8: the empty-seen-set guard is untouched by the delta path.""" + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("a.py", "x = 1\n")}, + present={_key("a.py", "x = 1\n")}, + ) + counts = _index(conn, []) + + assert "sweep-update" not in conn.kinds + assert "sweep-delete" not in conn.kinds + assert counts == IndexCounts(files=0, symbols=0, swept=0, edges=0) + + +@pytest.mark.unit +def test_zero_parse_run_does_not_advance_the_semantics_version() -> None: + """The §2.3 base case, at the statement level: an empty seen-set stamps the + commit but leaves ``index_semantics_version`` at the statement-2 baseline.""" + conn = _FakeConn(baseline=("sha_old", INDEX_SEMANTICS_VERSION - 1)) + _index(conn, []) + + assert conn.stamp_values["last_indexed_commit"] == "sha_new" + assert conn.stamp_values["index_semantics_version"] == INDEX_SEMANTICS_VERSION - 1 + + +@pytest.mark.unit +def test_non_empty_run_advances_the_semantics_version() -> None: + """The complement: a run that wrote something DOES claim the current version.""" + conn = _FakeConn(baseline=("sha_old", INDEX_SEMANTICS_VERSION - 1)) + _index(conn, [_item("a.py", "x = 1\n")]) + + assert conn.stamp_values["last_indexed_commit"] == "sha_new" + assert conn.stamp_values["index_semantics_version"] == INDEX_SEMANTICS_VERSION From e6ce969a44319f32d97a32b0c47a587bcbfcc3f7 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:14:22 -0700 Subject: [PATCH 3/9] store: batched membership-only union, provenance gate, and chunk write --- indexer/store.py | 127 ++++++++++++++++++++++++++++++++- tests/unit/test_store_delta.py | 123 +++++++++++++++++++++++++++++++ 2 files changed, 249 insertions(+), 1 deletion(-) diff --git a/indexer/store.py b/indexer/store.py index d164894..11e04e3 100644 --- a/indexer/store.py +++ b/indexer/store.py @@ -176,6 +176,12 @@ def index_repo( * **unchanged** (the pair is already on a row carrying this branch): no statement at all. + * **membership-only** (the pair is stored for this repo but on a row this + branch does not carry, AND statement 4 proves every branch of this repo + is at the current semantics version): no symbol/edge work; the whole + class is unioned in by ONE batched ``UPDATE ... RETURNING`` after the + loop, which also supplies the ``file_id`` for its ``chunk_writer`` call + (see :func:`_union_membership`). * **changed/new** (everything else): an array-union upsert on ``uq_files_repo_path_sha`` -- a file whose content already exists under another branch gets THIS branch unioned into its ``branches`` array (one @@ -220,6 +226,10 @@ def index_repo( COMPLETED run, and that run ran at ``baseline_version == INDEX_SEMANTICS_VERSION``. In it the row was either written full-path (so it is current), or skipped as unchanged (current, by induction). + * ... or acquired membership-only, which + :func:`_repo_is_wholly_at_current_version` only permits when every branch + of the repo is at the current version (so every surviving row of the repo + was last written at it). * Base case: the first run after ANY version transition has ``baseline_version != INDEX_SEMANTICS_VERSION``, so it is full-path for every parsed file. A zero-parse run cannot manufacture a spurious base @@ -278,13 +288,22 @@ def index_repo( ) baseline_commit, baseline_version = conn.execute(branch_stmt).one() - # Statements 3a/3b, issued only behind the delta gate. Everything the + # Statements 3a/3b/4, issued only behind the delta gate. Everything the # classification below needs is now in hand; nothing else is read. delta_on = baseline_version == INDEX_SEMANTICS_VERSION carried: set[tuple[str, str]] = set() present: set[tuple[str, str]] = set() + membership_ok = False if delta_on: carried, present = read_repo_content_shas(conn, repo_id=repo_id, branch=branch) + membership_ok = _repo_is_wholly_at_current_version(conn, repo_id=repo_id) + + # (pf, content_sha) for each membership-only file, held until the batched + # UPDATE below can hand back their file ids. A bounded exception to the + # "items stream through the transaction" rule: membership-only is the + # rare class (a branch ACQUIRING content another branch already stored), + # not the steady state, and chunk_writer's seam takes the ParsedFile. + membership: list[tuple[ParsedFile, str]] = [] for pf, ex in items: sha = content_sha(pf.content) @@ -301,6 +320,14 @@ def index_repo( # delete-reinsert, no chunk_writer call. continue + if membership_ok and (pf.path, sha) in present: + # Membership-only: the row exists (written by another branch) but + # does not carry this branch yet. Statement 4 has proven every + # branch of this repo is at the current semantics version, so its + # symbols/edges are current and only the array union is owed. + membership.append((pf, sha)) + continue + file_stmt = ( pg_insert(File) .values( @@ -380,6 +407,18 @@ def index_repo( if chunk_writer is not None: chunk_writer(conn, repo_id, file_id, pf) + # ONE statement for the whole membership-only class, skipped entirely + # when that class is empty (rather than issued as a no-op) so the + # statement inventory stays stable and greppable. + if membership: + _union_membership( + conn, + repo_id=repo_id, + branch=branch, + membership=membership, + chunk_writer=chunk_writer, + ) + # Timed into indexer.job's ambient per-branch PhaseTimer, if one is # installed -- a no-op otherwise, so a direct index_repo call (tests, # scripts) is unaffected. Deliberately NOT a return value: IndexCounts is @@ -412,6 +451,92 @@ def index_repo( return IndexCounts(files=file_count, symbols=symbol_count, swept=swept, edges=edge_count) +def _repo_is_wholly_at_current_version(conn: Connection, *, repo_id: int) -> bool: + """Statement 4: is EVERY ``repo_branches`` row for this repo at the current version? + + The provenance gate the membership-only class depends on, and the hole + ``(path, content_sha)`` alone does not close. Counter-example it exists for: + branch ``b`` is stamped at the current version (delta on); sibling branch + ``a`` was written at an OLDER version and has not re-indexed since. ``b``'s + HEAD moves and acquires a file whose exact ``(path, content)`` already exists + as ``a``'s stale-version row. Taking the membership-only path would skip the + symbol/edge rewrite, so ``b`` would serve old-extractor symbols under a + current-version stamp -- silently, and exactly the failure + ``INDEX_SEMANTICS_VERSION`` exists to prevent. + + Given this gate, every surviving ``files`` row of the repo was last written + at the current version: every row carries at least one branch (both sweep + sites delete rows at ``cardinality(branches) = 0``), and every branch string + on a ``branches`` array has a ``repo_branches`` row (``index_repo`` writes + statement 2 before any file row for that branch, and + ``reconcile_retired_branches`` deletes both in one transaction). Both + directions are load-bearing and both are pinned by tests. + """ + return bool( + conn.execute( + text( + "SELECT NOT EXISTS (SELECT 1 FROM repo_branches " + "WHERE repo_id = :repo_id " + "AND index_semantics_version IS DISTINCT FROM :version)" + ), + {"repo_id": repo_id, "version": INDEX_SEMANTICS_VERSION}, + ).scalar_one() + ) + + +def _union_membership( + conn: Connection, + *, + repo_id: int, + branch: str, + membership: list[tuple[ParsedFile, str]], + chunk_writer: ChunkWriter | None, +) -> None: + """Union ``branch`` into every membership-only row in ONE statement, then write their chunks. + + ``array_agg(DISTINCT ...)`` rather than ``||`` alone so the stored array + stays sorted-distinct, matching ``index_repo``'s per-file upsert idiom -- + existing assertions compare ``branches`` by value. + + ``RETURNING id, path, content_sha`` supplies each row's ``file_id`` without a + second lookup, which is what makes the ``chunk_writer`` call below possible. + **Membership-only DOES write chunks** even though it writes no symbols or + edges: the acquired row may legitimately have zero chunk rows (the branch + that first wrote it ran semantic-off, or its precompute failed), and skipping + the write would make that gap permanent for the acquiring branch where the + full path would have filled it. The vectors are already in hand -- ``job.py`` + embeds every file the advisory read did not call unchanged. + """ + paths = [pf.path for pf, _sha in membership] + shas = [sha for _pf, sha in membership] + rows = conn.execute( + text( + "UPDATE files SET branches = (SELECT array_agg(DISTINCT e) FROM " + "unnest(files.branches || CAST(:branch_arr AS text[])) e) " + "WHERE repo_id = :repo_id " + "AND EXISTS (SELECT 1 FROM unnest(CAST(:paths AS text[]), CAST(:shas AS text[])) " + "AS t(p, s) WHERE t.p = files.path AND t.s = files.content_sha) " + "RETURNING id, path, content_sha" + ), + {"repo_id": repo_id, "branch_arr": [branch], "paths": paths, "shas": shas}, + ).all() + + if chunk_writer is None: + return + file_ids = {(row.path, row.content_sha): row.id for row in rows} + for pf, sha in membership: + file_id = file_ids.get((pf.path, sha)) + if file_id is None: + # Unreachable while the single-writer invariant holds: the row was in + # statement 3b's projection moments ago, inside this transaction. + logger.warning( + "membership-only row for %s vanished before its union; skipping its chunk write", + pf.path, + ) + continue + chunk_writer(conn, repo_id, file_id, pf) + + def _sweep_membership( conn: Connection, *, diff --git a/tests/unit/test_store_delta.py b/tests/unit/test_store_delta.py index 14bacdb..fec24f6 100644 --- a/tests/unit/test_store_delta.py +++ b/tests/unit/test_store_delta.py @@ -251,6 +251,7 @@ def test_delta_on_all_unchanged_writes_nothing() -> None: "repo-branches-insert", "read-carried", "read-present", + "provenance-gate", "sweep-update", "sweep-delete", "stamp", @@ -295,6 +296,127 @@ def test_changed_content_at_a_known_path_takes_the_full_path() -> None: assert counts == IndexCounts(files=1, symbols=1, swept=0, edges=1) +# --- Membership-only: one batched union, a chunk write, no symbol/edge work --- + + +@pytest.mark.unit +def test_membership_only_issues_one_batched_union_and_no_symbol_work() -> None: + """T3 (AC3): a file stored for this repo but not carried by this branch takes + the membership path -- ONE batched UPDATE for the whole class, one + chunk_writer call per file, and no symbols/reference_edges statements.""" + calls: list[tuple[int, str]] = [] + items = [_item("a.py", "x = 1\n"), _item("b.py", "y = 2\n")] + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried=set(), + present={_key("a.py", "x = 1\n"), _key("b.py", "y = 2\n")}, + ) + counts = _index(conn, items, chunk_writer=lambda _c, _r, fid, pf: calls.append((fid, pf.path))) + + assert conn.kinds.count("membership-union") == 1 + assert "file-upsert" not in conn.kinds + assert "symbols-delete" not in conn.kinds + assert "edges-delete" not in conn.kinds + # The file_id each chunk write used came from the UPDATE's RETURNING, not a + # second lookup. + assert calls == [(1, "a.py"), (2, "b.py")] + assert conn.membership_params["paths"] == ["a.py", "b.py"] + assert conn.membership_params["branch_arr"] == ["main"] + # symbols/edges legitimately fall to zero: no rows were inserted. + assert counts == IndexCounts(files=2, symbols=0, swept=0, edges=0) + + +@pytest.mark.unit +def test_membership_only_is_refused_when_a_sibling_branch_is_stale() -> None: + """T4 (AC6): statement 4 false -- some branch of this repo sits at another + semantics version -- forces every would-be membership file down the full + path, so it can never inherit stale-version symbols/edges.""" + items = [_item("a.py", "x = 1\n")] + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried=set(), + present={_key("a.py", "x = 1\n")}, + provenance=False, + ) + counts = _index(conn, items) + + assert "provenance-gate" in conn.kinds + assert "membership-union" not in conn.kinds + assert conn.kinds.count("file-upsert") == 1 + assert conn.kinds.count("symbols-delete") == 1 + assert counts == IndexCounts(files=1, symbols=1, swept=0, edges=0) + + +@pytest.mark.unit +def test_mixed_classification_statement_inventory() -> None: + """T5 (AC2): 1 unchanged, 1 membership-only, 1 changed, 1 new -> the exact + inventory, with the batched union issued once, AFTER the per-file loop.""" + unchanged = _item("keep.py", "k = 1\n") + member = _item("shared.py", "s = 1\n") + changed = _item("moved.py", "m = 2\n") + added = _item("new.py", "n = 1\n") + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("keep.py", "k = 1\n"), _key("moved.py", "m = 1\n")}, + present={ + _key("keep.py", "k = 1\n"), + _key("moved.py", "m = 1\n"), + _key("shared.py", "s = 1\n"), + }, + ) + counts = _index(conn, [unchanged, member, changed, added]) + + assert conn.kinds == [ + "repos-insert", + "repo-branches-insert", + "read-carried", + "read-present", + "provenance-gate", + # moved.py -- changed content at a known path + "file-upsert", + "symbols-delete", + "symbols-insert", + "edges-delete", + # new.py -- never seen + "file-upsert", + "symbols-delete", + "symbols-insert", + "edges-delete", + # shared.py -- the whole membership class, batched, after the loop + "membership-union", + "sweep-update", + "sweep-delete", + "stamp", + ] + assert conn.membership_params["paths"] == ["shared.py"] + assert counts == IndexCounts(files=4, symbols=2, swept=0, edges=0) + + +@pytest.mark.unit +def test_empty_membership_class_issues_no_union_statement() -> None: + """T5, the stability half: an empty membership set is SKIPPED, never issued + as a no-op UPDATE -- which is what keeps the inventories above stable.""" + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("a.py", "x = 1\n")}, + present={_key("a.py", "x = 1\n")}, + ) + _index(conn, [_item("a.py", "x = 1\n")]) + assert "membership-union" not in conn.kinds + + +@pytest.mark.unit +def test_membership_without_a_chunk_writer_issues_only_the_union() -> None: + """The semantic-off path: the union still runs, nothing else does.""" + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried=set(), + present={_key("a.py", "x = 1\n")}, + ) + _index(conn, [_item("a.py", "x = 1\n")], chunk_writer=None) + assert conn.kinds.count("membership-union") == 1 + + # --- Transaction shape and the untouched guards ------------------------------ @@ -313,6 +435,7 @@ def test_transaction_shape_is_pinned() -> None: assert conn.kinds[1] == "repo-branches-insert" assert conn.kinds[2] == "read-carried" assert conn.kinds[3] == "read-present" + assert conn.kinds[4] == "provenance-gate" assert conn.kinds[-1] == "stamp" From e80c273b302871f184252aff7e203a87a06d563b Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:15:21 -0700 Subject: [PATCH 4/9] store: log the delta write set --- indexer/store.py | 32 +++++++++++++++++++++++ tests/unit/test_store_delta.py | 48 ++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+) diff --git a/indexer/store.py b/indexer/store.py index 11e04e3..af6f308 100644 --- a/indexer/store.py +++ b/indexer/store.py @@ -248,10 +248,21 @@ class is unioned in by ONE batched ``UPDATE ... RETURNING`` after the ``commit:`` filter resolves from ``repo_branches.last_indexed_commit``, and the column is documented write-only and ambiguous under dedup in ``app/db/models.py``). This makes it staler; it makes nothing wrong. + + The breakdown is reported on one INFO line per call, immediately before the + sweep, in both the gate-open and gate-closed cases:: + + acme/widgets@main: delta write set 412/30214 files (unchanged=29790 membership=12, + semantics gate open) + + ``IndexCounts`` is unchanged: ``files`` still counts files SEEN this run, and + ``symbols``/``edges`` still count rows actually inserted -- so they + legitimately read ``0`` on an all-unchanged run. That is the correct signal. """ file_count = 0 symbol_count = 0 edge_count = 0 + unchanged_count = 0 seen_paths: list[str] = [] seen_shas: list[str] = [] @@ -318,6 +329,7 @@ class is unioned in by ONE batched ``UPDATE ... RETURNING`` after the # Unchanged: this exact content is already stored on a row this # branch already carries. No file upsert, no symbol/edge # delete-reinsert, no chunk_writer call. + unchanged_count += 1 continue if membership_ok and (pf.path, sha) in present: @@ -419,6 +431,26 @@ class is unioned in by ONE batched ``UPDATE ... RETURNING`` after the chunk_writer=chunk_writer, ) + # One INFO line per index_repo call, immediately before the sweep, in + # BOTH the gate-open and gate-closed cases -- one format string, no + # conditional fields, so the line is always present and always greppable. + # The reason tail is the only part that varies. IndexCounts is + # deliberately NOT extended to carry this: it is a frozen dataclass + # compared by value in existing assertions, and `files` keeps meaning + # "files seen this run" (the seen-set size). + logger.info( + "%s@%s: delta write set %d/%d files (unchanged=%d membership=%d, %s)", + name, + branch, + file_count - unchanged_count - len(membership), + file_count, + unchanged_count, + len(membership), + "semantics gate open" + if delta_on + else f"semantics gate closed: stored v{baseline_version} != v{INDEX_SEMANTICS_VERSION}", + ) + # Timed into indexer.job's ambient per-branch PhaseTimer, if one is # installed -- a no-op otherwise, so a direct index_repo call (tests, # scripts) is unaffected. Deliberately NOT a return value: IndexCounts is diff --git a/tests/unit/test_store_delta.py b/tests/unit/test_store_delta.py index fec24f6..4f90ff0 100644 --- a/tests/unit/test_store_delta.py +++ b/tests/unit/test_store_delta.py @@ -16,6 +16,7 @@ from __future__ import annotations import contextlib +import logging from typing import Any, NamedTuple import pytest @@ -417,6 +418,53 @@ def test_membership_without_a_chunk_writer_issues_only_the_union() -> None: assert conn.kinds.count("membership-union") == 1 +# --- The `delta write set` line: always emitted, one shape ------------------- + + +@pytest.mark.unit +def test_delta_write_set_line_reports_the_breakdown_with_the_gate_open( + caplog: pytest.LogCaptureFixture, +) -> None: + """T7: the breakdown rides its OWN line from indexer.store -- IndexCounts is + unchanged, so this is where unchanged/membership become visible.""" + conn = _FakeConn( + baseline=("sha_old", INDEX_SEMANTICS_VERSION), + carried={_key("keep.py", "k = 1\n")}, + present={_key("keep.py", "k = 1\n"), _key("shared.py", "s = 1\n")}, + ) + with caplog.at_level(logging.INFO, logger="indexer.store"): + _index( + conn, + [ + _item("keep.py", "k = 1\n"), + _item("shared.py", "s = 1\n"), + _item("new.py", "n = 1\n"), + ], + ) + + assert ( + "acme/widgets@main: delta write set 1/3 files " + "(unchanged=1 membership=1, semantics gate open)" + ) in caplog.text + + +@pytest.mark.unit +def test_delta_write_set_line_is_present_with_the_gate_closed( + caplog: pytest.LogCaptureFixture, +) -> None: + """T7, the other half: the line never disappears -- a closed gate reports the + full-path reason instead, so no grep an operator writes breaks.""" + conn = _FakeConn(baseline=("sha_old", INDEX_SEMANTICS_VERSION - 1)) + with caplog.at_level(logging.INFO, logger="indexer.store"): + _index(conn, [_item("a.py", "x = 1\n"), _item("b.py", "y = 2\n")]) + + assert ( + f"acme/widgets@main: delta write set 2/2 files (unchanged=0 membership=0, " + f"semantics gate closed: stored v{INDEX_SEMANTICS_VERSION - 1} " + f"!= v{INDEX_SEMANTICS_VERSION})" + ) in caplog.text + + # --- Transaction shape and the untouched guards ------------------------------ From a6efa6fb35f63d9225e5e86fa5c01a8fb20db807 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:51:36 -0700 Subject: [PATCH 5/9] job: advisory shas_fn seam, covered-set guard, degraded-semantics summary Thread an injected shas_fn (default indexer.store.read_indexed_shas) through run -> _index_one -> _index_one_inner -> _index_one_branch. Called on a short-lived engine.connect() before embedding, and only when the branch's stored index_semantics_version already matches INDEX_SEMANTICS_VERSION (the same gate index_repo applies authoritatively). Narrows _precompute_chunk_writer's file list to every file the advisory read did not classify as unchanged. _precompute_chunk_writer's chunk_writer closure now guards against an uncovered path (warn-and-skip, never delete an uncovered file's chunks -- defence-in-depth for a path the single-writer invariant says is unreachable). BranchOutcome gains semantic_degraded; run() aggregates every degraded branch into one run-completion WARNING, since chunk coverage no longer self-heals on the next run under delta indexing. --- indexer/job.py | 147 +++++++++++++++++- tests/unit/test_job.py | 344 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 483 insertions(+), 8 deletions(-) diff --git a/indexer/job.py b/indexer/job.py index 3a0f1d8..11a5fe4 100644 --- a/indexer/job.py +++ b/indexer/job.py @@ -79,6 +79,24 @@ loop as symbols. Flag-off: no chunking, no embedder, no import of ``app.embed``'s lazy ``databricks-sdk`` dependency. +File-level delta indexing (issue #104) skips re-embedding a file this branch +already carries unchanged, at ``(path, content_sha)``. Before building the +embed list, this module calls the injected ``shas_fn`` (default +:func:`indexer.store.read_indexed_shas`) -- but ONLY when this branch's stored +``index_semantics_version`` already matches +:data:`app.db.models.INDEX_SEMANTICS_VERSION`, the same gate +``indexer.store.index_repo`` applies authoritatively inside its own +transaction -- on a short-lived connection that closes before the embedder is +called. ``index_repo``'s per-file classification (unchanged / membership-only / +changed-new) is the one that actually decides what gets written; this module's +copy is advisory only, used to decide what needs a vector. Both reads can only +diverge if a second writer touched this repo between them, which the +single-writer-per-repo invariant above forbids. A branch whose semantic +precompute fails keeps whatever chunk coverage it already had -- see +:func:`_precompute_chunk_writer` and ``indexer/store.py``'s module docstring +for why that is no longer self-healing on the next run under delta indexing, +and the aggregate run-completion WARNING this module emits for it. + Every INDEXED branch also emits one ``phase timing`` line accounting for its whole wall clock -- resolve / download / extract / parse / embed / db / sweep, plus an ``other`` residual so no time is silently unattributed. The fields are fixed and @@ -130,15 +148,18 @@ resolve_branch_head, resolve_ref, ) +from indexer.hashing import content_sha from indexer.languages import Chunk, FileExtraction, IndexCounts, ParsedFile from indexer.parse import iter_chunks, iter_source_files from indexer.repo_config import RepoConfig, effective_workers, load_config, normalize_repo from indexer.resolve import MAX_REPOS, RepoEntry, resolve_repos from indexer.store import ( ChunkWriter, + ContentShaSets, ReconcileCounts, StaleIndexError, index_repo, + read_indexed_shas, reconcile_removed_repos, reconcile_retired_branches, ) @@ -166,11 +187,20 @@ class BranchOutcome: ``StaleIndexError`` maps to ``"conflict"`` and any other exception to ``"failed"``, caught INSIDE the per-branch loop so one branch's failure never stops its repo's other branches from being attempted. + + ``semantic_degraded`` is ``True`` only for an ``"indexed"`` outcome whose + chunk precompute raised (a chunk-cap breach or an embedder failure) -- + never for semantic-off, and never for a build-time embedder misconfiguration + (that degrades ``embed_fn`` to ``None`` before any branch starts, so no + per-branch precompute is ever attempted for it -- see ``_index_one_branch``). + ``run()`` aggregates every branch with this flag set into one + run-completion WARNING. """ branch: str status: Literal["indexed", "skipped", "conflict", "failed"] counts: IndexCounts | None = None + semantic_degraded: bool = False @dataclass(frozen=True) @@ -276,6 +306,7 @@ def run( max_repos: int = MAX_REPOS, reconcile_retired_fn: Callable[..., ReconcileCounts] = reconcile_retired_branches, reconcile_removed_fn: Callable[..., list[str]] = reconcile_removed_repos, + shas_fn: Callable[..., ContentShaSets] = read_indexed_shas, ) -> int: """Index every configured repo and return a process exit code (0 = all ok). @@ -414,6 +445,7 @@ def run( # -- and how a shrunken disk becomes visible before it becomes an outage. ok = skipped = conflicts = failures = 0 repo_outcomes: list[RepoOutcome] = [] + degraded_branches: list[str] = [] reconciliation_attempted = False reconciliation_failed = False reconcile_skip_reason = "" @@ -451,6 +483,7 @@ def run( cfg=cfg, embed_fn=embed_fn, stamps=stamps, + shas_fn=shas_fn, ): entry for entry in entries } @@ -479,6 +512,8 @@ def run( else: ok += 1 assert outcome.counts is not None + if outcome.semantic_degraded: + degraded_branches.append(f"{entry.name}@{outcome.branch}") logger.info( "indexed %s@%s: files=%d symbols=%d edges=%d swept=%d", entry.name, @@ -548,6 +583,28 @@ def run( # the WARNING logged at the conflict site is the record. It is NOT that the # work was redundant. + # One aggregate, greppable WARNING for every branch that finished "indexed" + # with degraded semantic coverage this run (its precompute failed -- a + # chunk-cap breach or an embedder outage -- so its core index is current but + # its chunks are not). Under file-level delta indexing this gap is NOT + # self-healing on the next run (see indexer/store.py's module docstring and + # docs/runbooks/indexing-parallelism.md §4): only a changed file re-embeds, + # so a branch that never changes again would carry stale/missing chunks + # forever unless an operator clears its semantics stamp. The per-branch + # WARNING already logged at the precompute site is easy to miss in a large + # run's log; this line exists so the condition is greppable after the fact. + # Deliberately does NOT fail the run -- see the per-branch warning site for + # why this is an additive-layer failure, not a core-index one. + if degraded_branches: + logger.warning( + "%d branch(es) finished with degraded semantic coverage this run (chunk precompute " + "failed; core index is current, chunks are not, and delta indexing will NOT catch " + "them up on their own -- clear their repo_branches.index_semantics_version stamp to " + "force a full re-embed, see docs/runbooks/indexing-parallelism.md §4): %s", + len(degraded_branches), + ", ".join(sorted(degraded_branches)), + ) + # Exactly one reconciliation summary line, always after "indexing complete". # The failure/withheld path already logged its own ERROR incident line # inside _reconcile -- this branch must not double-log it. @@ -803,8 +860,21 @@ def _precompute_chunk_writer( embedder itself. Raises ``ValueError`` if the repo's total chunk count exceeds ``max_chunks_per_repo`` (the documented hard ceiling, not a streaming bound -- see ``app.config.semantic_max_chunks_per_repo``). + + ``files`` is whatever the caller decided needs vectors -- under file-level + delta indexing (``_index_one_branch``) that is every file the advisory + ``shas_fn`` read did NOT classify as unchanged, not necessarily every parsed + file in the branch. The closure below therefore closes over ``covered = + set(per_file)`` -- every path THIS call embedded, including zero-chunk files + -- and refuses to write chunks for any other path. This is defence-in-depth + for a path the single-writer-per-repo invariant says is unreachable: ` + `write_chunks`` deletes a file's chunk rows before inserting, so calling it + for a path this precompute never embedded would silently delete that file's + chunks rather than merely leave them stale. See ``indexer/store.py``'s + ``_union_membership`` for the authoritative-side analogue of this guard. """ per_file: dict[str, list[Chunk]] = {pf.path: list(iter_chunks(pf)) for pf in files} + covered = set(per_file) total = sum(len(chunks) for chunks in per_file.values()) if total > max_chunks_per_repo: raise ValueError( @@ -834,6 +904,17 @@ def _precompute_chunk_writer( i += len(chunks) def chunk_writer(conn: Any, repo_id: int, file_id: int, pf: ParsedFile) -> None: + if pf.path not in covered: + # Unreachable while the single-writer invariant holds (see the + # docstring): index_repo only ever calls chunk_writer for a file this + # same precompute either embedded or classified membership-only (and + # index_repo's own _union_membership guards that case separately). + # Warn-and-skip rather than raise, matching this module's established + # additive-layer posture for the semantic path. + logger.warning( + "no precomputed chunks for %s; leaving its chunk rows untouched", pf.path + ) + return write_chunks(conn, file_id=file_id, chunks=by_path.get(pf.path, [])) return chunk_writer @@ -848,6 +929,7 @@ def _index_one( cfg: Settings, embed_fn: EmbedFn | None, stamps: dict[tuple[str, str], tuple[str | None, int | None]], + shas_fn: Callable[..., ContentShaSets], ) -> RepoOutcome: """Run the full fetch -> parse -> symbols -> store pipeline for every branch of one repo. @@ -873,6 +955,7 @@ def _index_one( cfg=cfg, embed_fn=embed_fn, stamps=stamps, + shas_fn=shas_fn, ) finally: _repo_ctx.reset(token) @@ -888,6 +971,7 @@ def _index_one_inner( cfg: Settings, embed_fn: EmbedFn | None, stamps: dict[tuple[str, str], tuple[str | None, int | None]], + shas_fn: Callable[..., ContentShaSets], ) -> RepoOutcome: """The body of :func:`_index_one`, run with the repo log context already set. @@ -948,6 +1032,7 @@ def _index_one_inner( stamps=stamps, started=started, max_chunks_per_repo=max_chunks_per_repo, + shas_fn=shas_fn, ) for branch in resolution.branches ] @@ -1015,6 +1100,7 @@ def _index_one_branch( stamps: dict[tuple[str, str], tuple[str | None, int | None]], started: float, max_chunks_per_repo: int, + shas_fn: Callable[..., ContentShaSets], ) -> BranchOutcome: """Fetch, parse, and store ONE branch. Never raises -- every failure is classified. @@ -1023,6 +1109,22 @@ def _index_one_branch( A stored ``None`` version means the provenance of the stored index is unknown, so the branch is always re-indexed. + ``shas_fn`` is the ADVISORY copy of :func:`indexer.store.read_repo_content_shas` + (see that module's docstring for the authoritative one). It is called ONLY + when this branch's stored ``index_semantics_version`` already matches + :data:`app.db.models.INDEX_SEMANTICS_VERSION` -- the same gate ``index_repo`` + applies inside its transaction, from data already in hand here -- and ONLY on + a separate, short-lived ``engine.connect()`` that closes before embedding + starts, never on a connection held across the embedder's network I/O. Its + result narrows the file list handed to ``_precompute_chunk_writer`` to every + file NOT already carried by this branch (changed/new *and* membership-only -- + ``index_repo`` may reclassify a membership-only file as changed/new inside its + own transaction if the provenance gate fails there, so it must already have a + vector to attach). The two reads can only disagree if a second writer touched + this repo between them, which the single-writer-per-repo invariant (see the + module docstring) forbids; see ``indexer/store.py``'s ``_union_membership`` + for the defence-in-depth guard on the authoritative side. + ``max_chunks_per_repo`` is the caller's (``_index_one_inner``'s) already-resolved effective cap -- this repo's ``semantic_max_chunks_per_repo`` override if one matched, else ``cfg.semantic_max_chunks_per_repo``. Taken as a parameter rather @@ -1083,6 +1185,7 @@ def _index_one_branch( timer.add("extract", timer.clock() - t0) chunk_writer: ChunkWriter | None = None + precompute_failed = False if cfg.semantic_enabled and embed_fn is not None: # Chunking/embedding needs the full file list up front -- unlike # the lazy items generator below, it cannot stream through @@ -1100,17 +1203,42 @@ def _index_one_branch( # In a `finally`, unlike every other phase wrap: the degrade path # below still burned this time (a downed embedder can burn a lot - # of it before it gives up) and must still be reported. + # of it before it gives up) and must still be reported. The + # advisory shas_fn read is charged here too, deliberately NOT as + # its own timed phase: it exists solely to decide what this block + # embeds, and #103's `phase timing` line is pinned exhaustive + # (nine fixed fields, tests/unit/test_job.py) -- adding a tenth + # field is out of this change's scope. t0 = timer.clock() try: - chunk_writer = _precompute_chunk_writer(files, embed_fn, max_chunks_per_repo) + files_to_embed = files + if stamps.get((name.casefold(), branch), (None, None))[1] == ( + INDEX_SEMANTICS_VERSION + ): + # Delta gate open (same test index_repo will apply + # authoritatively, from data already in hand): narrow to + # every file this branch does not already carry. A short- + # lived connection, closed before embedding starts -- + # never held across the embedder's network I/O. + with engine.connect() as shas_conn: + carried, _present = shas_fn(shas_conn, name=name, branch=branch) + files_to_embed = [ + pf for pf in files if (pf.path, content_sha(pf.content)) not in carried + ] + chunk_writer = _precompute_chunk_writer( + files_to_embed, embed_fn, max_chunks_per_repo + ) except Exception: # The semantic layer is ADDITIVE: a chunk-ceiling breach, a downed embedder, - # or a dim/count mismatch must not cost this branch its core index. Letting - # it propagate would skip files/symbols AND the mark-and-sweep, silently - # leaving the branch stale -- worse than stale chunks. Chunks catch up on - # the next successful run; the failure is logged with a traceback, never - # swallowed silently. + # a dim/count mismatch, or a failure reading the advisory shas_fn projection + # must not cost this branch its core index. Letting it propagate would skip + # files/symbols AND the mark-and-sweep, silently leaving the branch stale -- + # worse than stale chunks. Under file-level delta indexing this is NOT + # self-healing the way it was before: only a changed file re-embeds, so a + # branch that never changes again carries this gap forever unless an + # operator clears its semantics stamp (see run()'s aggregate WARNING and + # docs/runbooks/indexing-parallelism.md §4). The failure is logged with a + # traceback here too, never swallowed silently. logger.warning( "semantic precompute failed for %s@%s; indexing core corpus without chunks", name, @@ -1118,6 +1246,7 @@ def _index_one_branch( exc_info=True, ) chunk_writer = None + precompute_failed = True finally: timer.add("embed", timer.clock() - t0) items = ((pf, extract_file(pf)) for pf in files) @@ -1198,7 +1327,9 @@ def _index_one_branch( ) except Exception: logger.warning("phase timing unavailable for %s@%s", name, branch, exc_info=True) - return BranchOutcome(branch=branch, status="indexed", counts=counts) + return BranchOutcome( + branch=branch, status="indexed", counts=counts, semantic_degraded=precompute_failed + ) except StaleIndexError as exc: # The repo_branches row for THIS branch changed under this worker, so # its whole transaction rolled back and THIS BRANCH IS NOT INDEXED. diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py index d4ca8dd..de21d19 100644 --- a/tests/unit/test_job.py +++ b/tests/unit/test_job.py @@ -32,6 +32,7 @@ from app.config import Settings from app.db.models import INDEX_SEMANTICS_VERSION +from indexer.hashing import content_sha from indexer.job import ( BranchOutcome, RepoOutcome, @@ -325,6 +326,18 @@ def _noop_removed_fn(conn: Any, *, desired_repos: Any) -> list[str]: return [] +def _noop_shas_fn(conn: Any, *, name: str, branch: str) -> tuple[set[Any], set[Any]]: + """Default ``shas_fn`` for ``_run()`` -- ``_FakeConn`` cannot answer the real + ``read_indexed_shas`` query (it routes only on ``"repo_branches" in str(stmt)``, + per its own docstring), so any test that reaches the advisory read without an + explicit override would call the REAL primitive against the fake engine and + fail. Returning two empty sets degrades to "everything is changed/new", + matching ``read_indexed_shas``' own behaviour for a never-indexed repo -- safe + in the correct direction and inert for every test that never exercises delta + embedding at all.""" + return set(), set() + + class _RecordingReconcile: """Fake ``reconcile_retired_fn``/``reconcile_removed_fn`` pair, explicitly opted into. @@ -373,6 +386,7 @@ def _run( engine: _FakeEngine | None = None, reconcile_retired_fn: Any = _noop_retired_fn, reconcile_removed_fn: Any = _noop_removed_fn, + shas_fn: Any = _noop_shas_fn, ) -> int: """Drive run() with a faked config read but a REAL resolve_repos. @@ -382,6 +396,11 @@ def _run( against ``_FakeEngine``/``_FakeConn`` (neither implements ``conn.begin()``). Tests that assert on reconciliation itself pass an explicit :class:`_RecordingReconcile`'s bound methods. + + ``shas_fn`` defaults to :func:`_noop_shas_fn` for the same reason: a + version-matching, sha-mismatching stamp with semantic indexing on would + otherwise reach the REAL ``read_indexed_shas`` against ``_FakeConn``, which + cannot answer it (see that fake's docstring). """ wc = _FakeWorkspaceClient("tok") engine = engine if engine is not None else _FakeEngine() @@ -402,6 +421,7 @@ def _run( config_loader=lambda _client, _path: config, reconcile_retired_fn=reconcile_retired_fn, reconcile_removed_fn=reconcile_removed_fn, + shas_fn=shas_fn, ) @@ -756,6 +776,326 @@ def test_unbuildable_embedder_does_not_abort_the_whole_run() -> None: assert idx.chunk_writer is None +# --- file-level delta indexing: the chunk_writer covered-set guard (#104) --- + + +@pytest.mark.unit +def test_chunk_writer_covered_guard_skips_an_uncovered_path( + caplog: pytest.LogCaptureFixture, +) -> None: + """T14: a path outside _precompute_chunk_writer's own file list -> WARNING, and + chunk_writer issues NO statement at all (specifically no DELETE FROM chunks, + which write_chunks always opens with -- see indexer/chunk_store.py). This is + the defence-in-depth guard for a path the single-writer-per-repo invariant + says index_repo can never actually pass it; unreachable in production, but + the alternative (silently deleting an uncovered file's chunk rows) is worse + than a loud skip. + """ + from indexer.job import _precompute_chunk_writer + + pf = ParsedFile(path="ghost.py", lang="python", size=10, content="x = 1\n") + chunk_writer = _precompute_chunk_writer([], lambda texts: [[0.0] for _ in texts], 100) + conn = _FakeChunkConn() + with caplog.at_level(logging.WARNING, logger="indexer.job"): + chunk_writer(conn, 1, 99, pf) + assert conn.calls == [] + assert any( + "no precomputed chunks for ghost.py" in r.getMessage() + for r in caplog.records + if r.name == "indexer.job" + ) + + +@pytest.mark.unit +def test_chunk_writer_covers_every_embedded_path_including_zero_chunk_files() -> None: + """The covered set is `set(per_file)`, not `set(by_path)` -- a file that embeds + to zero chunks (e.g. an empty file) is still COVERED, so its chunk_writer call + reaches write_chunks([]) (the delete-only, zero-row-insert shape) rather than + the warn-and-skip guard above.""" + from indexer.job import _precompute_chunk_writer + + empty_pf = ParsedFile(path="empty.py", lang="python", size=0, content="") + chunk_writer = _precompute_chunk_writer([empty_pf], lambda texts: [[0.0] for _ in texts], 100) + conn = _FakeChunkConn() + chunk_writer(conn, 1, 99, empty_pf) + # write_chunks always issues its DELETE even for zero chunks (see + # indexer/chunk_store.py) -- so a real (non-warning) statement was issued. + assert len(conn.calls) >= 1 + + +# --- file-level delta indexing: the advisory shas_fn seam (#104) ------------ +# job.py's copy of indexer.store.read_repo_content_shas is ADVISORY -- it only +# narrows what _precompute_chunk_writer embeds. index_repo's own read (behind +# index_fn, unexercised by _RecordingIndex) is the authoritative classification; +# these tests pin job.py's half of the contract only: when shas_fn is called, +# with what, and what that narrows the embedder's input to. _DEFAULT_FILES is +# {"main.py": b"def f():\n return 1\n", "README.md": b"# hi\n"} (both text, +# both chunked -- indexer.parse.iter_chunks has no language gate). + +_MAIN_SHA = content_sha("def f():\n return 1\n") +_README_SHA = content_sha("# hi\n") + + +class _RecordingShas: + """Fake ``shas_fn``: records every ``(name, branch)`` call and returns a + scripted ``(carried, present)`` pair (``present`` is unused by job.py, which + only classifies "unchanged" from ``carried`` -- the provenance gate that + would also need ``present`` is index_repo's alone).""" + + def __init__( + self, + *, + carried: set[tuple[str, str]] | None = None, + present: set[tuple[str, str]] | None = None, + ) -> None: + self.calls: list[tuple[str, str]] = [] + self._carried = carried or set() + self._present = present or set() + + def __call__(self, conn: Any, *, name: str, branch: str) -> tuple[set[Any], set[Any]]: + self.calls.append((name, branch)) + return set(self._carried), set(self._present) + + +@pytest.mark.unit +def test_shas_fn_called_once_before_embedding_when_version_matches_and_sha_differs() -> None: + """T9: the gate is (stored version == current) AND (stored sha != HEAD) -- the + latter is implied by reaching this code at all (an exact stamp match skips the + branch entirely before any of this runs, see the Step 4 tests above).""" + order: list[str] = [] + shas = _RecordingShas() + + def _shas_fn(conn: Any, *, name: str, branch: str) -> tuple[set[Any], set[Any]]: + order.append("shas") + return shas(conn, name=name, branch=branch) + + def _embed(texts: list[str]) -> list[list[float]]: + order.append("embed") + return [[0.0] for _ in texts] + + engine = _FakeEngine(stamps={("acme/widgets", "main"): ("sha_old", INDEX_SEMANTICS_VERSION)}) + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=_embed, + engine=engine, + shas_fn=_shas_fn, + ) + assert code == 0 + assert shas.calls == [("acme/widgets", "main")] + assert order == ["shas", "embed"] + + +@pytest.mark.unit +def test_shas_fn_not_called_when_the_semantics_version_is_stale() -> None: + """T10: version mismatch -> the delta gate is closed, so job.py never issues the + advisory read either, and the embedder receives every file's chunk text.""" + shas = _RecordingShas() + embed_calls: list[list[str]] = [] + + def _embed(texts: list[str]) -> list[list[float]]: + embed_calls.append(list(texts)) + return [[0.0] for _ in texts] + + engine = _FakeEngine( + stamps={("acme/widgets", "main"): ("sha_widgets", INDEX_SEMANTICS_VERSION - 1)} + ) + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=_embed, + engine=engine, + shas_fn=shas, + ) + assert code == 0 + assert shas.calls == [] + assert len(embed_calls) == 1 + # Both main.py's and README.md's chunk text are present -- nothing narrowed. + joined = "\n".join(embed_calls[0]) + assert "def f" in joined + assert "# hi" in joined + + +@pytest.mark.unit +def test_shas_fn_not_called_for_a_never_indexed_repo() -> None: + """T10b: a missing stamp degrades to (None, None) -- version is None, never equal + to INDEX_SEMANTICS_VERSION, so this is the same gate-closed path as a stale + version, exercised separately because it is the far more common real case (a + repo's first index) than an explicit version regression.""" + shas = _RecordingShas() + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=lambda texts: [[0.0] for _ in texts], + shas_fn=shas, + ) + assert code == 0 + assert shas.calls == [] + + +@pytest.mark.unit +def test_shas_fn_narrows_the_embed_set_to_not_unchanged_files() -> None: + """T11: README.md is reported unchanged (carried); main.py is not -- the embedder + must receive only main.py's chunk text.""" + embed_calls: list[list[str]] = [] + + def _embed(texts: list[str]) -> list[list[float]]: + embed_calls.append(list(texts)) + return [[0.0] for _ in texts] + + shas_fn = _RecordingShas(carried={("README.md", _README_SHA)}) + engine = _FakeEngine(stamps={("acme/widgets", "main"): ("sha_old", INDEX_SEMANTICS_VERSION)}) + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=_embed, + engine=engine, + shas_fn=shas_fn, + ) + assert code == 0 + assert len(embed_calls) == 1 + joined = "\n".join(embed_calls[0]) + assert "def f" in joined + assert "# hi" not in joined + + +@pytest.mark.unit +def test_shas_fn_all_unchanged_calls_the_embedder_with_zero_texts() -> None: + """T12 (issue AC1): every file carried -> the embed list is empty, and + _precompute_chunk_writer's own `all_texts` guard means the embedder is not + even called (matching read_indexed_shas' safe-degrade convention: an unused + injection point is never exercised, not called with an empty list).""" + embed_calls: list[list[str]] = [] + + def _embed(texts: list[str]) -> list[list[float]]: + embed_calls.append(list(texts)) + return [[0.0] for _ in texts] + + shas_fn = _RecordingShas( + carried={("README.md", _README_SHA), ("main.py", _MAIN_SHA)}, + ) + engine = _FakeEngine(stamps={("acme/widgets", "main"): ("sha_old", INDEX_SEMANTICS_VERSION)}) + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + idx = _RecordingIndex() + code = _run( + _config(repos=["acme/widgets"]), + idx, + cfg=cfg, + embed_fn=_embed, + engine=engine, + shas_fn=shas_fn, + ) + assert code == 0 + assert embed_calls == [] + # The core index still ran over BOTH files -- narrowing the embed set never + # narrows what index_fn (the store) sees; that is index_repo's classification + # to make, from its own authoritative read. + assert idx.counts == [IndexCounts(files=2, symbols=1, swept=0, edges=0)] + assert idx.chunk_writer is not None + + +@pytest.mark.unit +def test_indexed_summary_line_is_byte_identical_regardless_of_delta_narrowing( + caplog: pytest.LogCaptureFixture, +) -> None: + """T13: the drain loop's `indexed name@branch: files=.. symbols=.. edges=.. swept=..` + line is IndexCounts' own format -- narrowing the embed set must not touch it.""" + shas_fn = _RecordingShas(carried={("README.md", _README_SHA)}) + engine = _FakeEngine(stamps={("acme/widgets", "main"): ("sha_old", INDEX_SEMANTICS_VERSION)}) + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + with caplog.at_level(logging.INFO, logger="indexer.job"): + code = _run( + _config(repos=["acme/widgets"]), + _RecordingIndex(), + cfg=cfg, + embed_fn=lambda texts: [[0.0] for _ in texts], + engine=engine, + shas_fn=shas_fn, + ) + assert code == 0 + indexed_lines = [ + r.getMessage() + for r in caplog.records + if r.name == "indexer.job" and r.getMessage().startswith("indexed ") + ] + assert indexed_lines == ["indexed acme/widgets@main: files=2 symbols=1 edges=0 swept=0"] + + +# --- file-level delta indexing: degraded-semantics run summary (#104) ------- + + +@pytest.mark.unit +def test_precompute_failure_marks_the_branch_outcome_degraded() -> None: + """A chunk-precompute failure marks the resulting BranchOutcome, not just the + per-branch WARNING already covered by + test_embedder_failure_degrades_but_still_indexes_the_core.""" + from indexer.job import _index_one_branch + + def _down(_texts: list[str]) -> list[list[float]]: + raise RuntimeError("serving endpoint unavailable") + + with httpx.Client(transport=httpx.MockTransport(_GitHub())) as client: + outcome = _index_one_branch( + "acme/widgets", + org="acme", + repo="widgets", + branch="main", + is_default=True, + default_head_sha="sha_widgets", + http_client=client, + engine=_FakeEngine(), + index_fn=_RecordingIndex(), + cfg=Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100), + embed_fn=_down, + stamps={}, + started=time.monotonic(), + max_chunks_per_repo=100, + shas_fn=_noop_shas_fn, + ) + assert outcome.status == "indexed" + assert outcome.semantic_degraded is True + + +@pytest.mark.unit +def test_run_emits_one_aggregate_warning_naming_every_degraded_branch( + caplog: pytest.LogCaptureFixture, +) -> None: + """The run-completion WARNING is the greppable record the runbook remedy + (clear the branch's semantics stamp) depends on -- the per-branch warning at + the precompute site is easy to miss in a large run's log.""" + + def _down(_texts: list[str]) -> list[list[float]]: + raise RuntimeError("serving endpoint unavailable") + + cfg = Settings(semantic_enabled=True) + with caplog.at_level(logging.WARNING, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex(), cfg=cfg, embed_fn=_down) + assert code == 0 + warnings = [r.getMessage() for r in caplog.records if r.name == "indexer.job"] + aggregate = [m for m in warnings if m.startswith("1 branch(es) finished with degraded")] + assert len(aggregate) == 1 + assert "acme/widgets@main" in aggregate[0] + + +@pytest.mark.unit +def test_run_emits_no_aggregate_warning_when_nothing_degraded( + caplog: pytest.LogCaptureFixture, +) -> None: + with caplog.at_level(logging.WARNING, logger="indexer.job"): + code = _run(_config(repos=["acme/widgets"]), _RecordingIndex()) + assert code == 0 + warnings = [r.getMessage() for r in caplog.records if r.name == "indexer.job"] + assert not any("degraded semantic coverage" in m for m in warnings) + + # --- config.yaml `semantic:` overlay onto cfg (config.yaml > env > default) -- @@ -1224,6 +1564,7 @@ def test_index_one_inner_reports_discovery_complete_for_a_normal_run() -> None: cfg=Settings(semantic_enabled=False), embed_fn=None, stamps={}, + shas_fn=_noop_shas_fn, ) assert outcome.name == "acme/widgets" assert outcome.discovery_complete is True @@ -1250,6 +1591,7 @@ def test_index_one_inner_reports_discovery_incomplete_when_capped() -> None: cfg=Settings(semantic_enabled=False), embed_fn=None, stamps={}, + shas_fn=_noop_shas_fn, ) assert outcome.discovery_complete is False assert len(outcome.outcomes) == SOFT_BRANCH_CAP @@ -1277,6 +1619,7 @@ def test_index_one_inner_default_flip_mirror_is_complete() -> None: cfg=Settings(semantic_enabled=False), embed_fn=None, stamps={}, + shas_fn=_noop_shas_fn, ) assert [o.branch for o in outcome.outcomes] == ["main"] assert outcome.discovery_complete is True @@ -1584,6 +1927,7 @@ def test_repo_context_is_reset_even_when_the_repo_fails() -> None: cfg=Settings(semantic_enabled=False), embed_fn=None, stamps={}, + shas_fn=_noop_shas_fn, ) assert _repo_ctx.get() == "-" From bf428440be630d55ea89537e6992bbcff0b7a563 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:51:43 -0700 Subject: [PATCH 6/9] test: update the six re-index cases for the delta gate Six tests in tests/integration/test_store.py re-index at a version-matching stamp, so the delta gate introduced by earlier commits on this branch opens and their assertions change: - test_rerun_is_idempotent, test_mark_and_sweep_removes_deleted_file, test_index_repo_records_the_sweep_phase, test_sweep_is_repo_scoped, test_per_branch_cas_resume_is_independent_per_branch: symbols=1/2 -> 0 (a re-run at unchanged content classifies unchanged, not a re-insert count). Companion test_rerun_is_idempotent_preserves_row_identity added for the stronger row-identity property the original was reaching for. - test_reindex_replaces_stale_edges_for_the_same_file, test_reindex_to_zero_edges_sheds_all_rows: these two deliberately hold content_sha identical while varying extraction output (the "unconditional-delete guard"), which the delta path would otherwise skip. Forced closed via the existing `UPDATE repo_branches SET index_semantics_version = NULL` idiom rather than relaxing their assertions -- in production, identical content_sha with divergent extraction can only happen via an extractor change, which mandates the same version bump. --- tests/integration/test_store.py | 78 ++++++++++++++++++++++++++++----- 1 file changed, 67 insertions(+), 11 deletions(-) diff --git a/tests/integration/test_store.py b/tests/integration/test_store.py index 22cc75e..3d83bc4 100644 --- a/tests/integration/test_store.py +++ b/tests/integration/test_store.py @@ -136,16 +136,41 @@ def test_first_run_populates_and_stamps_commit(conn: Connection) -> None: @pytest.mark.integration def test_rerun_is_idempotent(conn: Connection) -> None: + """The first run stamps INDEX_SEMANTICS_VERSION, so the second run (identical + content, identical head_sha) hits the file-level delta gate: both files are + classified unchanged, so `symbols` in the returned IndexCounts is 0 (files + seen, not files re-inserted) -- not the pre-delta re-insert count of 2. See + test_rerun_is_idempotent_preserves_row_identity below for the stronger + row-identity property this test was originally reaching for. + """ _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL)) counts = _index_default( conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL) ) - assert counts == IndexCounts(files=2, symbols=2, swept=0, edges=0) + assert counts == IndexCounts(files=2, symbols=0, swept=0, edges=0) assert _count(conn, "repos") == 1 assert _count(conn, "files") == 2 assert _count(conn, "symbols") == 2 +@pytest.mark.integration +def test_rerun_is_idempotent_preserves_row_identity(conn: Connection) -> None: + """The stronger property test_rerun_is_idempotent was originally reaching + for: an unchanged re-run does not delete-and-reinsert ANYTHING -- files.id + and symbols.id survive byte-identical across the two runs (a delete-reinsert + would renumber the serials).""" + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL)) + files_before = dict(conn.execute(text("SELECT path, id FROM files ORDER BY path")).all()) + symbols_before = sorted(conn.execute(text("SELECT id FROM symbols")).scalars().all()) + conn.rollback() + + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL)) + files_after = dict(conn.execute(text("SELECT path, id FROM files ORDER BY path")).all()) + symbols_after = sorted(conn.execute(text("SELECT id FROM symbols")).scalars().all()) + assert files_after == files_before + assert symbols_after == symbols_before + + @pytest.mark.integration def test_mark_and_sweep_removes_deleted_file(conn: Connection) -> None: # util.py has a real call site so the writer produces a reference_edges row @@ -168,15 +193,23 @@ def test_mark_and_sweep_removes_deleted_file(conn: Connection) -> None: # (production hands a fresh engine.connect() per repo). conn.rollback() - # Re-run without util.py and with a new head SHA -> util.py is swept. + # Re-run without util.py and with a new head SHA -> util.py is swept. MAIN's + # content is identical across both runs, and the first run already stamped + # INDEX_SEMANTICS_VERSION, so the delta gate classifies MAIN unchanged: no + # symbol rewrite (symbols=0, not a re-insert count). counts = _index_default(conn, name="acme/widgets", head_sha="sha_second", items=_items(MAIN)) - assert counts == IndexCounts(files=1, symbols=1, swept=1, edges=0) + assert counts == IndexCounts(files=1, symbols=0, swept=1, edges=0) assert _count(conn, "files", "path = 'util.py'") == 0 assert _count(conn, "symbols", f"file_id = {removed_file_id}") == 0 # cascade assert _count(conn, "reference_edges", f"file_id = {removed_file_id}") == 0 # cascade assert _count(conn, "files") == 1 - assert _count(conn, "files", "commit = 'sha_second'") == 1 + # MAIN's row was classified unchanged (skipped), so its `commit` column stays + # at the FIRST run's SHA -- files.commit goes staler under delta indexing by + # design (no production read path resolves `commit:` from it; see + # indexer/store.py's index_repo docstring). + assert _count(conn, "files", "commit = 'sha_first'") == 1 + assert _count(conn, "files", "commit = 'sha_second'") == 0 @pytest.mark.integration @@ -196,14 +229,15 @@ def test_index_repo_records_the_sweep_phase(conn: Connection) -> None: token = install_timer(timer) try: # Re-run without util.py at a new SHA: the same scenario as - # test_mark_and_sweep_removes_deleted_file, so the counts are unchanged. + # test_mark_and_sweep_removes_deleted_file, so the counts are unchanged + # (MAIN classified unchanged under the delta gate -> symbols=0). counts = _index_default( conn, name="acme/widgets", head_sha="sha_second", items=_items(MAIN) ) finally: reset_timer(token) - assert counts == IndexCounts(files=1, symbols=1, swept=1, edges=0) + assert counts == IndexCounts(files=1, symbols=0, swept=1, edges=0) assert timer.total("sweep") > 0.0 # index_repo measures the sweep and nothing else -- every other phase is # job.py's to record. @@ -224,8 +258,10 @@ def test_sweep_is_repo_scoped(conn: Connection) -> None: conn.rollback() # Re-index A without util.py at a new SHA -> A's util.py swept, B untouched. + # MAIN unchanged under the delta gate (A's first run already stamped + # INDEX_SEMANTICS_VERSION) -> symbols=0, same as the tests above. counts = _index_default(conn, name="acme/a", head_sha="a_second", items=_items(MAIN)) - assert counts == IndexCounts(files=1, symbols=1, swept=1, edges=0) + assert counts == IndexCounts(files=1, symbols=0, swept=1, edges=0) assert _count(conn, "files", "repo_id = (SELECT id FROM repos WHERE name = 'acme/b')") == ( b_files_before @@ -267,6 +303,17 @@ def test_reindex_replaces_stale_edges_for_the_same_file(conn: Connection) -> Non deleted and reinserted, not accumulated -- proven by driving the two runs' ``ex.edges`` directly rather than depending on the real extractor to disagree with itself on unchanged content. + + This is precisely the case the file-level delta gate (#104) would otherwise + skip: identical content_sha, different extraction output. In production that + combination can only arise from an extractor change, which mandates an + INDEX_SEMANTICS_VERSION bump and therefore closes the gate on its own -- so + forcing it closed here (the same ``UPDATE repo_branches SET + index_semantics_version = NULL`` idiom as + test_legacy_null_semantics_version_is_rewritten) is faithful to production, + not a workaround. This is the "unconditional-delete guard" and it must keep + its ORIGINAL assertion, not a relaxed one -- see indexer/AGENTS.md / + the plan for issue #104, §2.6a. """ symbol = ExtractedSymbol("f", "function", 1, 3) content = "def f():\n target()\n return 1\n" @@ -279,7 +326,8 @@ def test_reindex_replaces_stale_edges_for_the_same_file(conn: Connection) -> Non ) file_id = conn.execute(text("SELECT id FROM files WHERE path = 'main.py'")).scalar_one() assert _count(conn, "reference_edges", f"file_id = {file_id}") == 1 - conn.rollback() + conn.execute(text("UPDATE repo_branches SET index_semantics_version = NULL")) + conn.commit() second_edge = ExtractedEdge(kind="call", target="new_target", line=2, enclosing=symbol) _index_default( @@ -326,6 +374,12 @@ def test_reindex_to_zero_edges_sheds_all_rows(conn: Connection) -> None: upsert resolves to the SAME ``file_id`` -- isolating the write-side guard (``ex.edges`` empty must still run the delete) from the unrelated delete-and-reinsert-under-a-new-file-id path already covered above. + + Same rationale as test_reindex_replaces_stale_edges_for_the_same_file: this + is identical content_sha with divergent extraction output, which in + production can only happen via an extractor change (mandating a semantics + version bump). The gate is forced closed here rather than the expected + values relaxed -- see the plan for issue #104, §2.6a. """ symbol = ExtractedSymbol("f", "function", 1, 3) content = "def f():\n helper()\n return 1\n" @@ -338,7 +392,8 @@ def test_reindex_to_zero_edges_sheds_all_rows(conn: Connection) -> None: ) file_id = conn.execute(text("SELECT id FROM files WHERE path = 'main.py'")).scalar_one() assert _count(conn, "reference_edges", f"file_id = {file_id}") == 1 - conn.rollback() + conn.execute(text("UPDATE repo_branches SET index_semantics_version = NULL")) + conn.commit() # Same content (same file_id) but this run's extraction yields zero edges. counts = _index_default( @@ -598,7 +653,8 @@ def test_per_branch_cas_resume_is_independent_per_branch(conn: Connection) -> No conn.rollback() # Re-indexing 'a' again must succeed against its own baseline, unaffected by - # 'b' having indexed in between. + # 'b' having indexed in between. MAIN unchanged under the delta gate ('a's + # first run already stamped INDEX_SEMANTICS_VERSION) -> symbols=0. counts = index_repo( conn, name="acme/widgets", @@ -607,7 +663,7 @@ def test_per_branch_cas_resume_is_independent_per_branch(conn: Connection) -> No head_sha="sha_a2", items=_items(MAIN), ) - assert counts == IndexCounts(files=1, symbols=1, swept=0, edges=0) + assert counts == IndexCounts(files=1, symbols=0, swept=0, edges=0) stamps = dict(conn.execute(text("SELECT branch, last_indexed_commit FROM repo_branches")).all()) assert stamps == {"a": "sha_a2", "b": "sha_b"} From a7ab105dddaf600b7768666be704de10fcb75563 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:51:52 -0700 Subject: [PATCH 7/9] test: integration delta coverage (row identity, sweep, CAS, semantics gate, EXPLAIN) New tests/integration/test_store_delta.py, runnable locally against codesearch-pg (no chunks table, no lakebase_* extension -- that's what makes it runnable at all): row-identity proof for the unchanged path, a small 1-changed/1-added/1-deleted delta, multi-branch membership-only dedup, a semantics-version mismatch forcing the full path, the provenance gate (a stale sibling branch forces the full path), both directions of the branches-array/repo_branches membership invariant, the empty-seen-set guard and CAS still holding with delta on, the zero-parse stamp fix (the BLOCKER regression this branch's earlier commits fixed), and an EXPLAIN proof that the pre-read is index-served (ix_files_branches_gin / Index Only Scan on uq_files_repo_path_sha, zero heap fetches) rather than a Seq Scan. tests/integration/test_store_chunk_writer.py gains the chunk-side delta cases (unchanged file never calls chunk_writer and preserves chunk ids; membership-only backfills previously-missing chunks) and reviews test_reindex_is_idempotent_for_chunks against the same six-test treatment (symbols 2 -> 0). All Lakebase-deferred -- this module's fixture needs lakebase_vector, which no local Postgres provides -- reasoned through rather than locally verified. --- tests/integration/test_store_chunk_writer.py | 125 +++- tests/integration/test_store_delta.py | 644 +++++++++++++++++++ 2 files changed, 767 insertions(+), 2 deletions(-) create mode 100644 tests/integration/test_store_delta.py diff --git a/tests/integration/test_store_chunk_writer.py b/tests/integration/test_store_chunk_writer.py index 0dd67f5..30815b4 100644 --- a/tests/integration/test_store_chunk_writer.py +++ b/tests/integration/test_store_chunk_writer.py @@ -12,6 +12,17 @@ chunk_writer ride the same conn.begin() as the rest of that file's row, and cascade-delete when the file is swept (FK ON DELETE CASCADE), exactly like symbols. + +**Every test in this module is Lakebase-deferred for issue #104**: this is the +one module whose fixture builds the ``chunks`` table and needs +``lakebase_vector``, which no local Postgres image provides (see +``tests/integration/test_store_delta.py``'s module docstring for why the +core delta suite deliberately lives elsewhere instead). The delta-specific +additions here (the unchanged-file / membership-only chunk cases, and +``test_reindex_is_idempotent_for_chunks``'s reviewed expectations) are +reasoned through against ``indexer/store.py``'s ``_union_membership`` and +the per-file loop, never verified by a local run -- flagged explicitly in the +PR body, per the plan for issue #104, §2.6a / §3.2. """ from __future__ import annotations @@ -136,6 +147,19 @@ def test_chunk_writer_writes_inside_the_transaction(conn: Connection) -> None: @pytest.mark.integration def test_reindex_is_idempotent_for_chunks(conn: Connection) -> None: + """Reviewed against the plan for issue #104, §2.6a's treatment table (Lakebase- + deferred, so this review is reasoned through rather than locally verified -- + see the module docstring). Unlike the two GUARD tests in + ``tests/integration/test_store.py`` (which deliberately hold content_sha + identical while varying EXTRACTION output, and must keep their gate forced + closed), this run is identical in every respect -- content, head_sha, AND + extraction. The first run stamps INDEX_SEMANTICS_VERSION, so the second + classifies both files unchanged: ``symbols`` drops from 2 (a delete-reinsert + count) to 0 (nothing rewritten), a plain (a)-style value update. The chunk + row COUNT is unaffected either way -- 2 rows survive whether by an idempotent + delete-reinsert (pre-#104) or by never being touched at all (unchanged, under + #104) -- so that assertion needed no change, only its reasoning. + """ items = _items(MAIN, UTIL) index_repo( conn, @@ -156,8 +180,8 @@ def test_reindex_is_idempotent_for_chunks(conn: Connection) -> None: items=items, chunk_writer=_stub_chunk_writer, ) - assert counts == IndexCounts(files=2, symbols=2, swept=0, edges=0) - assert _count(conn, "chunks") == 2 # delete-and-reinsert, not duplicated + assert counts == IndexCounts(files=2, symbols=0, swept=0, edges=0) + assert _count(conn, "chunks") == 2 # untouched, not delete-and-reinserted @pytest.mark.integration @@ -191,3 +215,100 @@ def test_chunks_cascade_delete_when_file_is_swept(conn: Connection) -> None: assert _count(conn, "files", "path = 'util.py'") == 0 assert _count(conn, "chunks", f"file_id = {util_file_id}") == 0 # cascade assert _count(conn, "chunks") == 1 + + +# --- File-level delta indexing (#104): unchanged / membership-only chunks --- +# Lakebase-deferred -- see the module docstring. Reasoned through against +# indexer/store.py's per-file loop and _union_membership, never locally run. + + +@pytest.mark.integration +def test_unchanged_file_never_calls_chunk_writer_and_preserves_chunk_ids( + conn: Connection, +) -> None: + """An unchanged file's chunks.id values are IDENTICAL before/after, and + chunk_writer is never called for it at all -- proven with a call-tracking + wrapper, not just by the row count staying flat (which an idempotent + delete-reinsert of identical content would also produce, as + test_reindex_is_idempotent_for_chunks above shows).""" + calls: list[str] = [] + + def _tracking_chunk_writer( + conn: Connection, repo_id: int, file_id: int, pf: ParsedFile + ) -> None: + calls.append(pf.path) + _stub_chunk_writer(conn, repo_id, file_id, pf) + + index_repo( + conn, + name="acme/widgets", + branch="main", + is_default=True, + head_sha="sha_first", + items=_items(MAIN, UTIL), + chunk_writer=_tracking_chunk_writer, + ) + chunk_ids_before = sorted(conn.execute(text("SELECT id FROM chunks")).scalars().all()) + conn.rollback() + calls.clear() + + index_repo( + conn, + name="acme/widgets", + branch="main", + is_default=True, + head_sha="sha_second", + items=_items(MAIN, UTIL), + chunk_writer=_tracking_chunk_writer, + ) + assert calls == [] + chunk_ids_after = sorted(conn.execute(text("SELECT id FROM chunks")).scalars().all()) + assert chunk_ids_after == chunk_ids_before + + +@pytest.mark.integration +def test_membership_only_file_backfills_previously_missing_chunks(conn: Connection) -> None: + """§2.4 note 3: a membership-acquired file whose row had ZERO chunk rows + (branch 'a' wrote it with chunk_writer=None, i.e. semantic-off) ends branch + 'b's acquiring run WITH chunk rows. Membership-only writes chunks even + though it writes no symbols/edges -- the vectors are already in hand + (job.py embeds every file the advisory read did not call unchanged), so + skipping the write would make a semantic-off-then-on transition's gap + permanent for the acquiring branch instead of backfilling it.""" + index_repo( + conn, + name="acme/widgets", + branch="a", + is_default=True, + head_sha="sha_a1", + items=_items(MAIN), # no chunk_writer -> main.py has zero chunk rows + ) + conn.rollback() + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b1", + items=_items(UTIL), + chunk_writer=_stub_chunk_writer, + ) + main_file_id = conn.execute(text("SELECT id FROM files WHERE path = 'main.py'")).scalar_one() + assert _count(conn, "chunks", f"file_id = {main_file_id}") == 0 + conn.rollback() + + # branch 'b' is now at the current semantics version (its own baseline), + # so acquiring main.py (identical content, still stored under 'a') takes + # the membership-only path. + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b2", + items=_items(MAIN, UTIL), + chunk_writer=_stub_chunk_writer, + ) + assert _count(conn, "chunks", f"file_id = {main_file_id}") == 1 + branches = conn.execute(text("SELECT branches FROM files WHERE path = 'main.py'")).scalar_one() + assert sorted(branches) == ["a", "b"] diff --git a/tests/integration/test_store_delta.py b/tests/integration/test_store_delta.py new file mode 100644 index 0000000..c10ca22 --- /dev/null +++ b/tests/integration/test_store_delta.py @@ -0,0 +1,644 @@ +"""Integration tests for the file-level delta path (issue #104) against real Postgres. + +Row-identity proof -- that skipping a file really does leave its stored serials +untouched -- is this module's job; ``tests/unit/test_store_delta.py`` pins the +exact *statement inventory* each classification produces against a fake +connection. Clones ``tests/integration/test_store.py``'s throwaway-schema fixture +idiom (own copy, per ``tests/integration/AGENTS.md``'s no-conftest convention): +a clean schema per run, the durable-core DDL via ``Base.metadata.create_all``, +and a per-connection ``search_path`` that propagates into ``index_repo``'s DML. + +**Deliberately builds no ``chunks`` table and creates no ``lakebase_*`` +extension.** That is the entire reason this module runs locally at all -- +``test_store_chunk_writer.py`` errors at *module fixture* setup on +``CREATE EXTENSION IF NOT EXISTS lakebase_vector CASCADE``, which no local +Postgres image provides. Chunk-touching delta cases live there instead +(Lakebase-deferred; see that module's docstring). +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Iterator +from typing import Any + +import pytest +from sqlalchemy import Connection, text + +from app.db.client import create_db_engine +from app.db.models import INDEX_SEMANTICS_VERSION, Base +from indexer.languages import ExtractedSymbol, FileExtraction, IndexCounts, ParsedFile +from indexer.store import index_repo + +SCHEMA = "test_store_delta" + + +@pytest.fixture +def conn() -> Iterator[Connection]: + engine = create_db_engine() + connection = engine.connect() + try: + connection.execute(text(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE")) + connection.execute(text(f"CREATE SCHEMA {SCHEMA}")) + connection.execute(text(f"SET search_path TO {SCHEMA}, public")) + connection.commit() + + Base.metadata.create_all(bind=connection) + connection.commit() + + yield connection + finally: + connection.rollback() + connection.execute(text(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE")) + connection.commit() + connection.close() + engine.dispose() + + +def _pf(path: str, content: str) -> ParsedFile: + return ParsedFile(path=path, lang="python", size=len(content.encode()), content=content) + + +def _items( + *specs: tuple[str, str, list[ExtractedSymbol]], +) -> list[tuple[ParsedFile, FileExtraction]]: + return [ + (_pf(path, content), FileExtraction(symbols=syms, edges=[])) + for path, content, syms in specs + ] + + +def _fn(name: str, n: int) -> tuple[str, str, list[ExtractedSymbol]]: + """One deterministic (path, content, symbols) triple, distinguishable by ``n``.""" + content = f"def f{n}():\n return {n}\n" + return (f"{name}{n}.py", content, [ExtractedSymbol(f"f{n}", "function", 1, 2)]) + + +MAIN = ("main.py", "def f():\n return 1\n", [ExtractedSymbol("f", "function", 1, 2)]) +UTIL = ("util.py", "def g():\n return 2\n", [ExtractedSymbol("g", "function", 1, 2)]) + + +def _index_default( + conn: Connection, *, name: str, head_sha: str, items: list[tuple[ParsedFile, FileExtraction]] +) -> IndexCounts: + return index_repo( + conn, name=name, branch="main", is_default=True, head_sha=head_sha, items=items + ) + + +def _count(conn: Connection, table: str, where: str = "") -> int: + sql = f"SELECT count(*) FROM {table}" + if where: + sql += f" WHERE {where}" + return int(conn.execute(text(sql)).scalar_one()) + + +def _symbol_ids(conn: Connection, path: str) -> list[int]: + return sorted( + conn.execute( + text("SELECT s.id FROM symbols s JOIN files f ON f.id = s.file_id WHERE f.path = :p"), + {"p": path}, + ) + .scalars() + .all() + ) + + +def _delta_lines(caplog: pytest.LogCaptureFixture) -> list[str]: + return [ + r.getMessage() + for r in caplog.records + if r.name == "indexer.store" and "delta write set" in r.getMessage() + ] + + +# --- Test 15: row-identity proof, the "unchanged" class writes NOTHING ------ + + +@pytest.mark.integration +def test_unchanged_rerun_preserves_every_row_identity( + conn: Connection, caplog: pytest.LogCaptureFixture +) -> None: + """Re-index identical content at a NEW head_sha: files.id and symbols.id are + IDENTICAL before/after -- a delete-reinsert would renumber the serials, so + identical ids are the precise proof that zero rows were written. + ``counts.symbols == 0`` while ``counts.files == N``, and the + ``delta write set 0/N`` line is emitted. + """ + items = _items(MAIN, UTIL) + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=items) + files_before = dict(conn.execute(text("SELECT path, id FROM files ORDER BY path")).all()) + symbols_before = sorted(conn.execute(text("SELECT id FROM symbols")).scalars().all()) + conn.rollback() + + with caplog.at_level(logging.INFO, logger="indexer.store"): + counts = _index_default( + conn, name="acme/widgets", head_sha="sha_second", items=_items(MAIN, UTIL) + ) + assert counts == IndexCounts(files=2, symbols=0, swept=0, edges=0) + + files_after = dict(conn.execute(text("SELECT path, id FROM files ORDER BY path")).all()) + symbols_after = sorted(conn.execute(text("SELECT id FROM symbols")).scalars().all()) + assert files_after == files_before + assert symbols_after == symbols_before + + lines = _delta_lines(caplog) + assert len(lines) == 1 + assert "delta write set 0/2 files (unchanged=2 membership=0, semantics gate open)" in lines[0] + + +# --- Test 16: small delta (1 changed, 1 added, 1 deleted) -- issue AC2 ------- + + +@pytest.mark.integration +def test_small_delta_writes_only_the_changed_added_and_sweeps_the_deleted(conn: Connection) -> None: + """1 changed, 1 unchanged, 1 added, 1 deleted -- the changed file's symbol ids + change, the unchanged file's do not, the added file is present, and the + deleted file is swept. Issue #104's acceptance criterion 2. + """ + unchanged_symbol = ExtractedSymbol("f", "function", 1, 2) + changed_v1 = ( + "changed.py", + "def c():\n return 1\n", + [ExtractedSymbol("c", "function", 1, 2)], + ) + to_delete = ("gone.py", "def d():\n return 1\n", [ExtractedSymbol("d", "function", 1, 2)]) + _index_default( + conn, + name="acme/widgets", + head_sha="sha_first", + items=_items( + ("unchanged.py", "def f():\n return 1\n", [unchanged_symbol]), changed_v1, to_delete + ), + ) + unchanged_ids_before = _symbol_ids(conn, "unchanged.py") + changed_ids_before = _symbol_ids(conn, "changed.py") + assert unchanged_ids_before and changed_ids_before + conn.rollback() + + changed_v2 = ( + "changed.py", + "def c():\n return 2\n", + [ExtractedSymbol("c", "function", 1, 2)], + ) + added = ("added.py", "def a():\n return 1\n", [ExtractedSymbol("a", "function", 1, 2)]) + counts = _index_default( + conn, + name="acme/widgets", + head_sha="sha_second", + items=_items( + ("unchanged.py", "def f():\n return 1\n", [unchanged_symbol]), changed_v2, added + ), + ) + # swept=2: gone.py (removed outright) AND changed.py's OLD content_sha row + # (the changed-file upsert mints a NEW row under the new content_sha, since + # uq_files_repo_path_sha is keyed on content -- the stale old-sha row is + # exactly what the sweep exists to reap, unrelated to delta). + assert counts == IndexCounts(files=3, symbols=2, swept=2, edges=0) + + assert _symbol_ids(conn, "unchanged.py") == unchanged_ids_before + assert _symbol_ids(conn, "changed.py") != changed_ids_before + assert _count(conn, "files", "path = 'added.py'") == 1 + assert _count(conn, "files", "path = 'gone.py'") == 0 + + +# --- Test 17: multi-branch dedup takes the membership-only path -- AC3 ------ + + +@pytest.mark.integration +def test_membership_only_dedup_across_branches_preserves_symbol_ids( + conn: Connection, caplog: pytest.LogCaptureFixture +) -> None: + """Branch 'b' acquires content already stored under branch 'a': ``branches`` + becomes ``['a', 'b']`` (sorted, matching the array_agg(DISTINCT ...) idiom), + symbol ids are UNCHANGED (no rewrite), and the delta write set line reports + ``membership=1``. Issue #104's acceptance criterion 3. + """ + index_repo( + conn, + name="acme/widgets", + branch="a", + is_default=True, + head_sha="sha_a1", + items=_items(MAIN), + ) + conn.rollback() + # branch 'b' must ALREADY be at the current semantics version (its own + # baseline) for the delta gate to be open when it next acquires MAIN -- + # a brand-new branch's first run is always full-path. + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b1", + items=_items(UTIL), + ) + main_ids_before = _symbol_ids(conn, "main.py") + assert main_ids_before + conn.rollback() + + with caplog.at_level(logging.INFO, logger="indexer.store"): + counts = index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b2", + items=_items(MAIN, UTIL), + ) + assert counts == IndexCounts(files=2, symbols=0, swept=0, edges=0) + + assert _symbol_ids(conn, "main.py") == main_ids_before + branches = conn.execute(text("SELECT branches FROM files WHERE path = 'main.py'")).scalar_one() + assert sorted(branches) == ["a", "b"] + + lines = _delta_lines(caplog) + assert len(lines) == 1 + assert "membership=1" in lines[0] + + +@pytest.mark.integration +def test_membership_only_row_gets_no_symbol_or_edge_statement(conn: Connection) -> None: + """Companion to the row-identity proof above, phrased as a statement-absence + check rather than an id-equality check: the acquiring branch's run inserts + NO symbols row for the acquired file at all (there was never a duplicate to + delete-and-reinsert).""" + index_repo( + conn, + name="acme/widgets", + branch="a", + is_default=True, + head_sha="sha_a1", + items=_items(MAIN), + ) + conn.rollback() + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b1", + items=_items(UTIL), + ) + symbols_before = _count(conn, "symbols") + conn.rollback() + + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b2", + items=_items(MAIN, UTIL), + ) + assert _count(conn, "symbols") == symbols_before + + +# --- Test 18: a semantics-version mismatch forces the full path -- AC4 ------ + + +@pytest.mark.integration +def test_stale_semantics_version_forces_the_full_path_for_every_file(conn: Connection) -> None: + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL)) + main_ids_before = _symbol_ids(conn, "main.py") + util_ids_before = _symbol_ids(conn, "util.py") + conn.execute( + text("UPDATE repo_branches SET index_semantics_version = :v"), + {"v": INDEX_SEMANTICS_VERSION - 1}, + ) + conn.commit() + + counts = _index_default( + conn, name="acme/widgets", head_sha="sha_second", items=_items(MAIN, UTIL) + ) + assert counts == IndexCounts(files=2, symbols=2, swept=0, edges=0) + assert _symbol_ids(conn, "main.py") != main_ids_before + assert _symbol_ids(conn, "util.py") != util_ids_before + + +# --- Test 19: the provenance gate -- a stale SIBLING branch closes it ------- + + +@pytest.mark.integration +def test_provenance_gate_forces_the_full_path_when_a_sibling_branch_is_stale( + conn: Connection, +) -> None: + """The counter-example the provenance gate (statement 4) exists to close: + 'stale_sibling' wrote main.py at the current version, then regressed to an + older one (simulating a failed re-index). 'acquirer' is itself at the + current version and tries to acquire main.py membership-only -- but every + ``repo_branches`` row for this repo must be current for that path to be + taken, and stale_sibling's is not, so 'acquirer' gets the FULL path + instead (proven by main.py's symbol ids changing under 'acquirer'). + """ + index_repo( + conn, + name="acme/widgets", + branch="stale_sibling", + is_default=True, + head_sha="sha_s1", + items=_items(MAIN), + ) + main_ids_before = _symbol_ids(conn, "main.py") + conn.execute( + text( + "UPDATE repo_branches SET index_semantics_version = :v WHERE branch = 'stale_sibling'" + ), + {"v": INDEX_SEMANTICS_VERSION - 1}, + ) + conn.commit() + + index_repo( + conn, + name="acme/widgets", + branch="acquirer", + is_default=False, + head_sha="sha_a1", + items=_items(UTIL), + ) + conn.rollback() + + index_repo( + conn, + name="acme/widgets", + branch="acquirer", + is_default=False, + head_sha="sha_a2", + items=_items(MAIN, UTIL), + ) + + assert _symbol_ids(conn, "main.py") != main_ids_before + branches = conn.execute(text("SELECT branches FROM files WHERE path = 'main.py'")).scalar_one() + assert sorted(branches) == ["acquirer", "stale_sibling"] + + +# --- Tests 20a/20b: the membership invariant, both directions -------------- + + +@pytest.mark.integration +def test_no_files_row_ever_has_an_empty_branches_array(conn: Connection) -> None: + """Forward direction: both sweep sites (the per-branch sweep and the + array-remove path) delete a row once its branches array is emptied, never + leaving a zombie row behind. Exercised across additions, a shared-then- + removed-from-one-branch file, and a fully-removed file. + """ + index_repo( + conn, + name="acme/widgets", + branch="a", + is_default=True, + head_sha="sha_a1", + items=_items(MAIN, UTIL), + ) + conn.rollback() + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b1", + items=_items(UTIL), + ) + conn.rollback() + # 'a' drops both files -- util.py loses only 'a' (shared with 'b'), main.py + # loses its only branch and must be deleted outright. + index_repo(conn, name="acme/widgets", branch="a", is_default=True, head_sha="sha_a2", items=[]) + conn.rollback() + index_repo( + conn, + name="acme/widgets", + branch="a", + is_default=True, + head_sha="sha_a3", + items=_items(UTIL), + ) + conn.rollback() + index_repo(conn, name="acme/widgets", branch="a", is_default=True, head_sha="sha_a4", items=[]) + + assert _count(conn, "files", "cardinality(branches) = 0") == 0 + + +@pytest.mark.integration +def test_every_branch_on_a_files_row_has_a_repo_branches_row(conn: Connection) -> None: + """Converse direction: index_repo writes statement 2 (the repo_branches + upsert) before any file row for that branch, so no branches array element + can ever dangle without a matching repo_branches row. The provenance gate + (statement 4) depends on this holding.""" + index_repo( + conn, + name="acme/widgets", + branch="a", + is_default=True, + head_sha="sha_a1", + items=_items(MAIN), + ) + conn.rollback() + index_repo( + conn, + name="acme/widgets", + branch="b", + is_default=False, + head_sha="sha_b1", + items=_items(MAIN, UTIL), + ) + + dangling = conn.execute( + text( + "SELECT count(*) FROM files f, unnest(f.branches) AS b " + "WHERE NOT EXISTS (" + " SELECT 1 FROM repo_branches rb WHERE rb.repo_id = f.repo_id AND rb.branch = b" + ")" + ) + ).scalar_one() + assert dangling == 0 + + +# --- Test 21: empty-seen-set guard and per-branch CAS still hold with delta - + + +@pytest.mark.integration +def test_empty_seen_set_guard_holds_with_the_delta_gate_open(conn: Connection) -> None: + """The empty-seen-set guard (skip the sweep, WARN, return 0) is delta-blind by + construction -- it fires on ``seen_paths`` being empty, before any + classification runs. Pinned again here because it is exactly the run shape + the delta gate makes common (a branch whose HEAD moved but touched nothing + indexable).""" + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN)) + conn.rollback() + # Second run: delta gate is open (first run stamped INDEX_SEMANTICS_VERSION), + # but this run parses zero files. + counts = _index_default(conn, name="acme/widgets", head_sha="sha_second", items=[]) + assert counts == IndexCounts(files=0, symbols=0, swept=0, edges=0) + assert _count(conn, "files", "path = 'main.py'") == 1 + branches = conn.execute(text("SELECT branches FROM files WHERE path = 'main.py'")).scalar_one() + assert branches == ["main"] + + +@pytest.mark.integration +def test_cas_still_rejects_a_stale_baseline_with_delta_on(conn: Connection) -> None: + """The CAS predicate is statement 5, downstream of every delta statement -- + proven still load-bearing by forcing a conflict on a branch whose delta gate + is open (second run onward).""" + from indexer.store import StaleIndexError, _stamp_repo_branch + + _index_default(conn, name="acme/widgets", head_sha="sha_a", items=_items(MAIN, UTIL)) + conn.rollback() + _index_default(conn, name="acme/widgets", head_sha="sha_b", items=_items(MAIN, UTIL)) + files_before = _count(conn, "files") + repo_id = int( + conn.execute(text("SELECT id FROM repos WHERE name = 'acme/widgets'")).scalar_one() + ) + conn.rollback() + + with pytest.raises(StaleIndexError, match="wrong_sha"), conn.begin(): + conn.execute(text("DELETE FROM files WHERE path = 'util.py'")) + _stamp_repo_branch( + conn, + name="acme/widgets", + branch="main", + repo_id=repo_id, + head_sha="sha_c", + baseline_commit="wrong_sha", + baseline_version=INDEX_SEMANTICS_VERSION, + ) + assert _count(conn, "files") == files_before + + +# --- Test 22 (BLOCKER): a zero-parse run must NOT advance the semantics stamp + + +@pytest.mark.integration +def test_zero_parse_run_does_not_advance_the_semantics_stamp(conn: Connection) -> None: + """The base-case fix (§2.3): index a branch non-empty, force its stored + version DOWN to simulate a pre-transition stamp, then re-index with + ``items=[]`` at a NEW head SHA. The stored version must stay at the forced + value (NOT advance to INDEX_SEMANTICS_VERSION) -- proving a zero-parse run + cannot manufacture a spurious "current version" base case for the delta + induction. The commit still advances (the run DID look at that SHA); only + the version is held back. The FOLLOWING non-empty run must then full-path + every file, since the branch is still stamped stale. + """ + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(MAIN, UTIL)) + main_ids_before = _symbol_ids(conn, "main.py") + conn.execute(text("UPDATE repo_branches SET index_semantics_version = 3")) + conn.commit() + + counts = _index_default(conn, name="acme/widgets", head_sha="sha_zero", items=[]) + assert counts == IndexCounts(files=0, symbols=0, swept=0, edges=0) + stamp = conn.execute( + text( + "SELECT rb.last_indexed_commit, rb.index_semantics_version FROM repo_branches rb " + "JOIN repos r ON r.id = rb.repo_id WHERE r.name = 'acme/widgets'" + ) + ).one() + # last_indexed_commit DID advance (the run looked at sha_zero); the version + # did NOT (nothing was indexed at it). + assert stamp == ("sha_zero", 3) + conn.rollback() + + # The next non-empty run sees baseline_version=3 != INDEX_SEMANTICS_VERSION, + # so the gate is closed and every file takes the full path -- symbol ids + # change even though the content is byte-identical to sha_first's. + _index_default(conn, name="acme/widgets", head_sha="sha_second", items=_items(MAIN, UTIL)) + assert _symbol_ids(conn, "main.py") != main_ids_before + stamp_after = conn.execute( + text( + "SELECT rb.index_semantics_version FROM repo_branches rb " + "JOIN repos r ON r.id = rb.repo_id WHERE r.name = 'acme/widgets'" + ) + ).scalar_one() + assert stamp_after == INDEX_SEMANTICS_VERSION + + +# --- Test 23: the pre-read is index-served, not a Seq Scan ------------------ + + +def _explain( + conn: Connection, sql: str, params: dict[str, Any], *, analyze: bool = False +) -> dict[str, Any]: + """``EXPLAIN (FORMAT JSON)`` for ``sql`` with the seq-scan escape hatch disabled, + scoped to a SAVEPOINT so the ``enable_seqscan`` GUC change never leaks past this + call (clones ``tests/integration/test_query_compiler.py``'s ``_explain_plan`` + idiom, per ``tests/integration/AGENTS.md``).""" + mode = "ANALYZE, FORMAT JSON" if analyze else "FORMAT JSON" + savepoint = conn.begin_nested() + try: + conn.execute(text("SET LOCAL enable_seqscan = off")) + raw = conn.execute(text(f"EXPLAIN ({mode}) {sql}"), params).scalar_one() + finally: + savepoint.rollback() + plan_list = json.loads(raw) if isinstance(raw, str) else raw + plan: dict[str, Any] = plan_list[0]["Plan"] + return plan + + +def _plan_nodes(plan: dict[str, Any]) -> Iterator[dict[str, Any]]: + yield plan + for child in plan.get("Plans", []) or []: + yield from _plan_nodes(child) + + +@pytest.mark.integration +def test_the_delta_pre_read_is_index_served(conn: Connection) -> None: + """Statement 3a (``branches @> ...``) plans an index scan on + ``ix_files_branches_gin``; statement 3b (unqualified, repo-scoped) plans an + Index Only Scan on ``uq_files_repo_path_sha`` with zero heap fetches -- + proving ``read_repo_content_shas`` never resorts to a Seq Scan and never + pulls ``content`` off the heap. Seeded with 200 rows and ``VACUUM ANALYZE``d + (an Index Only Scan additionally needs a set visibility map, which + freshly-inserted, never-vacuumed rows do not have) so the plan is not a + tiny-corpus degenerate choice -- see + ``test_query_compiler.py::_explain_plan``'s docstring for that failure mode. + + ``VACUUM`` cannot run inside a transaction block, and SQLAlchemy 2.0 + autobegins one on every ``execute`` (a preceding ``conn.commit()`` does not + help -- the next ``execute`` autobegins again), so it runs on a SEPARATE + connection with ``execution_options(isolation_level="AUTOCOMMIT")``, never + on this module's transaction-bound ``conn`` fixture. + """ + items = _items(*[_fn("f", i) for i in range(200)]) + _index_default(conn, name="acme/widgets", head_sha="sha_first", items=items) + repo_id = int( + conn.execute(text("SELECT id FROM repos WHERE name = 'acme/widgets'")).scalar_one() + ) + conn.commit() + + engine = conn.engine + with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as vac_conn: + vac_conn.execute(text(f"SET search_path TO {SCHEMA}, public")) + vac_conn.execute(text("VACUUM ANALYZE files")) + + plan_a = _explain( + conn, + "SELECT path, content_sha FROM files " + "WHERE repo_id = :repo_id AND branches @> CAST(:branch_arr AS text[])", + {"repo_id": repo_id, "branch_arr": ["main"]}, + ) + a_nodes = list(_plan_nodes(plan_a)) + assert any(n.get("Index Name") == "ix_files_branches_gin" for n in a_nodes), a_nodes + assert not any(n.get("Node Type") == "Seq Scan" for n in a_nodes), a_nodes + a_output = " ".join(plan_a.get("Output") or []) + assert "content" not in a_output and "files.content" not in a_output, plan_a + + plan_b = _explain( + conn, + "SELECT path, content_sha FROM files WHERE repo_id = :repo_id", + {"repo_id": repo_id}, + analyze=True, + ) + b_nodes = list(_plan_nodes(plan_b)) + assert not any(n.get("Node Type") == "Seq Scan" for n in b_nodes), b_nodes + io_scan = next((n for n in b_nodes if n.get("Index Name") == "uq_files_repo_path_sha"), None) + assert io_scan is not None, b_nodes + assert io_scan.get("Node Type") == "Index Only Scan", io_scan + assert io_scan.get("Heap Fetches") == 0, io_scan + # The two expensive mistakes read_repo_content_shas' docstring names: the + # Index Only Scan's own output list proves neither `content` nor `branches` + # is ever fetched -- content_sha/path/repo_id are the constraint's own + # columns, so an Index Only Scan is fundamentally incapable of returning + # anything else. + io_output = " ".join(io_scan.get("Output") or []) + assert "content" not in io_output + assert "branches" not in io_output From 021162058f71e2c1a252e7a94b485e1a0cbc72d8 Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 07:51:59 -0700 Subject: [PATCH 8/9] docs: delta indexing in the parallelism runbook, the bump obligation, and the scope note MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - runbook §2: the new indexer.store `delta write set ...` line and how to read it. - runbook §4: correct the force-reindex SQL from `repos` to `repo_branches` -- the skip seam reads repo_branches only, so the documented `UPDATE repos SET index_semantics_version = NULL` has always been a no-op against it. New §4.1 documents the two accepted delta regressions (a degraded branch no longer self-heals its chunk coverage; the chunk cap is now enforced per run) and their remedies. §5: extends the INDEX_SEMANTICS_VERSION bump obligation to the embedding model and SEMANTIC_EMBEDDING_DIM, neither of which the tripwire watches. - app/config.py: comment on semantic_max_chunks_per_repo recording its new per-run scope. - app/db/models.py: INDEX_SEMANTICS_VERSION docstring extended to match; no version bump (indexer/languages.py and the other two tripwire-watched files are untouched by this branch). --- app/config.py | 7 ++ app/db/models.py | 16 +++- docs/runbooks/indexing-parallelism.md | 123 +++++++++++++++++++++++++- 3 files changed, 142 insertions(+), 4 deletions(-) diff --git a/app/config.py b/app/config.py index e9900f8..e6bb63a 100644 --- a/app/config.py +++ b/app/config.py @@ -99,6 +99,13 @@ class Settings(BaseSettings): # this loud check could ever fire, which would defeat the point of having a ceiling. # A repo that legitimately exceeds this needs a temp-table staging path, not a bigger # buffer. + # + # Scope note (#104): under file-level delta indexing this cap is enforced against + # whatever ONE RUN embeds (changed/new + membership-only files), not a branch's whole + # corpus -- a branch can legitimately drift above this number between full reindexes + # (a semantics bump, or its first index), re-enforced in full at each of those. This is + # a deliberate, accepted trade-off (see indexer/job.py's module docstring and + # docs/runbooks/indexing-parallelism.md §4.1), not a bug; the constant is unchanged. semantic_max_chunks_per_repo: int = 8000 # Chunk size bound (tokens) fed to the embedding model. Distinct from MAX_FILE_BYTES, diff --git a/app/db/models.py b/app/db/models.py index 45c7e1e..3278d1c 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -45,7 +45,21 @@ ``indexer/symbols.py``, to ``indexer/parse.py``'s chunking, or to ``indexer/languages.py``'s extraction contract. A bump forces every repo to re-index once, because a repo's stored ``repos.index_semantics_version`` no -longer matches. The CI tripwire enforces the bump obligation. +longer matches. The CI tripwire enforces the bump obligation for those three +files. + +**The obligation extends past the tripwire's reach (#104).** Swapping the +embedding MODEL (``app/embed.py``) or changing ``SEMANTIC_EMBEDDING_DIM`` +(``app/config.py``) also requires a bump, but the tripwire does not watch +either file (``app/embed.py`` deliberately -- it would otherwise fire on +unrelated retry/batching edits) -- this is a reviewed convention, not a +machine-enforced one. Before file-level delta indexing (issue #104) a missed +bump here was self-limiting: the next HEAD move re-embedded a branch's whole +corpus regardless. Under delta indexing only a CHANGED file re-embeds, so a +missed bump now leaves every unchanged file's vectors silently stale forever +-- exactly the failure this version column exists to prevent, and precisely +why version ``2`` was minted (turning semantic search on by default so +``chunks`` backfills) is precedent, not a special case. Migrations must never import this constant -- see ``app/alembic/versions/0002_index_semantics_version.py``. diff --git a/docs/runbooks/indexing-parallelism.md b/docs/runbooks/indexing-parallelism.md index fed089c..79c3e5e 100644 --- a/docs/runbooks/indexing-parallelism.md +++ b/docs/runbooks/indexing-parallelism.md @@ -154,6 +154,16 @@ you write keeps working. Read the largest field; that is the branch's bottleneck | `db` | per-file round trips | #105 (batched writes) | | any of the above, on **unchanged** content | redundant work | #104 (file-level delta indexing) | +`#104` narrows the **db** and **embed** costs to a branch's actual delta, not +its size — but it does NOT touch `parse`: extraction still runs on every +file every run (tree-sitter must produce a `FileExtraction` before +`index_repo` can classify it), so an all-unchanged branch on a large repo +still pays its full `download`+`extract`+`parse` cost. See `indexer.store`'s +`delta write set …` line (below) to tell "this branch is genuinely mostly-new" +from "this branch is mostly-unchanged but still parsing everything" — the +latter is exactly the case `#108` (process-pool extraction) or a future +extraction-skip step would address next. + Four fields need interpretation before you act on them: - **`resolve=0.00s` on a default branch is expected, not a bug.** That branch's @@ -184,6 +194,36 @@ Four fields need interpretation before you act on them: instrumentation merely makes visible for the first time — it is not a new regression. +### 2.2 The delta write set line (#104) + +Every `index_repo` call also emits one `indexer.store` INFO line, immediately +before the sweep, in **both** the gate-open and gate-closed cases — one format +string, no conditional fields, so it stays greppable either way: + +``` +INFO indexer.store [acme/widgets]: acme/widgets@main: delta write set 412/30214 files (unchanged=29790 membership=12, semantics gate open) +INFO indexer.store [acme/gadgets]: acme/gadgets@main: delta write set 812/812 files (unchanged=0 membership=0, semantics gate closed: stored v3 != v4) +``` + +`unchanged` files write nothing at all (no file upsert, no symbol/edge +delete-reinsert, no chunk write) and are never re-embedded. `membership` files +are already stored under another branch and only need their `branches` array +unioned in, plus a chunk write if semantic is on (see §4's accepted +regressions). The leading fraction is `(changed/new) / (total seen)`. The gate +is per-BRANCH: it opens only once that branch's own `repo_branches` stamp is +at the current `INDEX_SEMANTICS_VERSION` — a branch's first run, or any run +after a semantics bump, always shows `semantics gate closed`. + +``` +grep 'delta write set' run.log # one line per index_repo call +``` + +A branch stuck at a low `unchanged=` fraction run after run either genuinely +churns every run (nothing to fix) or has drifted out of delta eligibility — +check its `repo_branches.index_semantics_version` against the current +`INDEX_SEMANTICS_VERSION` and whether a sibling branch is stale (§4's +provenance gate). + --- ## 3. The three limits, and why raising concurrency is a bad trade @@ -265,17 +305,79 @@ There is deliberately **no `--force_reindex` flag.** Forcing a re-index means clearing the provenance stamp, after which the normal skip logic re-indexes the affected repos on the next scheduled or manual run. +**The stamp the skip seam actually reads is `repo_branches`, not `repos`.** +`indexer/job.py`'s `_read_stamps` selects +`RepoBranch.last_indexed_commit, RepoBranch.index_semantics_version` — the +`repos` table's `index_semantics_version` column is a deprecated legacy stamp +that no decision anywhere reads (`app/db/models.py` documents it write-only). +An `UPDATE repos SET index_semantics_version = NULL` is therefore a **no-op** +against the skip seam: the branch will look untouched and re-index on its own +next scheduled cycle, not immediately, and the operator following an older +version of this runbook would see nothing happen. + ```sql -- everything -UPDATE repos SET index_semantics_version = NULL; +UPDATE repo_branches SET index_semantics_version = NULL; + +-- one repo, every branch +UPDATE repo_branches SET index_semantics_version = NULL + WHERE repo_id = (SELECT id FROM repos WHERE name = 'acme/widgets'); --- one repo -UPDATE repos SET index_semantics_version = NULL WHERE name = 'acme/widgets'; +-- one repo, one branch +UPDATE repo_branches SET index_semantics_version = NULL + WHERE repo_id = (SELECT id FROM repos WHERE name = 'acme/widgets') + AND branch = 'main'; ``` Then run the job (`make index TARGET=` or `databricks bundle run code_search_index -t `). +### 4.1 File-level delta indexing (#104): what changes about this remedy + +Once a branch's `repo_branches.index_semantics_version` matches the current +`INDEX_SEMANTICS_VERSION`, `index_repo` skips rewriting any file whose +`(path, content_sha)` it already has stored for that branch — see +`indexer.store`'s module docstring for the full classification and the +correctness proof. Two consequences change what "clear the stamp" actually +buys you: + +**A degraded branch no longer self-heals on its own.** Before #104, ANY +re-index rewrote the whole branch, so a branch whose semantic precompute +failed (a chunk-cap breach, an embedder outage) caught its chunks up +automatically on the next successful run. Under delta indexing, only +*changed* files get re-embedded — a branch that never changes again carries +that gap **forever** unless you clear its stamp. `indexer.job` emits one +run-completion WARNING naming every branch that finished this way: + +``` +WARNING indexer.job [-]: 2 branch(es) finished with degraded semantic coverage this run (chunk precompute failed; core index is current, chunks are not, and delta indexing will NOT catch them up on their own -- clear their repo_branches.index_semantics_version stamp to force a full re-embed, see docs/runbooks/indexing-parallelism.md §4): acme/big-repo@main, acme/other@release +``` + +Grep for it (`grep 'degraded semantic coverage' run.log`) and clear the named +branches' stamps with the one-branch form above once the underlying cause +(chunk cap, embedder outage) is resolved. + +**The provenance gate can force a full re-index you did not ask for.** A +branch only takes the cheaper "membership-only" path (acquiring content a +*sibling* branch already stored, e.g. two branches sharing most of a +monorepo) when **every** `repo_branches` row for that repo is at the current +semantics version. If you clear one branch's stamp and leave siblings +untouched, that is fine — but a repo with one branch stuck at an old version +for any other reason (a persistently failing branch) will force every OTHER +branch of that repo through the full write path for any file it shares with +the stuck one, even though those branches are otherwise fully caught up. The +`delta write set …` line's `membership=` count going to zero across a whole +repo, with `unchanged=` still high, is the symptom — check for a sibling +branch stuck at a stale `index_semantics_version` before assuming something +is broken. + +**`semantic_max_chunks_per_repo` is enforced per RUN, not per branch's whole +corpus.** The cap is evaluated over whatever `_precompute_chunk_writer` +embeds, which under delta indexing is only the changed/new/membership-only +files. A branch can drift above the nominal cap between full reindexes (a +semantics bump, or its first index) — re-enforced in full at each of those. +Not a bug; see `app/config.py`'s `semantic_max_chunks_per_repo` comment. + ### Who can run this — read before you need it `UPDATE` on `repos` is held by **the identity that deployed the schema**, which @@ -305,6 +407,21 @@ If you change **what** gets extracted — `indexer/symbols.py`, `indexer/parse.py`, `indexer/languages.py` — you **must** bump `INDEX_SEMANTICS_VERSION` in `app/db/models.py`. +**The same obligation now extends past the tripwire's watched files (#104).** +`indexer/parse.py`'s chunker is already a watched path, so a change to +`iter_chunks` still fires the tripwire. Swapping the embedding MODEL +(`app/embed.py`) or changing `SEMANTIC_EMBEDDING_DIM` (`app/config.py`) +without bumping `INDEX_SEMANTICS_VERSION` is NOT caught by the tripwire (both +are deliberately unwatched — `app/embed.py` would otherwise fire on +unrelated retry/batching edits and a noisy tripwire gets disabled) and now +leaves every UNCHANGED file's vectors permanently stale under file-level +delta indexing — before #104 the next HEAD move re-embedded everything +anyway, so a missed bump here was self-limiting; it no longer is. This is not +a new pattern: `INDEX_SEMANTICS_VERSION` version `2` was minted for exactly +this reason (turning semantic search on by default so `chunks` backfills). +Treat this as a reviewed convention, the same posture `indexer.store`'s +module docstring takes for `lang`/`size` re-derivation. + Without a bump, every already-indexed repo keeps serving output from the *old* extractor and never re-indexes, because its stored stamp still matches HEAD. The failure is silent and open-ended: the index looks perfectly current. From 975117185fe735ca9e6750fa43ec510f8711c4eb Mon Sep 17 00:00:00 2001 From: IceRhymers Date: Sat, 25 Jul 2026 08:08:38 -0700 Subject: [PATCH 9/9] Address review pass 1: neutered test, vacuous EXPLAIN assertions, doc gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An independent code-reviewer pass (separate context) found no CRITICAL/HIGH issues but several real MEDIUM/LOW findings, all fixed here: - test_reindex_with_identical_items_does_not_duplicate_edges was silently neutered by the delta gate: run 2 re-indexed byte-identical content at the same head_sha, so main.py classified unchanged and the edge write path was never exercised at all -- the assertion passed for the wrong reason. Same §2.6a (b) treatment as the two guard tests: force the gate closed. - The EXPLAIN test's projected-columns assertions were vacuously true: EXPLAIN (FORMAT JSON) omits the `Output` field entirely without VERBOSE, so `"content" not in ""` always passed. Added VERBOSE and switched to an exact per-column check (a naive substring check would false-positive on `content_sha` containing `content`). - Re-enabling the indexer job's `semantic.enabled` flag after a period disabled no longer backfills chunks on its own under delta indexing (same root cause as a degraded-precompute branch); documented in semantic-enablement.md with the stamp-clearing remedy. - Runbook "Who can run this" still named `repos`; the statement above it was already corrected to `repo_branches`. - Local-coupling nit in the membership-only classification (`delta_on and membership_ok`, rather than relying on membership_ok's non-local initialization); a misattributed comment in the empty-seen-set-guard invariant test; a garbled sentence in the INDEX_SEMANTICS_VERSION docstring; a dangling indexer/AGENTS.md cross-reference; a docstring note on the advisory read's discarded `present` set and its cost. make lint, make test, and make test-integration all re-run clean after these changes (204 passed / 8 failed / 43 errors against local Postgres, same named failures as the pre-#104 baseline -- all Lakebase-only). --- app/db/models.py | 7 +++--- docs/runbooks/indexing-parallelism.md | 4 ++-- docs/runbooks/semantic-enablement.md | 16 ++++++++++++++ indexer/job.py | 14 ++++++++++++ indexer/store.py | 2 +- tests/integration/test_store.py | 16 +++++++++++--- tests/integration/test_store_delta.py | 32 +++++++++++++++++++-------- 7 files changed, 73 insertions(+), 18 deletions(-) diff --git a/app/db/models.py b/app/db/models.py index 3278d1c..af616db 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -57,9 +57,10 @@ bump here was self-limiting: the next HEAD move re-embedded a branch's whole corpus regardless. Under delta indexing only a CHANGED file re-embeds, so a missed bump now leaves every unchanged file's vectors silently stale forever --- exactly the failure this version column exists to prevent, and precisely -why version ``2`` was minted (turning semantic search on by default so -``chunks`` backfills) is precedent, not a special case. +-- exactly the failure this version column exists to prevent. This is not a +new kind of case: version ``2`` was minted for precisely this reason (turning +semantic search on by default, so every already-indexed branch had to +re-index once for ``chunks`` to backfill). Migrations must never import this constant -- see ``app/alembic/versions/0002_index_semantics_version.py``. diff --git a/docs/runbooks/indexing-parallelism.md b/docs/runbooks/indexing-parallelism.md index 79c3e5e..9de234b 100644 --- a/docs/runbooks/indexing-parallelism.md +++ b/docs/runbooks/indexing-parallelism.md @@ -380,8 +380,8 @@ Not a bug; see `app/config.py`'s `semantic_max_chunks_per_repo` comment. ### Who can run this — read before you need it -`UPDATE` on `repos` is held by **the identity that deployed the schema**, which -owns the tables. Concretely: +`UPDATE` on `repo_branches` (and `repos`) is held by **the identity that deployed +the schema**, which owns every table, `repo_branches` included. Concretely: - **dev:** the developer who ran `make migrate` / `scripts/deploy.sh`. Table ownership carries `UPDATE` implicitly; no explicit grant was ever issued for diff --git a/docs/runbooks/semantic-enablement.md b/docs/runbooks/semantic-enablement.md index ee195e0..51ba874 100644 --- a/docs/runbooks/semantic-enablement.md +++ b/docs/runbooks/semantic-enablement.md @@ -85,6 +85,22 @@ which makes the job a true semantic no-op (no embedder built, no chunking, the environment. Precedence for the job is `config.yaml > CODE_SEARCH_* env > default`, so `semantic.enabled: false` wins even if the env says enabled. +**Re-enabling the job's semantic flag does not backfill on its own (#104).** This +runbook's §2 above promises that an `INDEX_SEMANTICS_VERSION` bump "forces every +already-indexed branch to re-index once … which backfills `chunks`" — true for a +version bump, but **not** for flipping `semantic.enabled` back to `true` after a +period disabled. Under file-level delta indexing a branch whose stamp is already at +the current `INDEX_SEMANTICS_VERSION` classifies every unchanged file as unchanged +and skips embedding it, with no awareness that this run is the first to have an +embedder at all. A branch that never changes again after re-enabling never gets +chunks. Clear that branch's stamp to force the backfill (same remedy as a degraded +branch — see `docs/runbooks/indexing-parallelism.md` §4.1): + +```sql +UPDATE repo_branches SET index_semantics_version = NULL + WHERE repo_id = (SELECT id FROM repos WHERE name = 'acme/widgets'); +``` + **All three surfaces, not just one:** the flag must be off on the MCP app, the webui app (both via env), **and** the indexer job (via `config.yaml`) — each has its own config source. The webui SPA's Semantic tab is driven entirely by diff --git a/indexer/job.py b/indexer/job.py index 11a5fe4..704209f 100644 --- a/indexer/job.py +++ b/indexer/job.py @@ -1220,6 +1220,20 @@ def _index_one_branch( # every file this branch does not already carry. A short- # lived connection, closed before embedding starts -- # never held across the embedder's network I/O. + # + # `_present` is discarded -- this module only ever needs + # `carried` (job.py cannot replicate index_repo's + # provenance-gate check anyway, and doesn't need to: any + # not-carried file gets embedded here regardless of + # whether index_repo later classifies it membership-only + # or changed/new). shas_fn still computes and returns the + # full-repo `present` set -- read_repo_content_shas' + # signature is deliberately ONE shared query pair with + # index_repo's authoritative read (see its docstring), so + # this module pays for a second full-repo Index Only Scan + # it doesn't use rather than forking the query. That cost + # rides inside `embed=` on the phase timing line (see the + # comment above), not broken out separately. with engine.connect() as shas_conn: carried, _present = shas_fn(shas_conn, name=name, branch=branch) files_to_embed = [ diff --git a/indexer/store.py b/indexer/store.py index af6f308..17f8c63 100644 --- a/indexer/store.py +++ b/indexer/store.py @@ -332,7 +332,7 @@ class is unioned in by ONE batched ``UPDATE ... RETURNING`` after the unchanged_count += 1 continue - if membership_ok and (pf.path, sha) in present: + if delta_on and membership_ok and (pf.path, sha) in present: # Membership-only: the row exists (written by another branch) but # does not carry this branch yet. Statement 4 has proven every # branch of this repo is at the current semantics version, so its diff --git a/tests/integration/test_store.py b/tests/integration/test_store.py index 3d83bc4..0cd4af1 100644 --- a/tests/integration/test_store.py +++ b/tests/integration/test_store.py @@ -312,8 +312,8 @@ def test_reindex_replaces_stale_edges_for_the_same_file(conn: Connection) -> Non index_semantics_version = NULL`` idiom as test_legacy_null_semantics_version_is_rewritten) is faithful to production, not a workaround. This is the "unconditional-delete guard" and it must keep - its ORIGINAL assertion, not a relaxed one -- see indexer/AGENTS.md / - the plan for issue #104, §2.6a. + its ORIGINAL assertion, not a relaxed one -- see the plan for issue #104, + §2.6a. """ symbol = ExtractedSymbol("f", "function", 1, 3) content = "def f():\n target()\n return 1\n" @@ -351,6 +351,15 @@ def test_reindex_replaces_stale_edges_for_the_same_file(conn: Connection) -> Non @pytest.mark.integration def test_reindex_with_identical_items_does_not_duplicate_edges(conn: Connection) -> None: + """The delete-before-insert idempotency of the edge writer, proven by forcing + the delta gate CLOSED (the same idiom as the two guard tests above) so the + second run genuinely re-executes the write path rather than classifying + main.py unchanged and skipping it -- which would make the `== 1` assertion + pass vacuously (nothing touched at all) rather than proving delete-then- + insert doesn't duplicate. Without this the first run's first-index-ever + baseline (version None) makes run 2's baseline INDEX_SEMANTICS_VERSION, and + the delta gate would otherwise open and skip main.py entirely. + """ symbol = ExtractedSymbol("f", "function", 1, 3) item = ( "main.py", @@ -359,7 +368,8 @@ def test_reindex_with_identical_items_does_not_duplicate_edges(conn: Connection) [ExtractedEdge(kind="call", target="helper", line=2, enclosing=symbol)], ) _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(item)) - conn.rollback() + conn.execute(text("UPDATE repo_branches SET index_semantics_version = NULL")) + conn.commit() _index_default(conn, name="acme/widgets", head_sha="sha_first", items=_items(item)) file_id = conn.execute(text("SELECT id FROM files WHERE path = 'main.py'")).scalar_one() diff --git a/tests/integration/test_store_delta.py b/tests/integration/test_store_delta.py index c10ca22..08e528a 100644 --- a/tests/integration/test_store_delta.py +++ b/tests/integration/test_store_delta.py @@ -402,10 +402,14 @@ def test_no_files_row_ever_has_an_empty_branches_array(conn: Connection) -> None items=_items(UTIL), ) conn.rollback() - # 'a' drops both files -- util.py loses only 'a' (shared with 'b'), main.py - # loses its only branch and must be deleted outright. + # Empty-seen-set guard fires (see store.py's _sweep_membership): a NO-OP, + # not a drop. Included to prove the guard doesn't itself leave or create an + # empty-branches row. index_repo(conn, name="acme/widgets", branch="a", is_default=True, head_sha="sha_a2", items=[]) conn.rollback() + # 'a' now genuinely drops both files (UTIL is the only file it still + # parses): util.py loses only 'a' (row survives, still shared with 'b'); + # main.py loses its only remaining branch and is deleted outright. index_repo( conn, name="acme/widgets", @@ -415,6 +419,7 @@ def test_no_files_row_ever_has_an_empty_branches_array(conn: Connection) -> None items=_items(UTIL), ) conn.rollback() + # Empty-seen-set guard again, now that 'a' carries nothing -- still a no-op. index_repo(conn, name="acme/widgets", branch="a", is_default=True, head_sha="sha_a4", items=[]) assert _count(conn, "files", "cardinality(branches) = 0") == 0 @@ -561,8 +566,11 @@ def _explain( """``EXPLAIN (FORMAT JSON)`` for ``sql`` with the seq-scan escape hatch disabled, scoped to a SAVEPOINT so the ``enable_seqscan`` GUC change never leaks past this call (clones ``tests/integration/test_query_compiler.py``'s ``_explain_plan`` - idiom, per ``tests/integration/AGENTS.md``).""" - mode = "ANALYZE, FORMAT JSON" if analyze else "FORMAT JSON" + idiom). ``VERBOSE`` is load-bearing, not decoration: Postgres only emits each + node's ``Output`` column list under ``VERBOSE``, and the projected-columns + assertions below depend on that field actually being present rather than + silently absent (which would make them vacuously true).""" + mode = "VERBOSE, ANALYZE, FORMAT JSON" if analyze else "VERBOSE, FORMAT JSON" savepoint = conn.begin_nested() try: conn.execute(text("SET LOCAL enable_seqscan = off")) @@ -574,6 +582,14 @@ def _explain( return plan +def _projects_column(output: list[str] | None, column: str) -> bool: + """Does EXPLAIN VERBOSE's ``Output`` list include ``column``, exactly or + table-qualified (``files.content``)? A plain substring check would false- + positive on ``content_sha`` containing ``content`` -- this checks whole + output entries instead.""" + return any(entry == column or entry.endswith(f".{column}") for entry in output or []) + + def _plan_nodes(plan: dict[str, Any]) -> Iterator[dict[str, Any]]: yield plan for child in plan.get("Plans", []) or []: @@ -619,8 +635,7 @@ def test_the_delta_pre_read_is_index_served(conn: Connection) -> None: a_nodes = list(_plan_nodes(plan_a)) assert any(n.get("Index Name") == "ix_files_branches_gin" for n in a_nodes), a_nodes assert not any(n.get("Node Type") == "Seq Scan" for n in a_nodes), a_nodes - a_output = " ".join(plan_a.get("Output") or []) - assert "content" not in a_output and "files.content" not in a_output, plan_a + assert not _projects_column(plan_a.get("Output"), "content"), plan_a plan_b = _explain( conn, @@ -639,6 +654,5 @@ def test_the_delta_pre_read_is_index_served(conn: Connection) -> None: # is ever fetched -- content_sha/path/repo_id are the constraint's own # columns, so an Index Only Scan is fundamentally incapable of returning # anything else. - io_output = " ".join(io_scan.get("Output") or []) - assert "content" not in io_output - assert "branches" not in io_output + assert not _projects_column(io_scan.get("Output"), "content"), io_scan + assert not _projects_column(io_scan.get("Output"), "branches"), io_scan