Add incremental branch indexing for fast-moving branches

AGENTS.md

@@ -32,6 +32,7 @@ config is a YAML list, one entry per repo. See `repos.example.yml`.
 | `match` | yes | For `branch`/`tag`: pattern string or list of patterns matched against ref names (version DSL + glob) — a ref matches if any pattern hits. For `commit`: a commit SHA/prefix string or list of them (see below). |
 | `since` | no | Index-side inclusion floor: the earliest commit to start indexing from. See below. Not valid for `type: commit`. |
 | `retain` | no | Retention policy (see below). Omit to keep forever. For `type: commit`, only `age` is valid. |
+| `update` | no | `snapshot` (default) or `incremental`. `incremental` opts a `type: branch` selector into the isolated v2 incremental path (see below); it cannot be combined with `since` or `retain`. |
 
 #### `type: commit` (pinning an explicit commit)
 
@@ -155,6 +156,84 @@ Duration format (for `age`/`since.age`): `<n><unit>` where unit is `s` (seconds)
 
 Indexing is idempotent — re-running only indexes refs that are new or have moved.
 
+### Incremental branch indexing (`update: incremental`)
+
+By default a branch selector is a **snapshot** selector: each new commit produces a complete,
+immutable, commit-addressed file/line snapshot in the `sourcerer-v1-*` indices. That keeps
+coherent history, but the indexing cost scales with the whole repository on every move.
+
+`update: incremental` opts a `type: branch` selector into an isolated **v2** evaluation path
+for moving branches that need frequent, cheap refreshes:
+
+```yaml
+- org: elastic
+  repo: elasticsearch
+  refs:
+  - type: branch
+    match: main
+    update: incremental      # v2 incremental path; cannot combine with since/retain
+```
+
+How it behaves:
+
+- **Isolated schema.** Incremental content lives in `sourcerer-v2-files~<org>~<repo>` and
+  `sourcerer-v2-lines~<org>~<repo>`, with one mutable lookup document per branch in
+  `sourcerer-v2-refs`. The `sourcerer-v1-*` snapshot indices are never touched. Content is
+  **ref-addressed** (`git.ref_key` / `git.ref`), not commit-addressed — no commit SHA in the
+  document id — so a branch keeps exactly one live view.
+- **First run rebuilds.** The initial incremental run (or any run where the previous completed
+  commit is no longer available locally, e.g. after a force-push) rebuilds the whole branch
+  namespace into v2. It never migrates existing v1 data.
+- **Changed-file updates.** Subsequent runs diff the last completed commit against the new
+  remote tip and touch only the changed paths: deleted/modified/rename-source paths have their
+  prior file and line docs deleted, then added/modified/rename-destination files are re-indexed.
+  For a Customer Zero profile that changes roughly **10–20 files** per update, the work is
+  proportional to those paths, not the whole repo.
+- **No branch history.** Only the current view is retained — there are no per-commit snapshots
+  for an incremental branch, which is why `retain` (nothing to trim) and `since` (no inclusion
+  floor) are rejected at config parse time.
+- **Temporary mixed revisions.** Elasticsearch is eventually consistent during an update by
+  explicit design. While an update runs, the branch's `sourcerer-v2-refs` document reports
+  `status: indexing`, `git.commit` stays at the last completed commit, and `git.target_commit`
+  advertises the candidate. Queries stay available throughout and may briefly return a **mixed
+  revision**; the completed `git.commit` used for citations only advances to `status: ready`
+  after all deletes + indexing + a content refresh succeed.
+- **Retry / fallback.** A failed update leaves `status: indexing` with the old completed commit
+  intact plus a bounded `error`/`failed_at`; the next run retries old→current and clears those
+  on success. A missing diff base falls back to a full branch rebuild rather than trusting an
+  empty diff.
+- **Agents query by ref key.** `sourcerer.refs.list` returns each ref's `update_mode`. For an
+  incremental branch, pass its exact `git_ref_key` (not `git_commit`) to the code/file tools;
+  they attach the completed commit via a `LOOKUP JOIN` on `sourcerer-v2-refs` for citations.
+
+> Requires an Elasticsearch/ES|QL version that supports `index.mode: lookup` and `LOOKUP JOIN`.
+>
+> **Upgrade note:** the content and refs tools now query both schemas and `LOOKUP JOIN`
+> `sourcerer-v2-refs`, so they depend on the v2 lookup index and the schema-anchor indices that
+> `sourcerer setup` creates. Re-run `sourcerer setup` after upgrading **before** relying on the
+> tools — `setup` creates those indices before (re)deploying the Agent Builder tools, so a
+> single `sourcerer setup` keeps the ordering correct; querying with the new tools against a
+> cluster that has not been set up will fail with an unknown-column / missing-index error.
+
+#### Local evaluation
+
+A repeatable way to measure the incremental win against a real cluster:
+
+1. `sourcerer setup` — loads the v1 and v2 index templates (idempotent; leaves v1 data alone).
+2. Index a branch once in incremental mode:
+   `sourcerer index --config repos.yml` with an `update: incremental` branch selector. This is
+   the full first-run rebuild — note the reported processed-file count and duration.
+3. Push a commit to that branch changing 10–20 files (add/modify/delete/rename).
+4. Re-run `sourcerer index --config repos.yml`. Compare the reported processed-file count and
+   duration to the first run — only the changed paths should be processed.
+5. Verify with the tools: `sourcerer.refs.list` shows one v2 refs doc for the branch with
+   `update_mode: incremental`, `status`, and the completed commit; query the changed content
+   with a code/file tool using the branch's `git_ref_key`, and confirm deleted/renamed paths
+   return nothing.
+
+Schedule incremental runs (e.g. via cron) at an interval comfortably longer than a single
+update's duration, so consecutive runs never overlap on the same branch.
+
 ### Clone cache
 
 `index` keeps each repo cloned under a persistent cache directory and refreshes it with

README.md

@@ -75,6 +75,14 @@ questions about your software using an agent that analyzes the code.
 Its value shines for questions that span multiple repositories or multiple
 versions of software.
 
+Branches default to immutable per-commit **snapshots**. For a fast-moving branch that needs
+frequent, cheap refreshes, a `type: branch` selector can opt into `update: incremental`, which
+maintains a single mutable branch view in isolated `sourcerer-v2-*` indices and re-indexes only
+the files changed since the last run (typically 10-20) instead of re-snapshotting the whole
+repo. See [Incremental branch indexing](AGENTS.md#incremental-branch-indexing-update-incremental)
+for the consistency contract (temporary mixed revisions during an update) and a local
+evaluation procedure.
+
 ## Philosophy
 
 **Code is the primary source of truth for its own behavior.** Always authoriative,

repos.example.yml

@@ -75,4 +75,20 @@
     match:
     - cfefb3b2378ccbadefa7  # full 40-char SHA also accepted
     retain:
-      age: 2y                # keep while within this age, prune older (or omit -> keep forever)
+      age: 2y                # keep while within this age, prune older (or omit -> keep forever)
+
+# Example (Customer Zero): keep a fast-moving branch refreshed with the isolated v2 incremental
+# path. `update: incremental` maintains ONE mutable branch view in the sourcerer-v2-* indices:
+# the first run rebuilds the branch, and each later run diffs the last completed commit against
+# the new tip and touches only the changed paths (typically 10-20 files) instead of
+# re-snapshotting the whole repo. There is no per-commit history, so it cannot be combined with
+# `since` or `retain`. During an update the branch stays queryable and may briefly return a
+# mixed revision until it reaches status ready. Requires an ES|QL version with index.mode:
+# lookup and LOOKUP JOIN. A given branch may be indexed in snapshot OR incremental mode, never
+# both, so this uses a dedicated repo entry rather than doubling up main above.
+- org: acme
+  repo: customer-zero
+  refs:
+  - type: branch
+    match: main
+    update: incremental

src/sourcerer/commands/index/command.py

@@ -27,20 +27,36 @@
 from ...progress import ProgressReporter, Unit, make_reporter
 from ...utils import ES_ERRORS, make_client
 from ..prune import command as prune_cmd
-from .documents import index_repo
+from .documents import index_paths_v2, index_repo
 from .git import (
     checkout_branch,
     checkout_ref,
     commit_date,
     count_tracked_files,
     default_branch,
+    iter_tracked_files,
+    plan_changes,
     prepared_repo,
     ref_dates,
     resolve_cache_root,
     resolve_commit,
     _rev_info,
 )
-from .markers import commit_fully_indexed, count_commit_docs, pre_clone_skip, should_index, write_ref_marker
+from .markers import (
+    commit_fully_indexed,
+    count_commit_docs,
+    count_v2_branch_docs,
+    delete_v2_branch,
+    delete_v2_paths,
+    pre_clone_skip,
+    read_v2_ref,
+    refresh_v2_content,
+    should_index,
+    write_ref_marker,
+    write_v2_failed,
+    write_v2_indexing,
+    write_v2_ready,
+)
 from .report import dry_run_config
 from .runtime import _aborted, _tuning, bulk_indexing_settings, handle_interrupts
 from .selection import _effective_since_floor, _load_config, _resolve_entry
@@ -121,6 +137,104 @@ def index_ref_in_dir(
     reporter.finish(unit, status, files_count, lines_count)
 
 
+def index_incremental_in_dir(
+    es: Elasticsearch,
+    org: str,
+    repo: str,
+    repo_dir,
+    branch: str,
+    force: bool = False,
+    reporter: ProgressReporter | None = None,
+    unit: Unit | None = None,
+) -> None:
+    """Incremental (v2) update of one branch into an already-cloned `repo_dir`. Entirely
+    separate from the snapshot path: it never consults v1 markers, `should_index`, or the
+    retention planner. The branch's single v2 refs document drives the decision:
+
+      * completed SHA == remote HEAD and status ready (and not --force)  -> no-op skip.
+      * no marker, no completed SHA, --force, or a missing diff base     -> full branch
+        reconciliation: delete the whole ref namespace, then index every tracked file.
+      * otherwise -> apply the Git change plan: synchronously delete prior docs for
+        deleted/modified/rename-source paths, then index the current destination paths.
+
+    The completed pointer advances only after deletes + indexing + content refresh all succeed
+    (INV-005/INV-008). Any Git or Elasticsearch failure records failure state (status stays
+    `indexing`, completed SHA unchanged, bounded error) and re-raises so the caller reports the
+    unit as failed without stopping the batch.
+    """
+    if reporter is None:
+        reporter = ProgressReporter()
+    if unit is None:
+        unit = Unit(org=org, repo=repo, ref=branch, kind="branch", update_mode="incremental")
+
+    reporter.set_stage(unit, "checkout")
+    checkout_branch(repo_dir, branch)
+    new_sha = resolve_commit(repo_dir)
+    commit_date_iso = commit_date(repo_dir)
+    unit.ref = branch
+
+    prior = read_v2_ref(es, org, repo, branch)
+    completed = prior.get("git", {}).get("commit") if prior else None
+    prior_status = prior.get("status") if prior else None
+
+    # No-op: the last completed commit already equals the current tip and the branch is ready.
+    if not force and prior is not None and prior_status == "ready" and completed == new_sha:
+        reporter.finish(unit, "skipped")
+        return
+
+    # Advertise the in-flight update: status -> indexing, completed pointer held at the old SHA,
+    # candidate exposed as target_commit. Readers stay unblocked (may see a brief mixed revision).
+    write_v2_indexing(es, org, repo, branch, completed_commit=completed,
+                      target_commit=new_sha, prior=prior)
+
+    try:
+        # Decide full reconciliation vs targeted change plan. --force, a first index, or an
+        # unavailable diff base all rebuild the whole namespace (never treat a missing base as
+        # an empty diff, INV-007).
+        plan = None
+        if not force and completed is not None:
+            candidate = plan_changes(repo_dir, completed, new_sha)
+            plan = None if candidate.base_missing else candidate
+
+        reporter.set_stage(unit, "indexing")
+
+        def on_progress(f: int, l: int) -> None:
+            reporter.update_counts(unit, f, l)
+
+        if plan is None:
+            delete_v2_branch(es, org, repo, branch)
+            paths = list(iter_tracked_files(repo_dir))
+            reporter.set_total_files(unit, len(paths))
+            processed_files, processed_lines = index_paths_v2(
+                es, org, repo, repo_dir, branch, paths, on_progress=on_progress,
+            )
+        else:
+            delete_v2_paths(es, org, repo, branch, plan.delete_paths)
+            reporter.set_total_files(unit, len(plan.index_paths))
+            processed_files, processed_lines = index_paths_v2(
+                es, org, repo, repo_dir, branch, plan.index_paths, on_progress=on_progress,
+            )
+
+        # Publication boundary: refresh content first, count the authoritative branch totals,
+        # then advance the completed pointer and refresh the refs index (INV-008).
+        refresh_v2_content(es, org, repo)
+        files_total, lines_total = count_v2_branch_docs(es, org, repo, branch)
+        write_v2_ready(es, org, repo, branch, new_sha, commit_date_iso, files_total, lines_total)
+    except KeyboardInterrupt:
+        # Aborted mid-update: leave the marker at `indexing` with the old completed SHA (already
+        # written above); the next run retries. Do not record it as a failure.
+        raise
+    except Exception as e:
+        try:
+            write_v2_failed(es, org, repo, branch, completed_commit=completed,
+                            target_commit=new_sha, error=str(e), prior=prior)
+        except Exception:
+            pass  # a secondary failure writing the failure marker must not mask the original
+        raise
+
+    reporter.finish(unit, "indexed", processed_files, processed_lines)
+
+
 def index_one(
     es: Elasticsearch,
     org: str,
@@ -314,10 +428,19 @@ def process_group(item: tuple[tuple[str, str], list[Unit]]) -> None:
             if _aborted.is_set():
                 return
             (org, repo), group = item
-            # 2a. Cheap per-ref skip for the whole group (no clone yet). A transient ES error
+            # Incremental (v2) branch units bypass the entire v1 pre-clone/skip/retention path:
+            # their no-op vs retry decision is made post-checkout from the v2 marker, so they
+            # always require the clone (unless the whole repo is snapshot-only and already
+            # indexed). Split them out first; the snapshot units keep the existing behaviour.
+            incremental_units = [u for u in group if u.update_mode == "incremental"]
+            snapshot_group = [u for u in group if u.update_mode != "incremental"]
+            for unit in incremental_units:
+                reporter.start(unit)
+
+            # 2a. Cheap per-ref skip for the snapshot units (no clone yet). A transient ES error
             # here fails just that ref (the skip check hits the cluster) and the batch goes on.
             pending: list[tuple[Unit, str | None, str | None, str | None]] = []
-            for unit in group:
+            for unit in snapshot_group:
                 if _aborted.is_set():
                     return
                 reporter.start(unit)
@@ -337,19 +460,22 @@ def process_group(item: tuple[tuple[str, str], list[Unit]]) -> None:
                 else:
                     pending.append((unit, branch, tag, commit))
 
-            if not pending:
+            if not pending and not incremental_units:
                 return  # whole repo already indexed -> no clone at all
 
             # 2b. Clone/fetch once, then check out and index each pending ref. A failure on one
             # ref -- git, bad value, or a transient ES timeout/connection drop -- is reported and
             # the remaining refs (and other repos) continue. If the persistent cache dir is locked
             # by another run, prepared_repo yields None and the whole repo is skipped this round.
             try:
-                reporter.set_stage(pending[0][0], "cloning")
+                clone_leader = pending[0][0] if pending else incremental_units[0]
+                reporter.set_stage(clone_leader, "cloning")
                 with prepared_repo(org, repo, cache_root, ephemeral) as repo_dir:
                     if repo_dir is None:
                         for unit, _branch, _tag, _commit in pending:
                             reporter.finish(unit, "locked", detail="another sourcerer run holds this repo's cache lock")
+                        for unit in incremental_units:
+                            reporter.finish(unit, "locked", detail="another sourcerer run holds this repo's cache lock")
                         return
                     # Reorder pending refs newest-first by creation date so more-recent refs
                     # are indexed first. creatordate is available now that the clone exists;
@@ -412,13 +538,36 @@ def process_group(item: tuple[tuple[str, str], list[Unit]]) -> None:
                             with failures_lock:
                                 failures += 1
                             reporter.finish(unit, "error", detail=str(e))
+
+                    # 2d. Incremental branch units, indexed against the same clone. Each is
+                    # fully self-contained (v2 marker + change plan); a Git/ES failure records
+                    # failure state inside index_incremental_in_dir and is reported here without
+                    # stopping the remaining refs or repos.
+                    for unit in incremental_units:
+                        if _aborted.is_set():
+                            break
+                        try:
+                            index_incremental_in_dir(
+                                es, org, repo, repo_dir, unit.ref, force, reporter, unit,
+                            )
+                        except KeyboardInterrupt:
+                            break
+                        except (FileNotFoundError, subprocess.CalledProcessError, ValueError, *ES_ERRORS) as e:
+                            with failures_lock:
+                                failures += 1
+                            reporter.finish(unit, "error", detail=str(e))
             except (FileNotFoundError, subprocess.CalledProcessError, ValueError) as e:
                 # Clone failed: fail every still-pending ref of this repo, continue others.
                 for unit, _branch, _tag, _commit in pending:
                     if unit.status is None:
                         with failures_lock:
                             failures += 1
                         reporter.finish(unit, "error", detail=str(e))
+                for unit in incremental_units:
+                    if unit.status is None:
+                        with failures_lock:
+                            failures += 1
+                        reporter.finish(unit, "error", detail=str(e))
 
         with bulk_indexing_settings(es), ThreadPoolExecutor(
             max_workers=max(1, _tuning().index_repo_concurrency)