diff --git a/AGENTS.md b/AGENTS.md
index 3a5d6ea..ee849f9 100644
--- a/AGENTS.md
+++ b/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 unified v1 incremental path (see below); it cannot be combined with `since` or `retain`. |
 
 #### `type: commit` (pinning an explicit commit)
 
@@ -155,6 +156,78 @@ 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 the unified **v1** schema
+for moving branches that need frequent, cheap refreshes:
+
+```yaml
+- org: elastic
+  repo: elasticsearch
+  refs:
+  - type: branch
+    match: main
+    update: incremental      # incremental path; cannot combine with since/retain
+```
+
+How it behaves:
+
+- **Unified schema.** Snapshot and incremental rows share the real `sourcerer-v1-files~*`,
+  `sourcerer-v1-lines~*`, and `sourcerer-v1-refs` indices. `update_mode` separates them.
+  Incremental content is **ref-addressed** (`git.ref_key` / `git.ref`), not commit-addressed —
+  no commit SHA appears in its 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 in v1. This POC expects a clean rebuild rather than an in-place migration.
+- **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-v1-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-v1-refs` for citations.
+
+> Requires an Elasticsearch/ES|QL version that supports `index.mode: lookup` and `LOOKUP JOIN`.
+> `sourcerer setup` creates only the real v1 refs lookup index; it creates no empty schema anchors.
+> This POC changes the v1 mappings, so rebuild existing Sourcerer indices before setup.
+
+#### Local evaluation
+
+A repeatable way to measure the incremental win against a real cluster:
+
+1. Rebuild existing POC indices, then run `sourcerer setup` to load the unified v1 templates.
+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 incremental 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
diff --git a/README.md b/README.md
index 1aa24c0..5b3ac55 100644
--- a/README.md
+++ b/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 the unified `sourcerer-v1-*` 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,
diff --git a/repos.example.yml b/repos.example.yml
index 215e87c..8decd98 100644
--- a/repos.example.yml
+++ b/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)
\ No newline at end of file
+      age: 2y                # keep while within this age, prune older (or omit -> keep forever)
+
+# Example (Customer Zero): keep a fast-moving branch refreshed in the unified v1 schema.
+# `update: incremental` maintains ONE mutable branch view in the sourcerer-v1-* 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
\ No newline at end of file
diff --git a/src/sourcerer/commands/index/command.py b/src/sourcerer/commands/index/command.py
index ebd4be9..33257a2 100644
--- a/src/sourcerer/commands/index/command.py
+++ b/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_incremental_paths, 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_incremental_branch_docs,
+    delete_incremental_branch,
+    delete_incremental_paths,
+    pre_clone_skip,
+    read_incremental_ref,
+    refresh_incremental_content,
+    should_index,
+    write_ref_marker,
+    write_incremental_failed,
+    write_incremental_indexing,
+    write_incremental_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 (incremental) 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 incremental 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_incremental_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_incremental_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_incremental_branch(es, org, repo, branch)
+            paths = list(iter_tracked_files(repo_dir))
+            reporter.set_total_files(unit, len(paths))
+            processed_files, processed_lines = index_incremental_paths(
+                es, org, repo, repo_dir, branch, paths, on_progress=on_progress,
+            )
+        else:
+            delete_incremental_paths(es, org, repo, branch, plan.delete_paths)
+            reporter.set_total_files(unit, len(plan.index_paths))
+            processed_files, processed_lines = index_incremental_paths(
+                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_incremental_content(es, org, repo)
+        files_total, lines_total = count_incremental_branch_docs(es, org, repo, branch)
+        write_incremental_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_incremental_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 (incremental) branch units bypass the entire v1 pre-clone/skip/retention path:
+            # their no-op vs retry decision is made post-checkout from the incremental 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,7 +460,7 @@ 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
@@ -345,11 +468,14 @@ def process_group(item: tuple[tuple[str, str], list[Unit]]) -> None:
             # 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,6 +538,24 @@ 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 (incremental 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:
@@ -419,6 +563,11 @@ def process_group(item: tuple[tuple[str, str], list[Unit]]) -> 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)
diff --git a/src/sourcerer/commands/index/documents.py b/src/sourcerer/commands/index/documents.py
index 734c688..24643d2 100644
--- a/src/sourcerer/commands/index/documents.py
+++ b/src/sourcerer/commands/index/documents.py
@@ -6,7 +6,7 @@
 # Standard packages
 import pathlib
 import signal
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Iterable, Iterator
 from concurrent.futures import ProcessPoolExecutor
 
 # Third-party packages
@@ -15,7 +15,7 @@
 
 # App packages
 from ...indices import files_index, lines_index
-from ...utils import make_doc_id
+from ...utils import build_ref_key, make_doc_id
 from .git import iter_tracked_files
 from .runtime import _aborted, _tuning
 
@@ -48,10 +48,11 @@ def build_file_doc(
     except OSError:
         size = abs_path.lstat().st_size
     doc = {
+        "update_mode": "snapshot",
         "git": {
-            "org": org,
-            "repo": repo,
-            "commit": commit_sha,
+            "org": org.lower(),
+            "repo": repo.lower(),
+            "commit": commit_sha.lower(),
         },
         "file": {
             "path": rel_path,
@@ -65,10 +66,87 @@ def build_file_doc(
     # Content identity is (org, repo, commit, path): the same blob reached via any ref
     # (branch/tag/commit) collapses to one doc. Branch is intentionally absent -- it lives
     # only in the refs index (see markers.write_ref_marker) since a branch moves.
-    _id = make_doc_id(org, repo, commit_sha, rel_path)
+    _id = make_doc_id(org.lower(), repo.lower(), commit_sha.lower(), rel_path)
     return _id, doc
 
 
+def build_incremental_file_doc(
+    org: str,
+    repo: str,
+    ref: str,
+    rel_path: str,
+    abs_path: pathlib.Path,
+) -> tuple[str, dict]:
+    """incremental file doc: ref-addressed, NOT commit-addressed. Identity is
+    (org, repo, "branch", ref, path) so the same branch/path is one stable doc across commits
+    (INV-003); the completed commit lives only in the incremental refs lookup document. org/repo are
+    normalized to lowercase for both the stored fields and the id, since the incremental mappings carry
+    no normalizer, while ref stays case-sensitive."""
+    org_l, repo_l = org.lower(), repo.lower()
+    p = pathlib.PurePosixPath(rel_path)
+    directory = "" if str(p.parent) == "." else str(p.parent)
+    extension = p.suffix.lstrip(".") or None
+    try:
+        size = abs_path.stat().st_size
+    except OSError:
+        size = abs_path.lstat().st_size
+    doc = {
+        "update_mode": "incremental",
+        "git": {
+            "org": org_l,
+            "repo": repo_l,
+            "ref_key": build_ref_key(org, repo, ref),
+            "ref": ref,
+            "ref_type": "branch",
+        },
+        "file": {
+            "path": rel_path,
+            "directory": directory,
+            "name": p.name,
+            "extension": extension,
+            "size": size,
+            "attributes": file_attributes(abs_path) or None,
+        },
+    }
+    _id = make_doc_id(org_l, repo_l, "branch", ref, rel_path)
+    return _id, doc
+
+
+def iter_incremental_line_docs(
+    org: str,
+    repo: str,
+    ref: str,
+    rel_path: str,
+    content: str,
+) -> Iterator[tuple[str, dict]]:
+    """incremental per-line docs: ref-addressed line identity
+    (org, repo, "branch", ref, path, line_number), with no commit in the id or source."""
+    org_l, repo_l = org.lower(), repo.lower()
+    ref_key = build_ref_key(org, repo, ref)
+    p = pathlib.PurePosixPath(rel_path)
+    directory = "" if str(p.parent) == "." else str(p.parent)
+    extension = p.suffix.lstrip(".") or None
+    base = {
+        "update_mode": "incremental",
+        "git": {
+            "org": org_l,
+            "repo": repo_l,
+            "ref_key": ref_key,
+            "ref": ref,
+            "ref_type": "branch",
+        },
+        "file": {
+            "path": rel_path,
+            "directory": directory,
+            "name": p.name,
+            "extension": extension,
+        },
+    }
+    for line_num, line_content in enumerate(content.splitlines(), start=1):
+        _id = make_doc_id(org_l, repo_l, "branch", ref, rel_path, str(line_num))
+        yield _id, {**base, "line": {"number": line_num, "content": line_content}}
+
+
 def iter_line_docs(
     org: str,
     repo: str,
@@ -80,10 +158,11 @@ def iter_line_docs(
     directory = "" if str(p.parent) == "." else str(p.parent)
     extension = p.suffix.lstrip(".") or None
     base = {
+        "update_mode": "snapshot",
         "git": {
-            "org": org,
-            "repo": repo,
-            "commit": commit_sha,
+            "org": org.lower(),
+            "repo": repo.lower(),
+            "commit": commit_sha.lower(),
         },
         "file": {
             "path": rel_path,
@@ -93,7 +172,7 @@ def iter_line_docs(
         }
     }
     for line_num, line_content in enumerate(content.splitlines(), start=1):
-        _id = make_doc_id(org, repo, commit_sha, rel_path, str(line_num))
+        _id = make_doc_id(org.lower(), repo.lower(), commit_sha.lower(), rel_path, str(line_num))
         yield _id, {**base, "line": {"number": line_num, "content": line_content}}
 
 
@@ -206,3 +285,82 @@ def generate_actions():
     if on_progress is not None:
         on_progress(files_count, lines_count)
     return files_count, lines_count
+
+
+def build_incremental_file_actions(
+    org: str, repo: str, ref: str, repo_dir: pathlib.Path, rel_path: str,
+) -> list[dict]:
+    """Bulk actions for one incremental path: its ref-addressed file doc plus a line doc
+    per line of text. Paths absent from the checked-out tree are rejected -- they yield no
+    actions -- so a stale diff entry can never create a phantom document (INV/Task 3). Binary
+    files (NUL in the first 8 KB) or unreadable files get only their file doc, mirroring v1."""
+    abs_path = repo_dir / rel_path
+    # A tracked file may be a regular file or a symlink; a broken symlink still exists as a
+    # tracked entry, so is_symlink() must be checked before exists() (which follows the link).
+    if not (abs_path.is_symlink() or abs_path.exists()):
+        return []
+    f_index = files_index(org, repo)
+    l_index = lines_index(org, repo)
+    file_id, file_doc = build_incremental_file_doc(org, repo, ref, rel_path, abs_path)
+    actions = [{"_index": f_index, "_id": file_id, "_source": file_doc}]
+    try:
+        raw = abs_path.read_bytes()
+    except OSError:
+        return actions
+    if b"\x00" in raw[:8192]:
+        return actions  # binary: file metadata only, no line docs
+    content = raw.decode("utf-8", errors="surrogateescape")
+    for line_id, line_doc in iter_incremental_line_docs(org, repo, ref, rel_path, content):
+        actions.append({"_index": l_index, "_id": line_id, "_source": line_doc})
+    return actions
+
+
+def index_incremental_paths(
+    es: Elasticsearch,
+    org: str,
+    repo: str,
+    repo_dir: pathlib.Path,
+    ref: str,
+    paths: Iterable[str],
+    on_progress: Callable[[int, int], None] | None = None,
+) -> tuple[int, int]:
+    """Index a supplied iterable of tracked paths into the incremental content indices for `ref`, rather
+    than walking the whole repository. Used for both the changed-path update (10-20 files) and
+    full branch reconciliation (all tracked files); the caller decides which paths to pass.
+    Returns (files_count, lines_count) of the docs written.
+
+    Doc generation is inline (not farmed to a process pool like v1's index_repo): the typical
+    incremental batch is tiny, and a full rebuild still streams lazily into parallel_bulk, whose
+    worker threads overlap the network round-trips. Deterministic ref-addressed ids mean a
+    re-index overwrites in place, so this is safe to retry."""
+    files_count = 0
+    lines_count = 0
+    f_index = files_index(org, repo)
+    t = _tuning()
+
+    def generate_actions() -> Iterator[dict]:
+        for rel_path in paths:
+            yield from build_incremental_file_actions(org, repo, ref, repo_dir, rel_path)
+
+    processed = 0
+    for _ok, info in es_parallel_bulk(
+        es,
+        generate_actions(),
+        thread_count=t.bulk_threads,
+        chunk_size=t.bulk_chunk_size,
+        max_chunk_bytes=t.bulk_max_bytes,
+        queue_size=t.bulk_queue_size,
+    ):
+        if _aborted.is_set():
+            raise KeyboardInterrupt
+        meta = next(iter(info.values())) if info else {}
+        if meta.get("_index") == f_index:
+            files_count += 1
+        else:
+            lines_count += 1
+        processed += 1
+        if on_progress is not None and processed % 1000 == 0:
+            on_progress(files_count, lines_count)
+    if on_progress is not None:
+        on_progress(files_count, lines_count)
+    return files_count, lines_count
diff --git a/src/sourcerer/commands/index/git.py b/src/sourcerer/commands/index/git.py
index 251e7cd..2adf41b 100644
--- a/src/sourcerer/commands/index/git.py
+++ b/src/sourcerer/commands/index/git.py
@@ -14,6 +14,7 @@
 import tempfile
 import time
 from collections.abc import Iterator
+from dataclasses import dataclass, field
 
 # App packages
 from ...queries import _parse_dt
@@ -279,6 +280,112 @@ def count_tracked_files(repo_dir: pathlib.Path) -> int:
     return sum(1 for _ in iter_tracked_files(repo_dir))
 
 
+@dataclass
+class ChangePlan:
+    """A pure plan for turning one incremental branch update (old commit -> new commit) into
+    Elasticsearch work. `delete_paths` are the prior file/line docs to remove synchronously;
+    `index_paths` are the current tree paths to (re)index. A modified or type-changed file
+    appears in BOTH (delete its stale docs, then re-index every current line). `base_missing`
+    is True when the old commit object is unavailable locally -- the caller must then fall back
+    to full branch-namespace reconciliation instead of trusting an empty diff (INV-007)."""
+
+    delete_paths: list[str] = field(default_factory=list)
+    index_paths: list[str] = field(default_factory=list)
+    base_missing: bool = False
+
+
+def base_commit_available(repo_dir: pathlib.Path, old_sha: str) -> bool:
+    """True if `old_sha` resolves to a commit object present in the local clone. Uses
+    `git cat-file -e <sha>^{commit}` so a tag or partial object still fails closed. A False
+    here means the diff base is gone (e.g. force-push, shallow clone) and the caller must
+    rebuild rather than treat the missing base as an empty diff."""
+    try:
+        subprocess.run(
+            ["git", "-C", str(repo_dir), "cat-file", "-e", f"{old_sha}^{{commit}}"],
+            check=True,
+            capture_output=True,
+        )
+    except (subprocess.CalledProcessError, OSError):
+        return False
+    return True
+
+
+def _dedupe(paths: list[str]) -> list[str]:
+    """Order-preserving de-duplication (dict keeps first-seen order)."""
+    return list(dict.fromkeys(paths))
+
+
+def _parse_name_status_z(raw: bytes) -> tuple[list[str], list[str]]:
+    """Parse `git diff --name-status -z` output into (delete_paths, index_paths).
+
+    The `-z` stream is a flat run of NUL-terminated tokens: a status token, then one path
+    (add/modify/delete/type-change) or two paths (rename/copy: old then new). Parsing by NUL
+    boundary -- never by whitespace -- preserves spaces, tabs, and unusual bytes in paths.
+    Mapping (rename detection is an optimization; delete+add of the same paths is equivalent):
+      A (add)          -> index new
+      M (modify)       -> delete + index (replace at file granularity)
+      T (type change)  -> delete + index
+      D (delete)       -> delete
+      R (rename)       -> delete old + index new
+      C (copy)         -> index new (source is unchanged, stays indexed)
+    An unrecognized status fails safe as delete + index of its path."""
+    tokens = raw.split(b"\x00")
+    # Each real token is NUL-terminated, so a well-formed stream ends in an empty tail token;
+    # drop trailing empties so a complete record's last path isn't misread as "present but
+    # empty" and a truncated record is detected by running past the end.
+    while tokens and tokens[-1] == b"":
+        tokens.pop()
+    delete_paths: list[str] = []
+    index_paths: list[str] = []
+    i = 0
+    n = len(tokens)
+    while i < n:
+        status = tokens[i]
+        if not status:
+            i += 1
+            continue
+        code = status.decode("utf-8", errors="surrogateescape")[0]
+        if code in ("R", "C"):
+            if i + 2 >= n:
+                break  # truncated record
+            old = tokens[i + 1].decode("utf-8", errors="surrogateescape")
+            new = tokens[i + 2].decode("utf-8", errors="surrogateescape")
+            i += 3
+            if code == "R":
+                delete_paths.append(old)
+            index_paths.append(new)
+        else:
+            if i + 1 >= n:
+                break  # truncated record
+            path = tokens[i + 1].decode("utf-8", errors="surrogateescape")
+            i += 2
+            if code == "A":
+                index_paths.append(path)
+            elif code == "D":
+                delete_paths.append(path)
+            else:  # M, T, or an unknown status -> replace the whole file
+                delete_paths.append(path)
+                index_paths.append(path)
+    return _dedupe(delete_paths), _dedupe(index_paths)
+
+
+def plan_changes(repo_dir: pathlib.Path, old_sha: str, new_sha: str) -> ChangePlan:
+    """Build the ChangePlan for advancing an incremental branch from `old_sha` to `new_sha`.
+    Returns a `base_missing` plan (no paths) when the old commit is unavailable locally; the
+    caller then rebuilds the branch namespace. Otherwise runs a NUL-delimited name-status diff
+    with rename (-M) and copy (-C) detection and maps each record (see _parse_name_status_z)."""
+    if not base_commit_available(repo_dir, old_sha):
+        return ChangePlan(base_missing=True)
+    result = subprocess.run(
+        ["git", "-C", str(repo_dir), "diff", "--name-status", "-z", "-M", "-C",
+         old_sha, new_sha],
+        check=True,
+        capture_output=True,
+    )
+    delete_paths, index_paths = _parse_name_status_z(result.stdout)
+    return ChangePlan(delete_paths=delete_paths, index_paths=index_paths)
+
+
 def _ls_remote(url: str, *patterns: str, flags: tuple[str, ...] = ()) -> str | None:
     """
     Run `git ls-remote` against a remote without cloning. The URL must precede the ref
diff --git a/src/sourcerer/commands/index/markers.py b/src/sourcerer/commands/index/markers.py
index 1d4cf72..0ffca33 100644
--- a/src/sourcerer/commands/index/markers.py
+++ b/src/sourcerer/commands/index/markers.py
@@ -11,8 +11,8 @@
 from elasticsearch import Elasticsearch, NotFoundError
 
 # App packages
-from ...indices import REFS_INDEX, files_index
-from ...utils import make_doc_id
+from ...indices import REFS_INDEX, files_index, lines_index
+from ...utils import build_ref_key, make_doc_id
 from .git import resolve_remote
 
 
@@ -24,7 +24,7 @@ def build_ref_id(org: str, repo: str, ref_type: str, ref: str, commit_sha: str)
     alone would collapse them and clobber one on the next run. Folding commit in makes a
     moving branch append a new marker per commit (the append-only history that count/age
     pruning needs), while an immutable tag re-hashes to the same id and stays idempotent."""
-    return make_doc_id(org, repo, ref_type, ref, commit_sha)
+    return make_doc_id("snapshot", org.lower(), repo.lower(), ref_type, ref, commit_sha.lower())
 
 
 def count_commit_docs(es: Elasticsearch, index: str, org: str, repo: str, commit_sha: str) -> int:
@@ -144,14 +144,18 @@ def write_ref_marker(
     ref_id = build_ref_id(org, repo, ref_type, ref, commit_sha)
     doc = {
         "git": {
-            "org": org,
-            "repo": repo,
+            "ref_key": make_doc_id(
+                "snapshot", org.lower(), repo.lower(), ref_type, ref, commit_sha.lower()
+            ),
+            "org": org.lower(),
+            "repo": repo.lower(),
             "ref": ref,
             "ref_type": ref_type,
-            "commit": commit_sha,
+            "commit": commit_sha.lower(),
             "commit_date": commit_date_iso,
         },
         "status": "complete",
+        "update_mode": "snapshot",
         "files_count": files_count,
         "lines_count": lines_count,
         "indexed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
@@ -198,6 +202,249 @@ def pre_clone_skip(
     return False, ref_for_id, remote_sha
 
 
+# --- incremental mutable branch markers ------------------------------------------------
+# The incremental refs index holds exactly ONE document per incremental branch (INV-004), keyed by a
+# stable id that folds in only (org, repo, "branch", ref) -- never the commit -- so successive
+# updates overwrite the same document in place. Its `status`/`git.commit`/`git.target_commit`
+# fields make the update window observable without blocking readers (INV-005/INV-008).
+
+ERROR_MAX_LEN = 2000  # bound stored failure text so a giant git/ES error can't bloat the doc
+
+
+def build_incremental_ref_id(org: str, repo: str, ref: str) -> str:
+    """Stable id of an incremental branch's single incremental refs document (INV-004). Normalized
+    org/repo lowercasing matches the content ids and ref-key so identity is consistent; the
+    branch name stays case-sensitive."""
+    return make_doc_id("incremental", org.lower(), repo.lower(), "branch", ref)
+
+
+def read_incremental_ref(es: Elasticsearch, org: str, repo: str, ref: str) -> dict | None:
+    """The branch's incremental refs document _source, or None if it has never been indexed. A
+    real-time GET, so it reflects the last write even without an index refresh."""
+    try:
+        return es.get(index=REFS_INDEX, id=build_incremental_ref_id(org, repo, ref))["_source"]
+    except NotFoundError:
+        return None
+
+
+def _now_iso() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def _build_incremental_ref_doc(
+    org: str,
+    repo: str,
+    ref: str,
+    *,
+    status: str,
+    commit: str | None,
+    target_commit: str | None = None,
+    commit_date_iso: str | None = None,
+    files_count: int = 0,
+    lines_count: int = 0,
+    indexed_at: str | None = None,
+    update_started_at: str | None = None,
+    failed_at: str | None = None,
+    error: str | None = None,
+) -> dict:
+    return {
+        "git": {
+            "ref_key": build_ref_key(org, repo, ref),
+            "org": org.lower(),
+            "repo": repo.lower(),
+            "ref": ref,
+            "ref_type": "branch",
+            "commit": commit,
+            "target_commit": target_commit,
+            "commit_date": commit_date_iso,
+        },
+        "status": status,
+        "update_mode": "incremental",
+        "files_count": files_count,
+        "lines_count": lines_count,
+        "indexed_at": indexed_at,
+        "update_started_at": update_started_at,
+        "failed_at": failed_at,
+        "error": error[:ERROR_MAX_LEN] if error else None,
+    }
+
+
+def write_incremental_indexing(
+    es: Elasticsearch,
+    org: str,
+    repo: str,
+    ref: str,
+    completed_commit: str | None,
+    target_commit: str,
+    prior: dict | None = None,
+    refresh: bool = False,
+) -> None:
+    """Publish `status: indexing`: the completed pointer (`git.commit`) stays at the LAST
+    completed SHA (or None on a first index) while `git.target_commit` advertises the candidate
+    SHA (INV-005). Prior counts/commit_date/indexed_at are carried so readers keep meaningful
+    metadata during the window. Not a publication boundary -- refresh defaults off; a real-time
+    GET still sees it on retry."""
+    prior = prior or {}
+    pg = prior.get("git", {})
+    doc = _build_incremental_ref_doc(
+        org, repo, ref,
+        status="indexing",
+        commit=completed_commit,
+        target_commit=target_commit,
+        commit_date_iso=pg.get("commit_date"),
+        files_count=prior.get("files_count", 0),
+        lines_count=prior.get("lines_count", 0),
+        indexed_at=prior.get("indexed_at"),
+        update_started_at=_now_iso(),
+        failed_at=prior.get("failed_at"),
+        error=prior.get("error"),
+    )
+    es.index(index=REFS_INDEX, id=build_incremental_ref_id(org, repo, ref), document=doc, refresh=refresh)
+
+
+def write_incremental_ready(
+    es: Elasticsearch,
+    org: str,
+    repo: str,
+    ref: str,
+    commit: str,
+    commit_date_iso: str | None,
+    files_count: int,
+    lines_count: int,
+    refresh: bool = True,
+) -> None:
+    """Publish `status: ready` at the NEW completed commit, clearing the target and any prior
+    failure fields (INV-005/INV-008). This is the pointer-advancing publication boundary, so it
+    refreshes by default -- callers refresh the content indices first, then call this."""
+    doc = _build_incremental_ref_doc(
+        org, repo, ref,
+        status="ready",
+        commit=commit,
+        target_commit=None,
+        commit_date_iso=commit_date_iso,
+        files_count=files_count,
+        lines_count=lines_count,
+        indexed_at=_now_iso(),
+        update_started_at=None,
+        failed_at=None,
+        error=None,
+    )
+    es.index(index=REFS_INDEX, id=build_incremental_ref_id(org, repo, ref), document=doc, refresh=refresh)
+
+
+def write_incremental_failed(
+    es: Elasticsearch,
+    org: str,
+    repo: str,
+    ref: str,
+    completed_commit: str | None,
+    target_commit: str | None,
+    error: str,
+    prior: dict | None = None,
+    refresh: bool = False,
+) -> None:
+    """Record a failed update WITHOUT advancing the completed pointer: status stays `indexing`,
+    `git.commit` remains the last completed SHA, and a bounded `error` + `failed_at` are stored
+    for diagnosis (INV-005). The next run retries old->current and clears these on success."""
+    prior = prior or {}
+    pg = prior.get("git", {})
+    doc = _build_incremental_ref_doc(
+        org, repo, ref,
+        status="indexing",
+        commit=completed_commit,
+        target_commit=target_commit,
+        commit_date_iso=pg.get("commit_date"),
+        files_count=prior.get("files_count", 0),
+        lines_count=prior.get("lines_count", 0),
+        indexed_at=prior.get("indexed_at"),
+        update_started_at=prior.get("update_started_at") or _now_iso(),
+        failed_at=_now_iso(),
+        error=error,
+    )
+    es.index(index=REFS_INDEX, id=build_incremental_ref_id(org, repo, ref), document=doc, refresh=refresh)
+
+
+def _delete_by_query_sync(es: Elasticsearch, index: str, query: dict, refresh: bool) -> None:
+    """Synchronous delete-by-query used by the incremental path. Unlike the async prune
+    deletion, this waits for completion (`wait_for_completion=True`) so a subsequent re-index
+    can't race a still-running delete, and uses `conflicts="proceed"` so a concurrent version
+    bump doesn't abort the batch. Missing indices (a first index, before any content exists)
+    are ignored."""
+    try:
+        es.delete_by_query(
+            index=index,
+            query=query,
+            wait_for_completion=True,
+            conflicts="proceed",
+            refresh=refresh,
+            ignore_unavailable=True,
+            allow_no_indices=True,
+        )
+    except NotFoundError:
+        pass
+
+
+def delete_incremental_paths(
+    es: Elasticsearch, org: str, repo: str, ref: str, paths, refresh: bool = False,
+) -> None:
+    """Synchronously delete the file and line docs for `paths` on this exact branch from the incremental
+    content indices. Scoped by the exact `git.ref_key` (a single keyword term, so a branch
+    whose name is a prefix of another can't bleed) plus a `file.path` terms filter -- never a
+    wildcard. A no-op for an empty path set."""
+    paths = list(paths)
+    if not paths:
+        return
+    ref_key = build_ref_key(org, repo, ref)
+    query = {
+        "bool": {
+            "filter": [
+                {"term": {"git.ref_key": ref_key}},
+                {"terms": {"file.path": paths}},
+            ]
+        }
+    }
+    for index in (files_index(org, repo), lines_index(org, repo)):
+        _delete_by_query_sync(es, index, query, refresh)
+
+
+def delete_incremental_branch(
+    es: Elasticsearch, org: str, repo: str, ref: str, refresh: bool = False,
+) -> None:
+    """Delete EVERY incremental content doc for this branch (full namespace), scoped by the exact
+    `git.ref_key`. Used for the initial index and the missing-diff-base rebuild (INV-007)."""
+    ref_key = build_ref_key(org, repo, ref)
+    query = {"bool": {"filter": [{"term": {"git.ref_key": ref_key}}]}}
+    for index in (files_index(org, repo), lines_index(org, repo)):
+        _delete_by_query_sync(es, index, query, refresh)
+
+
+def count_incremental_branch_docs(es: Elasticsearch, org: str, repo: str, ref: str) -> tuple[int, int]:
+    """Authoritative (files, lines) totals for a branch's current incremental view, counted by exact
+    `git.ref_key`. Call AFTER refreshing the content indices so the counts reflect the just-
+    applied deletes and indexes -- these become the ready marker's files_count/lines_count.
+    Returns 0 for an index that does not exist yet."""
+    ref_key = build_ref_key(org, repo, ref)
+    query = {"bool": {"filter": [{"term": {"git.ref_key": ref_key}}]}}
+
+    def _count(index: str) -> int:
+        try:
+            return int(es.count(index=index, query=query)["count"])
+        except NotFoundError:
+            return 0
+
+    return _count(files_index(org, repo)), _count(lines_index(org, repo))
+
+
+def refresh_incremental_content(es: Elasticsearch, org: str, repo: str) -> None:
+    """Make the branch's just-written incremental content visible before the ready pointer is published
+    (INV-008: content refresh precedes the final refs write). Best-effort over missing indices."""
+    es.indices.refresh(
+        index=[files_index(org, repo), lines_index(org, repo)],
+        ignore_unavailable=True,
+        allow_no_indices=True,
+    )
+
+
 def resolve_head(es: Elasticsearch, org: str, repo: str, ref_type: str, ref: str) -> dict | None:
     """The current marker for a ref: the newest commit_date among its (possibly many)
     markers. For a branch with retained history this is its live tip; for a tag it's the
diff --git a/src/sourcerer/commands/index/selection.py b/src/sourcerer/commands/index/selection.py
index 18485c2..4340460 100644
--- a/src/sourcerer/commands/index/selection.py
+++ b/src/sourcerer/commands/index/selection.py
@@ -25,19 +25,36 @@ def _resolve_entry(cfg: RepoConfig) -> list[Unit]:
     ls-remote failures (after retries) are reported to stderr; that ref type contributes
     no units so the run continues with the repos that did resolve."""
     fetched: dict[str, list[str] | None] = {}  # kind -> names (None = ls-remote failed)
-    seen: set[tuple[str, str]] = set()
+    # (kind, name) -> the update_mode it was first selected in. Overlap is judged on concrete
+    # resolved names (INV-010): a same-mode repeat dedupes to the existing Unit, while the same
+    # branch selected once as snapshot and once as incremental is a config error -- one branch
+    # cannot use both update modes at once.
+    seen: dict[tuple[str, str], str] = {}
     units: list[Unit] = []
+
+    def _select(rt: str, name: str, mode: str) -> None:
+        key = (rt, name)
+        prior = seen.get(key)
+        if prior is not None:
+            if prior != mode:
+                raise ValueError(
+                    f"{cfg.org}/{cfg.repo}: {rt} {name!r} is selected in both "
+                    f"'update: snapshot' and 'update: incremental' modes; a branch cannot be "
+                    f"indexed under both the v1 snapshot and incremental schemas at once"
+                )
+            return  # same-mode overlap: dedupe to the already-appended Unit
+        seen[key] = mode
+        units.append(Unit(org=cfg.org, repo=cfg.repo, ref=name, kind=rt, update_mode=mode))
+
     for sel in cfg.selectors:
         rt = sel.ref_type
         if rt == "commit":
             # Pinned commits aren't enumerable via ls-remote (there's no remote listing of
             # commits) -- `match` already holds the literal SHA/prefix strings to index, one
             # Unit per pattern. checkout_ref resolves the (possibly short) SHA at clone time.
+            # Commits are always snapshot (incremental is branch-only, enforced at parse time).
             for prefix in sel.raw_patterns:
-                if (rt, prefix) in seen:
-                    continue
-                seen.add((rt, prefix))
-                units.append(Unit(org=cfg.org, repo=cfg.repo, ref=prefix, kind=rt))
+                _select(rt, prefix, sel.update_mode)
             continue
         if rt not in fetched:
             fetched[rt] = list_remote_ref_names(
@@ -48,15 +65,12 @@ def _resolve_entry(cfg: RepoConfig) -> list[Unit]:
             continue  # ls-remote failed for this ref type, skip
         floor = sel.since_version_floor()  # version-based `since: {ref}`, name-only
         for name in names:
-            if (rt, name) in seen:
-                continue
             v = sel.matches(rt, name)
             if v is None:
                 continue
             if floor is not None and v.components < floor:
                 continue  # below the since version floor
-            seen.add((rt, name))
-            units.append(Unit(org=cfg.org, repo=cfg.repo, ref=name, kind=rt))
+            _select(rt, name, sel.update_mode)
 
     failed_kinds = sorted(k for k, v in fetched.items() if v is None)
     if failed_kinds:
diff --git a/src/sourcerer/commands/prune/execute.py b/src/sourcerer/commands/prune/execute.py
index f652c85..4fe0281 100644
--- a/src/sourcerer/commands/prune/execute.py
+++ b/src/sourcerer/commands/prune/execute.py
@@ -10,7 +10,12 @@
 # App packages
 from ...indices import REFS_INDEX, files_index, lines_index
 from ...planner import OrphanPlan, content_delete_set, plan_orphans
-from ...queries import enumerate_ref_tuples, gather_content_commit_tuples, list_sourcerer_indices
+from ...queries import (
+    enumerate_ref_repositories,
+    enumerate_ref_tuples,
+    gather_content_commit_tuples,
+    list_sourcerer_indices,
+)
 
 
 def execute_deletions(
@@ -71,8 +76,9 @@ def plan_orphans_now(es: Elasticsearch) -> OrphanPlan:
     between classes."""
     index_names = list_sourcerer_indices(es)
     ref_tuples = enumerate_ref_tuples(es)
+    ref_repositories = enumerate_ref_repositories(es)
     content_tuples = gather_content_commit_tuples(es, index_names)
-    return plan_orphans(index_names, ref_tuples, content_tuples)
+    return plan_orphans(index_names, ref_tuples, content_tuples, ref_repositories)
 
 
 def execute_orphan_deletions(es: Elasticsearch, plan: OrphanPlan) -> tuple[int, int, int]:
@@ -111,6 +117,7 @@ def execute_orphan_deletions(es: Elasticsearch, plan: OrphanPlan) -> tuple[int,
     if plan.orphan_marker_commits:
         should = [
             {"bool": {"filter": [
+                {"term": {"update_mode": "snapshot"}},
                 {"term": {"git.org": org}},
                 {"term": {"git.repo": repo}},
                 {"terms": {"git.commit": sorted(commits)}},
diff --git a/src/sourcerer/commands/setup/command.py b/src/sourcerer/commands/setup/command.py
index e8a4b45..e7dfdb5 100644
--- a/src/sourcerer/commands/setup/command.py
+++ b/src/sourcerer/commands/setup/command.py
@@ -10,9 +10,11 @@
 import yaml
 
 # App packages
-from ...utils import make_client
+from ...indices import REFS_INDEX
+from ...utils import ES_ERRORS, make_client
 
 _ELASTIC = resources.files("sourcerer") / "elastic"
+
 ELASTICSEARCH_INDEX_TEMPLATES_DIR = _ELASTIC / "index_templates"
 AGENT_BUILDER_TOOLS_DIR = _ELASTIC / "agent_builder_tools"
 AGENT_BUILDER_AGENTS_DIR = _ELASTIC / "agent_builder_agents"
@@ -72,6 +74,19 @@ def load_index_templates(es, templates_dir: pathlib.Path = ELASTICSEARCH_INDEX_T
     return loaded
 
 
+def ensure_refs_index(es) -> bool:
+    """Create the unified refs index or reject an existing non-lookup index."""
+    if es.indices.exists(index=REFS_INDEX):
+        settings = es.indices.get_settings(index=REFS_INDEX)[REFS_INDEX]["settings"]["index"]
+        if settings.get("mode") != "lookup":
+            raise ValueError(
+                f"{REFS_INDEX} exists without index.mode=lookup; rebuild Sourcerer indices"
+            )
+        return False
+    es.indices.create(index=REFS_INDEX)
+    return True
+
+
 def load_agent_builder_tools(
     session: requests.Session, kb_url: str, tools_dir: pathlib.Path = AGENT_BUILDER_TOOLS_DIR
 ) -> list[str]:
@@ -145,6 +160,13 @@ def run(url: str, api_key: str | None, username: str | None, password: str | Non
     for name in loaded:
         click.echo(f"Loaded index template: {name}")
 
+    try:
+        if ensure_refs_index(es):
+            click.echo(f"Created refs lookup index: {REFS_INDEX}")
+    except (*ES_ERRORS, ValueError) as e:
+        click.echo(f"Error: could not ensure refs lookup index: {e}", err=True)
+        sys.exit(1)
+
     if not kb_url:
         click.echo("Skipping agent builder setup (KIBANA_URL not set).")
         return
diff --git a/src/sourcerer/config.py b/src/sourcerer/config.py
index 83cdead..797f85e 100644
--- a/src/sourcerer/config.py
+++ b/src/sourcerer/config.py
@@ -111,6 +111,7 @@ class Selector:
     since: Since | None
     retain: Retain | None
     levels: tuple[str, ...] = ()   # numeric levels shared by the versioned match patterns
+    update_mode: str = "snapshot"  # "snapshot" (v1 default) | "incremental" (incremental branch path)
 
     def matches(self, ref_type: str, ref: str) -> Version | None:
         if self.ref_type != ref_type:
@@ -232,11 +233,26 @@ def _parse_commit_match(raw: dict, ctx: str) -> list[str]:
 def _parse_selector(raw: dict, ctx: str) -> Selector:
     if raw.get("type") not in ("branch", "tag", "commit"):
         raise ValueError(f"{ctx}: 'type' must be 'branch', 'tag', or 'commit'")
-    unknown = set(raw) - {"type", "match", "since", "retain"}
+    unknown = set(raw) - {"type", "match", "since", "retain", "update"}
     if unknown:
         raise ValueError(f"{ctx}: unknown keys {sorted(unknown)}")
     ref_type = raw["type"]
 
+    # `update` opts a branch selector into the unified v1 incremental path; omitted or an
+    # explicit `snapshot` keeps the default v1 behaviour. Incremental is branch-only and cannot
+    # be combined with `since` or `retain`: it maintains a single mutable branch view with no
+    # history for retention to trim and no inclusion floor to apply.
+    update_mode = raw.get("update", "snapshot")
+    if update_mode not in ("snapshot", "incremental"):
+        raise ValueError(f"{ctx}: 'update' must be 'snapshot' or 'incremental' (got {update_mode!r})")
+    if update_mode == "incremental":
+        if ref_type != "branch":
+            raise ValueError(f"{ctx}: 'update: incremental' is only valid for 'type: branch'")
+        if raw.get("since") is not None:
+            raise ValueError(f"{ctx}: 'update: incremental' cannot be combined with 'since'")
+        if raw.get("retain") is not None:
+            raise ValueError(f"{ctx}: 'update: incremental' cannot be combined with 'retain'")
+
     if ref_type == "commit":
         # A pinned commit has no enumerable name to pattern-match against (see selection.py),
         # so `match` holds literal SHA/prefix strings instead of version.py DSL patterns, and
@@ -276,7 +292,7 @@ def _parse_selector(raw: dict, ctx: str) -> Selector:
             retain = None
 
     return Selector(ref_type=ref_type, raw_patterns=patterns, compiled=compiled,
-                    since=since, retain=retain, levels=levels)
+                    since=since, retain=retain, levels=levels, update_mode=update_mode)
 
 
 def _deep_merge(dst: dict, src: dict) -> None:
diff --git a/src/sourcerer/elastic/agent_builder_skills/sourcerer-code-citations.yml b/src/sourcerer/elastic/agent_builder_skills/sourcerer-code-citations.yml
index f2b0ecc..64846ca 100644
--- a/src/sourcerer/elastic/agent_builder_skills/sourcerer-code-citations.yml
+++ b/src/sourcerer/elastic/agent_builder_skills/sourcerer-code-citations.yml
@@ -42,6 +42,13 @@ content: |-
     - Never derive a line position by counting lines inside returned file content (see Integrity below).
 
   ## URL templates
+  Every content tool returns a `git.commit` to build these URLs. For a snapshot ref it is the
+  pinned commit; for an incremental branch it is the branch's last COMPLETED commit, attached
+  automatically via the refs lookup join (you query the branch by `git_ref_key`, and still cite
+  by the returned `git.commit`). While an incremental branch is `status: indexing`, that commit
+  remains the last completed pointer, so citations stay valid even though the live view may
+  briefly show a mixed revision. During a branch's first index, content tools return no rows
+  until a non-null completed commit is published at `status: ready`.
   - Directory: `https://github.com/{git.org}/{git.repo}/blob/{git.commit}/{file.directory}`
   - File: `https://github.com/{git.org}/{git.repo}/blob/{git.commit}/{file.path}`
   - Single line: `https://github.com/{git.org}/{git.repo}/blob/{git.commit}/{file.path}#L{line.number}`
diff --git a/src/sourcerer/elastic/agent_builder_skills/sourcerer-ref-resolution.yml b/src/sourcerer/elastic/agent_builder_skills/sourcerer-ref-resolution.yml
index a9ba4ba..d248a37 100644
--- a/src/sourcerer/elastic/agent_builder_skills/sourcerer-ref-resolution.yml
+++ b/src/sourcerer/elastic/agent_builder_skills/sourcerer-ref-resolution.yml
@@ -22,6 +22,14 @@ content: |-
   Combine with `git_ref_type: tag`, `git_ref_type: branch`, or `git_ref_type: commit` (a pinned,
   ad-hoc commit not on a tracked branch/tag tip) to narrow further.
 
+  Each ref carries an `update_mode` that dictates HOW you query its content:
+  - `update_mode: snapshot` — the ref is a v1 immutable snapshot. Pin its `git.commit` and pass
+    it as `git_commit` to every `sourcerer.code.*` / `sourcerer.files.*` call.
+  - `update_mode: incremental` — the ref is a mutable incremental branch view. Pass its exact
+    `git.ref_key` as `git_ref_key` to the content tools (NOT `git_commit`). The tools attach the
+    branch's completed commit automatically for citations. Never mix the two: supply either
+    `git_commit` (snapshot) or `git_ref_key` (incremental), never both.
+
   ## Resolution scenarios
 
   ### No ref specified — default to latest stable
@@ -43,8 +51,23 @@ content: |-
   ### Explicit ref (branch name, exact tag, commit hash)
   Use as given. If it is a branch, call `refs.list` with `git_ref_type: branch` to confirm it exists and retrieve its current commit. If it is a tag, confirm and get its commit. If it is a commit hash, use it directly — optionally confirm with `git_ref_type: commit` if it may be a pinned commit rather than one reached via a branch/tag.
 
+  ## Incremental branches (update_mode: incremental)
+  A incremental branch is a single mutable view, not a per-commit snapshot. Resolve it with
+  `refs.list` (`git_ref_type: branch`) and read its `update_mode`, `status`, `git.commit`
+  (last completed), `git.target_commit` (candidate being applied), and `git.ref_key`.
+  - Query its content by passing the exact `git_ref_key` to the code/file tools.
+  - `status: ready` means `git.commit` is the fully-applied tip.
+  - `status: indexing` means an update is in flight. After at least 1 completed run, results
+    remain available but may briefly reflect a MIXED revision while Elasticsearch applies the
+    change; `git.commit` stays at the last completed pointer for citations. During the first run,
+    `git.commit` is null and content tools return no rows until the branch reaches `status: ready`.
+    There is no per-commit history for an incremental branch — only its current view.
+
   ## Pinning the commit
-  Once a ref resolves to a `git.commit`, use that commit in every subsequent `sourcerer.code.*` and `sourcerer.files.*` call for that repo and ref. Re-invoke this skill only when the question introduces a new or additional ref.
+  For a snapshot ref, once it resolves to a `git.commit`, use that commit as `git_commit` in
+  every subsequent `sourcerer.code.*` and `sourcerer.files.*` call for that repo and ref. For an
+  incremental branch, reuse its `git_ref_key` the same way. Re-invoke this skill only when the
+  question introduces a new or additional ref.
 tool_ids:
 - sourcerer.refs.list
 referenced_content: []
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.grep.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.grep.yml
index a22d9ac..37f653a 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.grep.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.grep.yml
@@ -4,13 +4,21 @@ description: grep code in repo file(s) with regex
 tags: []
 configuration:
   query: |-
-    // Scope the search
-    FROM sourcerer-v1-lines*
-    | WHERE MATCH(git.org, ?git_org)
-        AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
-        AND file.path LIKE ?file_path
-        AND line.content RLIKE ?regex
+    // Scope to EXACTLY ONE schema (see sourcerer.code.search): snapshot by git.commit when
+    // no git_ref_key is supplied, or incremental by exact git.ref_key when it is.
+    FROM sourcerer-v1-lines~*
+    | WHERE (?git_ref_key == "" AND update_mode == "snapshot"
+                AND git.org LIKE ?git_org AND git.repo LIKE ?git_repo AND git.commit LIKE ?git_commit)
+         OR (?git_ref_key != "" AND update_mode == "incremental"
+                AND git.ref_key == ?git_ref_key)
+    | WHERE line.content RLIKE ?regex AND file.path LIKE ?file_path
+
+    // Attach the completed commit for incremental rows via the refs lookup join on exact git.ref_key.
+    | EVAL _org = git.org, _repo = git.repo, _commit = git.commit
+    | LOOKUP JOIN sourcerer-v1-refs ON git.ref_key
+    | EVAL git.org = _org, git.repo = _repo, git.commit = COALESCE(git.commit, _commit)
+    // Hide a first incremental build until it has a completed commit for valid citations.
+    | WHERE git.commit IS NOT NULL
 
     // Flag whether the file path pattern is recursive (e.g. "src/**/*.java")
     | EVAL fp_is_recursive = ?file_path != REPLACE(?file_path, "[*][*]", "")
@@ -40,9 +48,14 @@ configuration:
       defaultValue: "*"
     git_commit:
       type: string
-      description: Filter by git commit(s) (supports * wildcards)
+      description: Filter by git commit(s) for snapshot refs (supports * wildcards). Ignored when git_ref_key is set.
       optional: true
       defaultValue: "*"
+    git_ref_key:
+      type: string
+      description: Exact ref key of a incremental branch (from sourcerer.refs.list). When set, the grep targets that branch's incremental content instead of snapshot content; leave empty for snapshot (git_commit) queries.
+      optional: true
+      defaultValue: ""
     file_path:
       type: string
       description: File path(s) to grep (supports * and ** glob syntax, e.g. src/test, src/*/resources, src/**, src/**/*.xml)
@@ -56,4 +69,4 @@ configuration:
       type: integer
       description: Number of search results
       optional: true
-      defaultValue: 100
\ No newline at end of file
+      defaultValue: 100
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.search.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.search.yml
index b06340d..d1632ad 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.search.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.code.search.yml
@@ -4,13 +4,24 @@ description: search code in repo file(s) with fast, relevant, code-optimized bm2
 tags: []
 configuration:
   query: |-
-    // Scope the search
-    FROM sourcerer-v1-lines* METADATA _score
-    | WHERE MATCH(git.org, ?git_org)
-        AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
-        AND file.path LIKE ?file_path
-        AND MATCH(line.content.text, ?q)
+    // Scope the search to EXACTLY ONE schema, never both:
+    //   - no git_ref_key  -> snapshot content, matched by git.commit (default "*")
+    //   - git_ref_key set -> incremental content, matched by the exact git.ref_key
+    // The two branches are mutually exclusive, so one call selects exactly one update mode.
+    FROM sourcerer-v1-lines~* METADATA _score
+    | WHERE (?git_ref_key == "" AND update_mode == "snapshot"
+                AND git.org LIKE ?git_org AND git.repo LIKE ?git_repo AND git.commit LIKE ?git_commit)
+         OR (?git_ref_key != "" AND update_mode == "incremental"
+                AND git.ref_key == ?git_ref_key)
+    | WHERE MATCH(line.content.text, ?q) AND file.path LIKE ?file_path
+
+    // Attach the completed commit for incremental rows via the refs lookup join on exact git.ref_key
+    // (snapshot rows have a null ref_key, don't match, and keep their own commit).
+    | EVAL _org = git.org, _repo = git.repo, _commit = git.commit
+    | LOOKUP JOIN sourcerer-v1-refs ON git.ref_key
+    | EVAL git.org = _org, git.repo = _repo, git.commit = COALESCE(git.commit, _commit)
+    // Hide a first incremental build until it has a completed commit for valid citations.
+    | WHERE git.commit IS NOT NULL
 
     // Flag whether the file path pattern is recursive (e.g. "src/**/*.java")
     | EVAL fp_is_recursive = ?file_path != REPLACE(?file_path, "[*][*]", "")
@@ -40,9 +51,14 @@ configuration:
       defaultValue: "*"
     git_commit:
       type: string
-      description: Filter by git commit(s) (supports * wildcards)
+      description: Filter by git commit(s) for snapshot refs (supports * wildcards). Ignored when git_ref_key is set.
       optional: true
       defaultValue: "*"
+    git_ref_key:
+      type: string
+      description: Exact ref key of a incremental branch (from sourcerer.refs.list). When set, the search targets that branch's incremental content instead of snapshot content; leave empty for snapshot (git_commit) queries.
+      optional: true
+      defaultValue: ""
     file_path:
       type: string
       description: File path(s) to grep (supports * and ** glob syntax, e.g. src/test, src/*/resources, src/**, src/**/*.xml)
@@ -56,4 +72,4 @@ configuration:
       type: integer
       description: Number of search results
       optional: true
-      defaultValue: 10
\ No newline at end of file
+      defaultValue: 10
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.cat.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.cat.yml
index 4888cec..eeed6a9 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.cat.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.cat.yml
@@ -4,13 +4,22 @@ description: cat repo file(s)
 tags: []
 configuration:
   query: |-
-    // Scope the search
-    FROM sourcerer-v1-lines*
-    | WHERE MATCH(git.org, ?git_org)
-        AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
-        AND file.path LIKE ?file_path
-        AND MATCH(file.path, ?file_path)
+    // Scope to EXACTLY ONE schema: snapshot by git.commit when no git_ref_key is supplied,
+    // or incremental by exact git.ref_key when it is. The branches are mutually exclusive.
+    FROM sourcerer-v1-lines~*
+    | WHERE (?git_ref_key == "" AND update_mode == "snapshot"
+                AND git.org LIKE ?git_org AND git.repo LIKE ?git_repo AND git.commit LIKE ?git_commit)
+         OR (?git_ref_key != "" AND update_mode == "incremental"
+                AND git.ref_key == ?git_ref_key)
+    | WHERE file.path LIKE ?file_path AND MATCH(file.path, ?file_path)
+
+    // Attach the completed commit for incremental rows via the refs lookup join on exact git.ref_key,
+    // so the reconstructed file is grouped and cited by that commit (snapshot rows keep their own).
+    | EVAL _org = git.org, _repo = git.repo, _commit = git.commit
+    | LOOKUP JOIN sourcerer-v1-refs ON git.ref_key
+    | EVAL git.org = _org, git.repo = _repo, git.commit = COALESCE(git.commit, _commit)
+    // Hide a first incremental build until it has a completed commit for valid citations.
+    | WHERE git.commit IS NOT NULL
 
     // Construct the output for each file based on its lines
     // Prefix each line with its zero-padded line number so: (a) duplicate lines stay unique; and (b) lexical sort = line order
@@ -39,9 +48,14 @@ configuration:
       defaultValue: "*"
     git_commit:
       type: string
-      description: Filter by git commit(s) (supports * wildcards)
+      description: Filter by git commit(s) for snapshot refs (supports * wildcards). Ignored when git_ref_key is set.
       optional: true
       defaultValue: "*"
+    git_ref_key:
+      type: string
+      description: Exact ref key of a incremental branch (from sourcerer.refs.list). When set, cat targets that branch's incremental content instead of snapshot content; leave empty for snapshot (git_commit) queries.
+      optional: true
+      defaultValue: ""
     file_path:
       type: string
       description: File path(s) to cat (supports * and ** glob syntax, e.g. src/test, src/*/resources, src/**, src/**/*.xml)
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.head.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.head.yml
index f103b09..33800ce 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.head.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.head.yml
@@ -4,14 +4,22 @@ description: head repo file(s)
 tags: []
 configuration:
   query: |-
-    // Scope the search
-    FROM sourcerer-v1-lines*
-    | WHERE MATCH(git.org, ?git_org)
-        AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
-        AND file.path LIKE ?file_path
-        AND MATCH(file.path, ?file_path)
-    
+    // Scope to EXACTLY ONE schema: snapshot by git.commit when no git_ref_key is supplied,
+    // or incremental by exact git.ref_key when it is. The branches are mutually exclusive.
+    FROM sourcerer-v1-lines~*
+    | WHERE (?git_ref_key == "" AND update_mode == "snapshot"
+                AND git.org LIKE ?git_org AND git.repo LIKE ?git_repo AND git.commit LIKE ?git_commit)
+         OR (?git_ref_key != "" AND update_mode == "incremental"
+                AND git.ref_key == ?git_ref_key)
+    | WHERE file.path LIKE ?file_path AND MATCH(file.path, ?file_path)
+
+    // Attach the completed commit for incremental rows via the refs lookup join on exact git.ref_key.
+    | EVAL _org = git.org, _repo = git.repo, _commit = git.commit
+    | LOOKUP JOIN sourcerer-v1-refs ON git.ref_key
+    | EVAL git.org = _org, git.repo = _repo, git.commit = COALESCE(git.commit, _commit)
+    // Hide a first incremental build until it has a completed commit for valid citations.
+    | WHERE git.commit IS NOT NULL
+
     // Construct the output for each file based on its lines
     // Prefix each line with its zero-padded line number so: (a) duplicate lines stay unique; and (b) lexical sort = line order
     | EVAL pair = CONCAT(RIGHT(CONCAT("0000000000", TO_STRING(line.number)), 10), line.content)
@@ -42,9 +50,14 @@ configuration:
       defaultValue: "*"
     git_commit:
       type: string
-      description: Filter by git commit(s) (supports * wildcards)
+      description: Filter by git commit(s) for snapshot refs (supports * wildcards). Ignored when git_ref_key is set.
       optional: true
       defaultValue: "*"
+    git_ref_key:
+      type: string
+      description: Exact ref key of a incremental branch (from sourcerer.refs.list). When set, head targets that branch's incremental content instead of snapshot content; leave empty for snapshot (git_commit) queries.
+      optional: true
+      defaultValue: ""
     file_path:
       type: string
       description: File path(s) to head (supports * and ** glob syntax, e.g. src/test, src/*/resources, src/**, src/**/*.xml)
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.ls.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.ls.yml
index 4bc5b96..b887f90 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.ls.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.ls.yml
@@ -4,11 +4,22 @@ description: ls files and directories under a given path or glob pattern in a re
 tags: []
 configuration:
   query: |-
-    // Scope the search
-    FROM sourcerer-v1-files*
-    | WHERE MATCH(git.org, ?git_org)
-        AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
+    // Scope to EXACTLY ONE schema: snapshot by git.commit when no git_ref_key is supplied,
+    // or incremental by exact git.ref_key when it is. The branches are mutually exclusive,
+    // so an unscoped call never lists a mix of snapshot and incremental content.
+    FROM sourcerer-v1-files~*
+    | WHERE (?git_ref_key == "" AND update_mode == "snapshot"
+                AND git.org LIKE ?git_org AND git.repo LIKE ?git_repo AND git.commit LIKE ?git_commit)
+         OR (?git_ref_key != "" AND update_mode == "incremental"
+                AND git.ref_key == ?git_ref_key)
+
+    // Attach the completed commit for incremental rows via the refs lookup join on exact git.ref_key
+    // (kept for parity with the other content tools; ls emits only names).
+    | EVAL _org = git.org, _repo = git.repo, _commit = git.commit
+    | LOOKUP JOIN sourcerer-v1-refs ON git.ref_key
+    | EVAL git.org = _org, git.repo = _repo, git.commit = COALESCE(git.commit, _commit)
+    // Hide a first incremental build until it has a completed commit for valid citations.
+    | WHERE git.commit IS NOT NULL
 
     // Normalize the input file_path by stripping any trailing slash
     // (e.g. "src/" and "src" are equivalent)
@@ -88,9 +99,14 @@ configuration:
       defaultValue: "*"
     git_commit:
       type: string
-      description: Filter by git commit(s) (supports * wildcards)
+      description: Filter by git commit(s) for snapshot refs (supports * wildcards). Ignored when git_ref_key is set.
       optional: true
       defaultValue: "*"
+    git_ref_key:
+      type: string
+      description: Exact ref key of a incremental branch (from sourcerer.refs.list). When set, ls targets that branch's incremental content instead of snapshot content; leave empty for snapshot (git_commit) queries.
+      optional: true
+      defaultValue: ""
     file_path:
       type: string
       description: File path to ls (supports * and ** glob syntax, e.g. src/test, src/*/resources, src/**, src/**/*.xml)
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.tail.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.tail.yml
index f073f04..fa3c32a 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.tail.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.files.tail.yml
@@ -4,14 +4,22 @@ description: tail repo file(s)
 tags: []
 configuration:
   query: |-
-    // Scope the search
-    FROM sourcerer-v1-lines*
-    | WHERE MATCH(git.org, ?git_org)
-        AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
-        AND file.path LIKE ?file_path
-        AND MATCH(file.path, ?file_path)
-    
+    // Scope to EXACTLY ONE schema: snapshot by git.commit when no git_ref_key is supplied,
+    // or incremental by exact git.ref_key when it is. The branches are mutually exclusive.
+    FROM sourcerer-v1-lines~*
+    | WHERE (?git_ref_key == "" AND update_mode == "snapshot"
+                AND git.org LIKE ?git_org AND git.repo LIKE ?git_repo AND git.commit LIKE ?git_commit)
+         OR (?git_ref_key != "" AND update_mode == "incremental"
+                AND git.ref_key == ?git_ref_key)
+    | WHERE file.path LIKE ?file_path AND MATCH(file.path, ?file_path)
+
+    // Attach the completed commit for incremental rows via the refs lookup join on exact git.ref_key.
+    | EVAL _org = git.org, _repo = git.repo, _commit = git.commit
+    | LOOKUP JOIN sourcerer-v1-refs ON git.ref_key
+    | EVAL git.org = _org, git.repo = _repo, git.commit = COALESCE(git.commit, _commit)
+    // Hide a first incremental build until it has a completed commit for valid citations.
+    | WHERE git.commit IS NOT NULL
+
     // Construct the output for each file based on its lines
     // Prefix each line with its zero-padded line number so: (a) duplicate lines stay unique; and (b) lexical sort = line order
     | EVAL pair = CONCAT(RIGHT(CONCAT("0000000000", TO_STRING(line.number)), 10), line.content)
@@ -42,9 +50,14 @@ configuration:
       defaultValue: "*"
     git_commit:
       type: string
-      description: Filter by git commit(s) (supports * wildcards)
+      description: Filter by git commit(s) for snapshot refs (supports * wildcards). Ignored when git_ref_key is set.
       optional: true
       defaultValue: "*"
+    git_ref_key:
+      type: string
+      description: Exact ref key of a incremental branch (from sourcerer.refs.list). When set, tail targets that branch's incremental content instead of snapshot content; leave empty for snapshot (git_commit) queries.
+      optional: true
+      defaultValue: ""
     file_path:
       type: string
       description: File path(s) to tail (supports * and ** glob syntax, e.g. src/test, src/*/resources, src/**, src/**/*.xml)
@@ -53,4 +66,4 @@ configuration:
       type: integer
       description: Number of lines to return
       optional: true
-      defaultValue: 10
\ No newline at end of file
+      defaultValue: 10
diff --git a/src/sourcerer/elastic/agent_builder_tools/sourcerer.refs.list.yml b/src/sourcerer/elastic/agent_builder_tools/sourcerer.refs.list.yml
index 5c0dd95..e0108c8 100644
--- a/src/sourcerer/elastic/agent_builder_tools/sourcerer.refs.list.yml
+++ b/src/sourcerer/elastic/agent_builder_tools/sourcerer.refs.list.yml
@@ -1,22 +1,31 @@
 id: sourcerer.refs.list
 type: esql
-description: list the git refs (branches, tags) indexed for org(s)/repo(s)
+description: list the git refs (branches, tags) indexed for org(s)/repo(s), including snapshot refs and incremental branches
 tags: []
 configuration:
   query: |-
-    // List the indexed refs. Every filter is wildcard-friendly and defaults to "*",
-    // so calling with no params lists every ref in every repo. Narrow with any combo
-    // of org / repo / ref / ref_type / commit to browse what's available to
-    // cat/head/tail/grep/search.
-    FROM sourcerer-v1-refs*
+    // List snapshot and incremental refs from the unified v1 refs index.
+    // branch markers. Every filter is wildcard-friendly and defaults to "*", so calling with
+    // no params lists every ref in every repo. Narrow with any combo of org / repo / ref /
+    // ref_type / commit.
+    //
+    // update_mode tells the agent HOW to query the ref's content:
+    //   - snapshot    -> pass the completed git_commit to the code/file tools
+    //   - incremental -> pass the exact git_ref_key to the code/file tools
+    // For incremental refs, git.commit is the LAST COMPLETED commit and status may be
+    // "indexing" (an update is in flight); git.target_commit is the candidate being applied.
+    FROM sourcerer-v1-refs
     | WHERE git.org LIKE ?git_org
         AND git.repo LIKE ?git_repo
-        AND git.commit LIKE ?git_commit
+        // Tolerate a null completed commit: an incremental branch mid-first-index (or a failed
+        // first index) has git.commit == null and must still be listed so operators can see its
+        // status. `null LIKE "*"` is null (excluded), so short-circuit the default wildcard.
+        AND (?git_commit == "*" OR git.commit LIKE ?git_commit)
         AND git.ref LIKE ?git_ref
         AND git.ref_type LIKE ?git_ref_type
 
     // Most recent first so a moved branch's current commit leads
-    | KEEP git.org, git.repo, git.ref, git.ref_type, git.commit, git.commit_date, status, files_count, lines_count, indexed_at
+    | KEEP git.org, git.repo, git.ref, git.ref_type, git.ref_key, git.commit, git.target_commit, git.commit_date, status, update_mode, files_count, lines_count, indexed_at
     | SORT indexed_at DESC
     | LIMIT 1000
   params:
diff --git a/src/sourcerer/elastic/index_templates/sourcerer-v1-files.json b/src/sourcerer/elastic/index_templates/sourcerer-v1-files.json
index 70d690a..e826228 100644
--- a/src/sourcerer/elastic/index_templates/sourcerer-v1-files.json
+++ b/src/sourcerer/elastic/index_templates/sourcerer-v1-files.json
@@ -1,9 +1,9 @@
 {
   "_meta": {
-    "description": "sourcerer-v1-files"
+    "description": "Unified Sourcerer v1 file documents for snapshot and incremental modes"
   },
   "index_patterns": [
-    "sourcerer-v1-files*"
+    "sourcerer-v1-files~*"
   ],
   "template": {
     "settings": {
@@ -13,38 +13,33 @@
           "source": {
             "mode": "synthetic"
           }
-        },
-        "sort": {
-          "field": [
-            "git.org",
-            "git.repo",
-            "git.commit",
-            "file.path"
-          ],
-          "order": [
-            "asc",
-            "asc",
-            "asc",
-            "asc"
-          ]
         }
       }
     },
     "mappings": {
       "properties": {
+        "update_mode": {
+          "type": "keyword"
+        },
         "git": {
           "properties": {
             "org": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "repo": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "commit": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
+            },
+            "ref_key": {
+              "type": "keyword"
+            },
+            "ref": {
+              "type": "keyword"
+            },
+            "ref_type": {
+              "type": "keyword"
             }
           }
         },
@@ -75,4 +70,4 @@
       }
     }
   }
-}
\ No newline at end of file
+}
diff --git a/src/sourcerer/elastic/index_templates/sourcerer-v1-lines.json b/src/sourcerer/elastic/index_templates/sourcerer-v1-lines.json
index 2ed6289..47dca16 100644
--- a/src/sourcerer/elastic/index_templates/sourcerer-v1-lines.json
+++ b/src/sourcerer/elastic/index_templates/sourcerer-v1-lines.json
@@ -1,9 +1,9 @@
 {
   "_meta": {
-    "description": "sourcerer-v1-lines"
+    "description": "Unified Sourcerer v1 line documents for snapshot and incremental modes"
   },
   "index_patterns": [
-    "sourcerer-v1-lines*"
+    "sourcerer-v1-lines~*"
   ],
   "template": {
     "settings": {
@@ -13,22 +13,6 @@
           "source": {
             "mode": "synthetic"
           }
-        },
-        "sort": {
-          "field": [
-            "git.org",
-            "git.repo",
-            "git.commit",
-            "file.path",
-            "line.number"
-          ],
-          "order": [
-            "asc",
-            "asc",
-            "asc",
-            "asc",
-            "asc"
-          ]
         }
       },
       "analysis": {
@@ -74,19 +58,28 @@
     },
     "mappings": {
       "properties": {
+        "update_mode": {
+          "type": "keyword"
+        },
         "git": {
           "properties": {
             "org": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "repo": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "commit": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
+            },
+            "ref_key": {
+              "type": "keyword"
+            },
+            "ref": {
+              "type": "keyword"
+            },
+            "ref_type": {
+              "type": "keyword"
             }
           }
         },
@@ -128,4 +121,4 @@
       }
     }
   }
-}
\ No newline at end of file
+}
diff --git a/src/sourcerer/elastic/index_templates/sourcerer-v1-refs.json b/src/sourcerer/elastic/index_templates/sourcerer-v1-refs.json
index 18cb030..e4af755 100644
--- a/src/sourcerer/elastic/index_templates/sourcerer-v1-refs.json
+++ b/src/sourcerer/elastic/index_templates/sourcerer-v1-refs.json
@@ -1,51 +1,42 @@
 {
   "_meta": {
-    "description": "sourcerer-v1-refs"
+    "description": "Unified Sourcerer v1 ref metadata for snapshot and incremental modes"
   },
   "index_patterns": [
-    "sourcerer-v1-refs*"
+    "sourcerer-v1-refs"
   ],
   "template": {
     "settings": {
       "index": {
-        "codec": "best_compression",
-        "sort": {
-          "field": [
-            "git.org",
-            "git.repo",
-            "git.ref"
-          ],
-          "order": [
-            "asc",
-            "asc",
-            "asc"
-          ]
-        }
+        "mode": "lookup",
+        "number_of_shards": 1,
+        "codec": "best_compression"
       }
     },
     "mappings": {
       "properties": {
         "git": {
           "properties": {
+            "ref_key": {
+              "type": "keyword"
+            },
             "org": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "repo": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "ref": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "ref_type": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
             },
             "commit": {
-              "type": "keyword",
-              "normalizer": "lowercase"
+              "type": "keyword"
+            },
+            "target_commit": {
+              "type": "keyword"
             },
             "commit_date": {
               "type": "date"
@@ -53,8 +44,10 @@
           }
         },
         "status": {
-          "type": "keyword",
-          "normalizer": "lowercase"
+          "type": "keyword"
+        },
+        "update_mode": {
+          "type": "keyword"
         },
         "files_count": {
           "type": "long"
@@ -64,8 +57,18 @@
         },
         "indexed_at": {
           "type": "date"
+        },
+        "update_started_at": {
+          "type": "date"
+        },
+        "failed_at": {
+          "type": "date"
+        },
+        "error": {
+          "type": "keyword",
+          "index": false
         }
       }
     }
   }
-}
\ No newline at end of file
+}
diff --git a/src/sourcerer/indices.py b/src/sourcerer/indices.py
index 36e6686..1acec79 100644
--- a/src/sourcerer/indices.py
+++ b/src/sourcerer/indices.py
@@ -1,9 +1,5 @@
 # sourcerer/indices.py
-# Index-name constants and builders shared by the index and prune commands: every physical
-# per-repo content index is named from these, and the refs index name is the same constant
-# everywhere. Kept dependency-free (no ES, no click) so both command packages -- and anything
-# that reads index names without touching a cluster -- can import it without pulling in either
-# command's logic.
+# Unified v1 index-name constants and builders shared by snapshot and incremental indexing.
 
 FILES_INDEX_PREFIX = "sourcerer-v1-files"
 LINES_INDEX_PREFIX = "sourcerer-v1-lines"
@@ -11,10 +7,10 @@
 
 
 def files_index(org: str, repo: str) -> str:
-    """Return the per-repo files index name, e.g. sourcerer-v1-files~elastic~elasticsearch."""
+    """Return the per-repo files index name with normalized repository identity."""
     return f"{FILES_INDEX_PREFIX}~{org.lower()}~{repo.lower()}"
 
 
 def lines_index(org: str, repo: str) -> str:
-    """Return the per-repo lines index name, e.g. sourcerer-v1-lines~elastic~elasticsearch."""
+    """Return the per-repo lines index name with normalized repository identity."""
     return f"{LINES_INDEX_PREFIX}~{org.lower()}~{repo.lower()}"
diff --git a/src/sourcerer/planner.py b/src/sourcerer/planner.py
index 9af1e4d..3389e3c 100644
--- a/src/sourcerer/planner.py
+++ b/src/sourcerer/planner.py
@@ -311,13 +311,15 @@ def plan_orphans(
     index_names: list[str],
     ref_commit_tuples: set[tuple[str, str, str]],
     content_commit_tuples: set[tuple[str, str, str]],
+    ref_repositories: set[tuple[str, str]] | None = None,
 ) -> OrphanPlan:
     """Combine the three orphan classes into one plan from three cheap snapshots: the physical
     index names, the distinct (org, repo, commit) tuples in refs, and the distinct (org, repo,
     commit) tuples with content docs (already unioned across the files and lines indices
     present). Pure -- no ES calls -- so this is the one seam orphan-sweep tests need to hit."""
-    ref_orgs = {org for org, _, _ in ref_commit_tuples}
-    ref_repos = {(org, repo) for org, repo, _ in ref_commit_tuples}
+    snapshot_ref_repos = {(org, repo) for org, repo, _ in ref_commit_tuples}
+    ref_repos = snapshot_ref_repos if ref_repositories is None else ref_repositories
+    ref_orgs = {org for org, _ in ref_repos}
 
     orphan_index_names = orphan_indices(index_names, ref_orgs, ref_repos, ref_commit_tuples)
 
diff --git a/src/sourcerer/progress.py b/src/sourcerer/progress.py
index bb293be..c3630b1 100644
--- a/src/sourcerer/progress.py
+++ b/src/sourcerer/progress.py
@@ -47,6 +47,7 @@ class Unit:
     repo: str
     ref: str | None
     kind: str
+    update_mode: str = "snapshot"  # "snapshot" (v1) | "incremental" (incremental branch path)
     stage: str = "pending"  # pending|resolving|cloning|checkout|indexing|done
     total_files: int | None = None
     files: int = 0
diff --git a/src/sourcerer/queries.py b/src/sourcerer/queries.py
index fe7db8b..980c6a1 100644
--- a/src/sourcerer/queries.py
+++ b/src/sourcerer/queries.py
@@ -78,19 +78,48 @@ def list_sourcerer_indices(es: Elasticsearch) -> list[str]:
 
 
 def enumerate_ref_tuples(es: Elasticsearch) -> set[tuple[str, str, str]]:
-    """Every distinct (git.org, git.repo, git.commit) tuple recorded in sourcerer-v1-refs, via
-    a paginated composite aggregation (safe over an unbounded number of distinct tuples).
-    Returns an empty set if the refs index doesn't exist yet."""
-    return _composite_org_repo_commit_tuples(es, REFS_INDEX)
+    """Every snapshot (org, repo, commit) tuple in refs.
+
+    Incremental refs use ref-addressed content and must not participate in commit orphaning.
+    """
+    return _composite_org_repo_commit_tuples(es, REFS_INDEX, update_mode="snapshot")
+
+
+def enumerate_ref_repositories(es: Elasticsearch) -> set[tuple[str, str]]:
+    """Every repository represented by either update mode in the unified refs index."""
+    out: set[tuple[str, str]] = set()
+    after: dict | None = None
+    while True:
+        composite: dict = {
+            "size": _COMPOSITE_PAGE_SIZE,
+            "sources": [
+                {"org": {"terms": {"field": "git.org"}}},
+                {"repo": {"terms": {"field": "git.repo"}}},
+            ],
+        }
+        if after is not None:
+            composite["after"] = after
+        try:
+            resp = es.search(index=REFS_INDEX, size=0, aggs={"repos": {"composite": composite}})
+        except NotFoundError:
+            return out
+        agg = resp["aggregations"]["repos"]
+        for bucket in agg["buckets"]:
+            out.add((bucket["key"]["org"], bucket["key"]["repo"]))
+        after = agg.get("after_key")
+        if not agg["buckets"] or after is None:
+            return out
 
 
 def enumerate_content_commits(es: Elasticsearch, index: str) -> set[tuple[str, str, str]]:
     """Every distinct (git.org, git.repo, git.commit) tuple with at least one doc in `index` (a
     single files or lines physical index). Returns an empty set if `index` doesn't exist."""
-    return _composite_org_repo_commit_tuples(es, index)
+    return _composite_org_repo_commit_tuples(es, index, update_mode="snapshot")
 
 
-def _composite_org_repo_commit_tuples(es: Elasticsearch, index: str) -> set[tuple[str, str, str]]:
+def _composite_org_repo_commit_tuples(
+    es: Elasticsearch, index: str, update_mode: str | None = None,
+) -> set[tuple[str, str, str]]:
     out: set[tuple[str, str, str]] = set()
     after: dict | None = None
     while True:
@@ -105,7 +134,10 @@ def _composite_org_repo_commit_tuples(es: Elasticsearch, index: str) -> set[tupl
         if after is not None:
             composite["after"] = after
         try:
-            resp = es.search(index=index, size=0, aggs={"tuples": {"composite": composite}})
+            kwargs = {"index": index, "size": 0, "aggs": {"tuples": {"composite": composite}}}
+            if update_mode is not None:
+                kwargs["query"] = {"term": {"update_mode": update_mode}}
+            resp = es.search(**kwargs)
         except NotFoundError:
             return out
         agg = resp["aggregations"]["tuples"]
diff --git a/src/sourcerer/utils.py b/src/sourcerer/utils.py
index 206099b..ad22c09 100644
--- a/src/sourcerer/utils.py
+++ b/src/sourcerer/utils.py
@@ -1,5 +1,6 @@
 # Standard packages
 import hashlib
+import json
 
 # Elastic packages
 from elasticsearch import Elasticsearch, ApiError, TransportError
@@ -27,6 +28,23 @@ def make_doc_id(*parts: str) -> str:
     return hashlib.blake2b(joined, digest_size=ID_DIGEST_SIZE).hexdigest()
 
 
+def build_ref_key(org: str, repo: str, ref: str) -> str:
+    """Deterministic, case-sensitive ref key shared by incremental content and refs docs.
+
+    JSON-encodes the compact array [org.lower(), repo.lower(), "branch", ref] with UTF-8 and
+    no whitespace, so a ref name containing '/' or other punctuation (e.g. "release/8.x") can't
+    collide with a different (org, repo, ref) triple the way naive delimiter concatenation
+    would. org/repo are lowercased -- the physical index name and every query already fold
+    case there -- while the ref is left exactly as Git spells it, because Git ref names are
+    case-sensitive ("Main" and "main" are different branches). This is the single source of
+    lowercasing for incremental: the mappings carry NO normalizer (INV-003), so identity and query
+    scope agree only because this builder (and the matching _id builders) normalize the same
+    way.
+    """
+    return json.dumps([org.lower(), repo.lower(), "branch", ref],
+                      ensure_ascii=False, separators=(",", ":"))
+
+
 # Larger bulk batches sent concurrently (see parallel_bulk in commands/index/documents.py) take
 # longer per request, so give them a generous timeout and let the client retry transient timeouts.
 CLIENT_OPTS = {"request_timeout": 120, "max_retries": 3, "retry_on_timeout": True}
diff --git a/tests/test_agent_builder_assets.py b/tests/test_agent_builder_assets.py
new file mode 100644
index 0000000..4be3873
--- /dev/null
+++ b/tests/test_agent_builder_assets.py
@@ -0,0 +1,103 @@
+"""Structural tests for the unified-v1 Agent Builder assets."""
+
+import pathlib
+
+import pytest
+import yaml
+
+_ELASTIC = pathlib.Path(__file__).resolve().parents[1] / "src" / "sourcerer" / "elastic"
+_TOOLS_DIR = _ELASTIC / "agent_builder_tools"
+_SKILLS_DIR = _ELASTIC / "agent_builder_skills"
+_AGENTS_DIR = _ELASTIC / "agent_builder_agents"
+
+CONTENT_TOOLS = [
+    "sourcerer.code.search",
+    "sourcerer.code.grep",
+    "sourcerer.files.cat",
+    "sourcerer.files.head",
+    "sourcerer.files.tail",
+    "sourcerer.files.ls",
+]
+
+
+def _load(directory: pathlib.Path, name: str) -> dict:
+    return yaml.safe_load((directory / f"{name}.yml").read_text())
+
+
+def _tool(name: str) -> dict:
+    return _load(_TOOLS_DIR, name)
+
+
+def _query(name: str) -> str:
+    return _tool(name)["configuration"]["query"]
+
+
+class TestEveryYamlParses:
+    def test_all_assets_are_valid_yaml_with_ids(self):
+        for directory in (_TOOLS_DIR, _SKILLS_DIR, _AGENTS_DIR):
+            for path in sorted(directory.glob("*.yml")):
+                doc = yaml.safe_load(path.read_text())
+                assert isinstance(doc, dict), path
+                assert doc.get("id"), f"{path} missing id"
+
+
+class TestUnifiedContentTools:
+    @pytest.mark.parametrize("name", CONTENT_TOOLS)
+    def test_has_exact_ref_key_param(self, name):
+        params = _tool(name)["configuration"]["params"]
+        assert params["git_ref_key"]["optional"] is True
+        assert params["git_ref_key"]["defaultValue"] == ""
+
+    @pytest.mark.parametrize("name", CONTENT_TOOLS)
+    def test_uses_one_real_v1_pattern_and_no_incremental_or_anchor(self, name):
+        query = _query(name)
+        expected = "sourcerer-v1-files~*" if name.endswith(".ls") else "sourcerer-v1-lines~*"
+        assert f"FROM {expected}" in query
+        assert "sourcerer-incremental" not in query
+        assert "FROM sourcerer-v1-files*" not in query
+        assert "FROM sourcerer-v1-lines*" not in query
+
+    @pytest.mark.parametrize("name", CONTENT_TOOLS)
+    def test_mode_scoping_is_mutually_exclusive(self, name):
+        query = _query(name)
+        assert '?git_ref_key == ""' in query
+        assert 'update_mode == "snapshot"' in query
+        assert "git.commit LIKE ?git_commit" in query
+        assert '?git_ref_key != ""' in query
+        assert 'update_mode == "incremental"' in query
+        assert "git.ref_key == ?git_ref_key" in query
+
+    @pytest.mark.parametrize("name", CONTENT_TOOLS)
+    def test_incremental_commit_comes_from_unified_refs(self, name):
+        query = _query(name)
+        join = query.index("LOOKUP JOIN sourcerer-v1-refs ON git.ref_key")
+        coalesce = query.index("git.commit = COALESCE", join)
+        guard = query.index("| WHERE git.commit IS NOT NULL", coalesce)
+        assert join < coalesce < guard
+
+
+class TestRefsList:
+    def test_queries_only_unified_refs(self):
+        query = _query("sourcerer.refs.list")
+        assert "FROM sourcerer-v1-refs" in query
+        assert "sourcerer-incremental" not in query
+
+    def test_surfaces_mode_and_incremental_state(self):
+        query = _query("sourcerer.refs.list")
+        for field in ("update_mode", "status", "git.ref_key", "git.target_commit", "git.commit"):
+            assert field in query
+        assert '?git_commit == "*" OR git.commit LIKE ?git_commit' in query
+
+
+class TestSkillsDocumentIncremental:
+    def test_ref_resolution_mentions_ref_key_and_indexing_window(self):
+        text = (_SKILLS_DIR / "sourcerer-ref-resolution.yml").read_text()
+        assert "git_ref_key" in text
+        assert "incremental" in text
+        assert "status: indexing" in text
+        assert "mixed" in text.lower()
+
+    def test_citations_mentions_completed_commit(self):
+        text = (_SKILLS_DIR / "sourcerer-code-citations.yml").read_text()
+        assert "incremental" in text
+        assert "completed commit" in text.lower()
diff --git a/tests/test_config.py b/tests/test_config.py
index 180ea59..6329ad6 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -205,6 +205,49 @@ def test_retain_prerelease_superseded_raises(self):
             )])])
 
 
+class TestUpdateMode:
+    def test_omitted_defaults_to_snapshot(self):
+        cfgs = parse_config([_entry(refs=[_selector()])])
+        assert cfgs[0].selectors[0].update_mode == "snapshot"
+
+    def test_explicit_snapshot(self):
+        cfgs = parse_config([_entry(refs=[{"type": "branch", "match": "main", "update": "snapshot"}])])
+        assert cfgs[0].selectors[0].update_mode == "snapshot"
+
+    def test_valid_incremental_branch(self):
+        cfgs = parse_config([_entry(refs=[{"type": "branch", "match": "main", "update": "incremental"}])])
+        assert cfgs[0].selectors[0].update_mode == "incremental"
+
+    def test_unknown_value_raises(self):
+        with pytest.raises(ValueError, match="'update' must be"):
+            parse_config([_entry(refs=[{"type": "branch", "match": "main", "update": "delta"}])])
+
+    def test_incremental_tag_raises(self):
+        with pytest.raises(ValueError, match="only valid for 'type: branch'"):
+            parse_config([_entry(refs=[{"type": "tag", "match": "v{major}.{minor}.{patch}", "update": "incremental"}])])
+
+    def test_incremental_commit_raises(self):
+        with pytest.raises(ValueError, match="only valid for 'type: branch'"):
+            parse_config([_entry(refs=[{"type": "commit", "match": "cfefb3b", "update": "incremental"}])])
+
+    def test_incremental_with_since_raises(self):
+        with pytest.raises(ValueError, match="cannot be combined with 'since'"):
+            parse_config([_entry(refs=[{
+                "type": "branch", "match": "main", "update": "incremental", "since": {"age": "1y"},
+            }])])
+
+    def test_incremental_with_retain_raises(self):
+        with pytest.raises(ValueError, match="cannot be combined with 'retain'"):
+            parse_config([_entry(refs=[{
+                "type": "branch", "match": "main", "update": "incremental", "retain": {"count": 5},
+            }])])
+
+    def test_update_dotted_key(self):
+        # `update` is a plain scalar leaf, so dotted-key expansion leaves it untouched.
+        cfgs = parse_config([_entry(refs=[{"type": "branch", "match": "main", "update": "incremental"}])])
+        assert cfgs[0].selectors[0].update_mode == "incremental"
+
+
 class TestParseSince:
     def test_exactly_one_required_zero_raises(self):
         with pytest.raises(ValueError, match="exactly one"):
diff --git a/tests/test_documents.py b/tests/test_documents.py
index 2a937b4..50d368f 100644
--- a/tests/test_documents.py
+++ b/tests/test_documents.py
@@ -7,16 +7,27 @@
 import os
 import pathlib
 import stat
+from unittest.mock import MagicMock, patch
 
 # App packages
 from sourcerer.commands.index import documents
 from sourcerer.commands.index.documents import (
     build_file_actions,
+    build_incremental_file_actions,
     build_file_doc,
+    build_incremental_file_doc,
     file_attributes,
+    index_incremental_paths,
     iter_line_docs,
+    iter_incremental_line_docs,
 )
-from sourcerer.indices import files_index, lines_index
+from sourcerer.indices import (
+    files_index,
+    files_index,
+    lines_index,
+    lines_index,
+)
+from sourcerer.utils import build_ref_key, make_doc_id
 
 
 def _set_worker_ctx(org: str, repo: str, commit_sha: str, repo_dir) -> None:
@@ -70,6 +81,119 @@ def test_git_fields(self, tmp_path):
         assert doc["git"] == {"org": "acme", "repo": "widgets", "commit": "deadbeef"}
 
 
+class TestSnapshotDocsUnchanged:
+    """Guard that the v1 (snapshot) builders keep commit-addressed identity and never grow
+    incremental's ref fields -- the isolation half of INV-009."""
+
+    def test_snapshot_file_doc_keeps_commit_and_no_ref_key(self, tmp_path):
+        p = tmp_path / "a.txt"
+        p.write_text("hi")
+        _id, doc = build_file_doc("acme", "widgets", "deadbeef", "a.txt", p)
+        assert doc["git"]["commit"] == "deadbeef"
+        assert "ref_key" not in doc["git"]
+        assert "ref" not in doc["git"]
+
+    def test_snapshot_id_depends_on_commit(self, tmp_path):
+        p = tmp_path / "a.txt"
+        p.write_text("hi")
+        id1, _ = build_file_doc("acme", "widgets", "c0ffee", "a.txt", p)
+        id2, _ = build_file_doc("acme", "widgets", "deadbeef", "a.txt", p)
+        assert id1 != id2  # snapshot content is per-commit
+
+
+class TestBuildFileDocIncremental:
+    def test_source_has_ref_fields_and_no_commit(self, tmp_path):
+        p = tmp_path / "a.txt"
+        p.write_text("hello")
+        _id, doc = build_incremental_file_doc("acme", "widgets", "main", "src/a.txt", p)
+        assert doc["git"]["ref"] == "main"
+        assert doc["git"]["ref_type"] == "branch"
+        assert doc["git"]["ref_key"] == build_ref_key("acme", "widgets", "main")
+        assert "commit" not in doc["git"]
+        assert doc["file"]["directory"] == "src"
+        assert doc["file"]["name"] == "a.txt"
+
+    def test_org_repo_lowercased_in_source_and_id(self, tmp_path):
+        p = tmp_path / "a.txt"
+        p.write_text("hello")
+        _id, doc = build_incremental_file_doc("ACME", "Widgets", "main", "a.txt", p)
+        assert doc["git"]["org"] == "acme"
+        assert doc["git"]["repo"] == "widgets"
+        assert _id == make_doc_id("acme", "widgets", "branch", "main", "a.txt")
+
+    def test_id_is_ref_addressed_not_commit_addressed(self, tmp_path):
+        # INV-003: identity is (org, repo, "branch", ref, path) with no commit -- two runs at
+        # two different commit SHAs for the same branch/path collapse to one id.
+        p = tmp_path / "a.txt"
+        p.write_text("hello")
+        id1, _ = build_incremental_file_doc("acme", "widgets", "main", "a.txt", p)
+        id2, _ = build_incremental_file_doc("acme", "widgets", "main", "a.txt", p)
+        assert id1 == id2
+
+    def test_branch_case_sensitivity_changes_id(self, tmp_path):
+        p = tmp_path / "a.txt"
+        p.write_text("hello")
+        lower, _ = build_incremental_file_doc("acme", "widgets", "main", "a.txt", p)
+        upper, _ = build_incremental_file_doc("acme", "widgets", "Main", "a.txt", p)
+        assert lower != upper
+
+
+class TestIterLineDocsIncremental:
+    def test_line_id_is_ref_addressed(self):
+        docs = list(iter_incremental_line_docs("acme", "widgets", "main", "a.txt", "one\ntwo"))
+        first_id, first_doc = docs[0]
+        assert first_id == make_doc_id("acme", "widgets", "branch", "main", "a.txt", "1")
+        assert "commit" not in first_doc["git"]
+        assert first_doc["git"]["ref"] == "main"
+
+    def test_line_numbering_and_content(self):
+        docs = list(iter_incremental_line_docs("acme", "widgets", "main", "a.txt", "one\ntwo\nthree"))
+        assert [d["line"]["number"] for _i, d in docs] == [1, 2, 3]
+        assert [d["line"]["content"] for _i, d in docs] == ["one", "two", "three"]
+
+
+class TestBuildFileActionsIncremental:
+    def test_text_file_yields_file_and_line_actions(self, tmp_path):
+        (tmp_path / "a.txt").write_text("one\ntwo\n")
+        actions = build_incremental_file_actions("acme", "widgets", "main", tmp_path, "a.txt")
+        assert actions[0]["_index"] == files_index("acme", "widgets")
+        line_actions = [a for a in actions if a["_index"] == lines_index("acme", "widgets")]
+        assert len(line_actions) == 2
+
+    def test_missing_path_yields_no_actions(self, tmp_path):
+        # A path absent from the checked-out tree must not create a phantom document.
+        actions = build_incremental_file_actions("acme", "widgets", "main", tmp_path, "gone.txt")
+        assert actions == []
+
+    def test_binary_file_yields_only_file_doc(self, tmp_path):
+        (tmp_path / "b.bin").write_bytes(b"\x00\x01\x02binary")
+        actions = build_incremental_file_actions("acme", "widgets", "main", tmp_path, "b.bin")
+        assert len(actions) == 1
+        assert actions[0]["_index"] == files_index("acme", "widgets")
+
+
+class TestIndexPathsIncremental:
+    def test_emits_actions_only_for_supplied_paths(self, tmp_path):
+        (tmp_path / "a.txt").write_text("one\n")
+        (tmp_path / "b.txt").write_text("two\n")
+        (tmp_path / "c.txt").write_text("three\n")  # present but NOT supplied
+        captured = []
+
+        def fake_parallel_bulk(es, actions, **kwargs):
+            for a in actions:
+                captured.append(a)
+                yield True, {"index": {"_index": a["_index"], "_id": a["_id"]}}
+
+        with patch.object(documents, "es_parallel_bulk", side_effect=fake_parallel_bulk):
+            files_count, lines_count = index_incremental_paths(
+                MagicMock(), "acme", "widgets", tmp_path, "main", ["a.txt", "b.txt"],
+            )
+        paths = {a["_source"]["file"]["path"] for a in captured}
+        assert paths == {"a.txt", "b.txt"}  # c.txt never indexed
+        assert files_count == 2
+        assert lines_count == 2
+
+
 class TestIterLineDocs:
     def test_line_numbering_starts_at_one(self):
         docs = list(iter_line_docs("acme", "widgets", "deadbeef", "a.txt", "one\ntwo\nthree"))
diff --git a/tests/test_git_changes.py b/tests/test_git_changes.py
new file mode 100644
index 0000000..08043bd
--- /dev/null
+++ b/tests/test_git_changes.py
@@ -0,0 +1,166 @@
+"""Tests for the NUL-safe Git diff planner in sourcerer.commands.index.git.
+
+Two layers:
+  1. Integration against a real temporary Git repository for the common change kinds (add,
+     modify, delete, rename, type change) and awkward path bytes (spaces, tabs, Unicode).
+  2. Pure-parser tests over crafted `-z` byte streams for copy/rename records and truncation,
+     which are hard to elicit deterministically from git itself.
+"""
+
+# Standard packages
+import os
+import subprocess
+
+# Third-party packages
+import pytest
+
+# App packages
+from sourcerer.commands.index.git import (
+    ChangePlan,
+    _parse_name_status_z,
+    base_commit_available,
+    plan_changes,
+)
+
+
+def _git(repo, *args, env=None):
+    subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True, env=env)
+
+
+def _commit_all(repo, message):
+    env = {
+        **os.environ,
+        "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@e",
+        "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@e",
+    }
+    _git(repo, "add", "-A", env=env)
+    _git(repo, "commit", "-m", message, env=env)
+    out = subprocess.run(
+        ["git", "-C", str(repo), "rev-parse", "HEAD"],
+        check=True, capture_output=True, text=True, env=env,
+    )
+    return out.stdout.strip()
+
+
+@pytest.fixture
+def repo(tmp_path):
+    _git(tmp_path, "init")
+    _git(tmp_path, "config", "user.email", "t@e")
+    _git(tmp_path, "config", "user.name", "t")
+    return tmp_path
+
+
+class TestBaseCommitAvailable:
+    def test_true_for_present_commit(self, repo):
+        (repo / "a.txt").write_text("hi\n")
+        sha = _commit_all(repo, "init")
+        assert base_commit_available(repo, sha) is True
+
+    def test_false_for_absent_commit(self, repo):
+        (repo / "a.txt").write_text("hi\n")
+        _commit_all(repo, "init")
+        assert base_commit_available(repo, "0" * 40) is False
+
+
+class TestPlanChangesIntegration:
+    def test_base_missing_returns_flagged_plan(self, repo):
+        (repo / "a.txt").write_text("hi\n")
+        new = _commit_all(repo, "init")
+        plan = plan_changes(repo, "0" * 40, new)
+        assert plan.base_missing is True
+        assert plan.delete_paths == [] and plan.index_paths == []
+
+    def test_add_modify_delete(self, repo):
+        (repo / "keep.txt").write_text("keep\n")
+        (repo / "gone.txt").write_text("gone\n")
+        (repo / "mod.txt").write_text("v1\n")
+        old = _commit_all(repo, "init")
+        (repo / "gone.txt").unlink()
+        (repo / "mod.txt").write_text("incremental\n")
+        (repo / "new.txt").write_text("new\n")
+        new = _commit_all(repo, "change")
+        plan = plan_changes(repo, old, new)
+        assert set(plan.index_paths) == {"mod.txt", "new.txt"}
+        assert set(plan.delete_paths) == {"gone.txt", "mod.txt"}
+
+    def test_rename(self, repo):
+        (repo / "old_name.txt").write_text("stable content here\nline two\n")
+        old = _commit_all(repo, "init")
+        _git(repo, "mv", "old_name.txt", "new_name.txt")
+        new = _commit_all(repo, "rename")
+        plan = plan_changes(repo, old, new)
+        assert "old_name.txt" in plan.delete_paths
+        assert "new_name.txt" in plan.index_paths
+        assert "old_name.txt" not in plan.index_paths
+
+    def test_type_change_file_to_symlink(self, repo):
+        (repo / "target.txt").write_text("target\n")
+        (repo / "thing").write_text("regular\n")
+        old = _commit_all(repo, "init")
+        (repo / "thing").unlink()
+        os.symlink("target.txt", repo / "thing")
+        new = _commit_all(repo, "typechange")
+        plan = plan_changes(repo, old, new)
+        # A type change replaces the whole file: delete the stale docs and re-index.
+        assert "thing" in plan.delete_paths
+        assert "thing" in plan.index_paths
+
+    def test_awkward_paths_preserved(self, repo):
+        (repo / "seed.txt").write_text("seed\n")
+        old = _commit_all(repo, "init")
+        weird = "dir with spaces/tab\tfile.txt"
+        unicode_path = "café/naïve.txt"
+        (repo / "dir with spaces").mkdir()
+        (repo / weird).write_text("x\n")
+        (repo / "café").mkdir()
+        (repo / unicode_path).write_text("y\n")
+        new = _commit_all(repo, "add weird paths")
+        plan = plan_changes(repo, old, new)
+        assert weird in plan.index_paths
+        assert unicode_path in plan.index_paths
+
+
+class TestParseNameStatusZ:
+    def test_copy_indexes_destination_only(self):
+        raw = b"C100\x00src.txt\x00dst.txt\x00"
+        delete, index = _parse_name_status_z(raw)
+        assert delete == []
+        assert index == ["dst.txt"]
+
+    def test_rename_deletes_source_indexes_destination(self):
+        raw = b"R100\x00from.txt\x00to.txt\x00"
+        delete, index = _parse_name_status_z(raw)
+        assert delete == ["from.txt"]
+        assert index == ["to.txt"]
+
+    def test_mixed_stream(self):
+        raw = (
+            b"A\x00added.txt\x00"
+            b"M\x00mod.txt\x00"
+            b"D\x00del.txt\x00"
+            b"R096\x00old.txt\x00new.txt\x00"
+        )
+        delete, index = _parse_name_status_z(raw)
+        assert delete == ["mod.txt", "del.txt", "old.txt"]
+        assert index == ["added.txt", "mod.txt", "new.txt"]
+
+    def test_dedupe_preserves_order(self):
+        raw = b"M\x00a.txt\x00M\x00a.txt\x00"
+        delete, index = _parse_name_status_z(raw)
+        assert delete == ["a.txt"]
+        assert index == ["a.txt"]
+
+    def test_truncated_rename_record_ignored(self):
+        raw = b"R100\x00only_one_path.txt\x00"
+        delete, index = _parse_name_status_z(raw)
+        assert delete == [] and index == []
+
+    def test_empty_stream(self):
+        assert _parse_name_status_z(b"") == ([], [])
+
+
+class TestChangePlanDefaults:
+    def test_defaults(self):
+        plan = ChangePlan()
+        assert plan.delete_paths == [] and plan.index_paths == []
+        assert plan.base_missing is False
diff --git a/tests/test_incremental_deletions.py b/tests/test_incremental_deletions.py
new file mode 100644
index 0000000..1b281fb
--- /dev/null
+++ b/tests/test_incremental_deletions.py
@@ -0,0 +1,70 @@
+"""Unit tests for the synchronous, ref-scoped incremental deletion helpers in
+sourcerer.commands.index.markers. Every ES call is mocked; the point is to prove the query
+scope (exact git.ref_key + path terms, never a wildcard), the synchronous options
+(wait_for_completion + conflicts=proceed), and that only incremental content indices are targeted.
+"""
+
+# Standard packages
+from unittest.mock import MagicMock
+
+# Third-party packages
+from elastic_transport import ApiResponseMeta, HttpHeaders
+from elasticsearch import NotFoundError
+
+# App packages
+from sourcerer.commands.index.markers import delete_incremental_branch, delete_incremental_paths
+from sourcerer.indices import files_index, lines_index
+from sourcerer.utils import build_ref_key
+
+
+def _not_found() -> NotFoundError:
+    meta = ApiResponseMeta(status=404, http_version="1.1", headers=HttpHeaders({}), duration=0.0, node=None)
+    return NotFoundError("index_not_found_exception", meta, None)
+
+
+class TestDeleteIncrementalPaths:
+    def test_targets_both_incremental_content_indices(self):
+        es = MagicMock()
+        delete_incremental_paths(es, "acme", "widgets", "main", ["a.txt", "b.txt"])
+        indices = {c.kwargs["index"] for c in es.delete_by_query.call_args_list}
+        assert indices == {files_index("acme", "widgets"), lines_index("acme", "widgets")}
+
+    def test_query_scope_is_exact_ref_key_and_path_terms(self):
+        es = MagicMock()
+        delete_incremental_paths(es, "acme", "widgets", "main", ["a.txt", "b.txt"])
+        query = es.delete_by_query.call_args_list[0].kwargs["query"]
+        filters = query["bool"]["filter"]
+        assert {"term": {"git.ref_key": build_ref_key("acme", "widgets", "main")}} in filters
+        assert {"terms": {"file.path": ["a.txt", "b.txt"]}} in filters
+
+    def test_synchronous_options(self):
+        es = MagicMock()
+        delete_incremental_paths(es, "acme", "widgets", "main", ["a.txt"])
+        kwargs = es.delete_by_query.call_args_list[0].kwargs
+        assert kwargs["wait_for_completion"] is True
+        assert kwargs["conflicts"] == "proceed"
+
+    def test_empty_paths_is_noop(self):
+        es = MagicMock()
+        delete_incremental_paths(es, "acme", "widgets", "main", [])
+        es.delete_by_query.assert_not_called()
+
+    def test_missing_index_is_ignored(self):
+        es = MagicMock()
+        es.delete_by_query.side_effect = _not_found()
+        delete_incremental_paths(es, "acme", "widgets", "main", ["a.txt"])  # must not raise
+
+
+class TestDeleteIncrementalBranch:
+    def test_scoped_to_exact_ref_key_only(self):
+        es = MagicMock()
+        delete_incremental_branch(es, "acme", "widgets", "main")
+        query = es.delete_by_query.call_args_list[0].kwargs["query"]
+        filters = query["bool"]["filter"]
+        assert filters == [{"term": {"git.ref_key": build_ref_key("acme", "widgets", "main")}}]
+
+    def test_targets_both_incremental_content_indices(self):
+        es = MagicMock()
+        delete_incremental_branch(es, "acme", "widgets", "main")
+        indices = {c.kwargs["index"] for c in es.delete_by_query.call_args_list}
+        assert indices == {files_index("acme", "widgets"), lines_index("acme", "widgets")}
diff --git a/tests/test_incremental_index.py b/tests/test_incremental_index.py
new file mode 100644
index 0000000..bef4ac1
--- /dev/null
+++ b/tests/test_incremental_index.py
@@ -0,0 +1,370 @@
+"""Integration-style unit tests for the incremental (incremental) orchestration in
+sourcerer.commands.index.command.index_incremental_in_dir.
+
+Runs against a real temporary Git repository (so the diff planner, checkout, and tracked-file
+walk are exercised for real) plus a small stateful fake Elasticsearch client that models the
+incremental content and refs indices in memory. Covers the Definition of done: initial index, targeted
+change update, deletion/rename cleanup, no-op, injected-failure pointer safety + retry
+convergence, and missing-base full reconciliation.
+"""
+
+# Standard packages
+import os
+import subprocess
+from collections import defaultdict
+
+# Third-party packages
+import pytest
+from elastic_transport import ApiResponseMeta, HttpHeaders
+from elasticsearch import NotFoundError
+
+# App packages
+from sourcerer.commands.index import command, documents
+from sourcerer.indices import REFS_INDEX, files_index, lines_index
+from sourcerer.commands.index.markers import build_incremental_ref_id
+from sourcerer.utils import build_ref_key, make_doc_id
+
+ORG, REPO, BRANCH = "acme", "widgets", "main"
+
+
+# --- git helpers ----------------------------------------------------------------------------
+
+def _git(repo, *args):
+    env = {
+        **os.environ,
+        "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@e",
+        "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@e",
+    }
+    subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True, env=env)
+
+
+def _commit(repo, message):
+    _git(repo, "add", "-A")
+    _git(repo, "commit", "-m", message)
+    return subprocess.run(
+        ["git", "-C", str(repo), "rev-parse", "HEAD"],
+        check=True, capture_output=True, text=True,
+    ).stdout.strip()
+
+
+@pytest.fixture
+def repo(tmp_path):
+    _git(tmp_path, "init")
+    _git(tmp_path, "config", "user.email", "t@e")
+    _git(tmp_path, "config", "user.name", "t")
+    _git(tmp_path, "checkout", "-b", BRANCH)
+    return tmp_path
+
+
+# --- stateful fake Elasticsearch ------------------------------------------------------------
+
+def _not_found() -> NotFoundError:
+    meta = ApiResponseMeta(status=404, http_version="1.1", headers=HttpHeaders({}), duration=0.0, node=None)
+    return NotFoundError("not_found", meta, None)
+
+
+def _get_field(source: dict, dotted: str):
+    cur = source
+    for part in dotted.split("."):
+        if not isinstance(cur, dict) or part not in cur:
+            return None
+        cur = cur[part]
+    return cur
+
+
+def _matches(source: dict, query: dict) -> bool:
+    for clause in query["bool"]["filter"]:
+        if "term" in clause:
+            (field, value), = clause["term"].items()
+            if _get_field(source, field) != value:
+                return False
+        elif "terms" in clause:
+            (field, values), = clause["terms"].items()
+            if _get_field(source, field) not in values:
+                return False
+    return True
+
+
+class _Indices:
+    def refresh(self, **kwargs):
+        pass
+
+
+class FakeES:
+    def __init__(self):
+        self.store: dict[str, dict[str, dict]] = defaultdict(dict)
+        self.indices = _Indices()
+        self.ref_write_history: list[tuple[str, str | None]] = []
+        self.fail_ready_once = False
+
+    def index(self, *, index, id, document, refresh=False):
+        if index == REFS_INDEX:
+            self.ref_write_history.append((document["status"], document["git"]["commit"]))
+        self.store[index][id] = document
+
+    def get(self, *, index, id):
+        bucket = self.store.get(index, {})
+        if id not in bucket:
+            raise _not_found()
+        return {"_source": bucket[id]}
+
+    def count(self, *, index, query):
+        bucket = self.store.get(index, {})
+        return {"count": sum(1 for s in bucket.values() if _matches(s, query))}
+
+    def delete_by_query(self, *, index, query, **kwargs):
+        bucket = self.store.get(index, {})
+        doomed = [i for i, s in bucket.items() if _matches(s, query)]
+        for i in doomed:
+            del bucket[i]
+        return {"deleted": len(doomed)}
+
+
+def _fake_parallel_bulk(es, actions, **kwargs):
+    for a in actions:
+        es.store[a["_index"]][a["_id"]] = a["_source"]
+        yield True, {"index": {"_index": a["_index"], "_id": a["_id"]}}
+
+
+@pytest.fixture(autouse=True)
+def patch_bulk_and_checkout(monkeypatch):
+    # Patch the bulk helper to write into the fake store, and checkout to a plain local branch
+    # checkout (the test repo has no `origin` remote to reset against).
+    monkeypatch.setattr(documents, "es_parallel_bulk", _fake_parallel_bulk)
+
+    def _local_checkout(repo_dir, branch):
+        subprocess.run(
+            ["git", "-C", str(repo_dir), "checkout", "--force", branch],
+            check=True, capture_output=True,
+        )
+
+    monkeypatch.setattr(command, "checkout_branch", _local_checkout)
+
+
+# --- helpers to inspect the fake store ------------------------------------------------------
+
+def _files(es):
+    return es.store.get(files_index(ORG, REPO), {})
+
+
+def _lines(es):
+    return es.store.get(lines_index(ORG, REPO), {})
+
+
+def _ref_doc(es):
+    return es.store[REFS_INDEX][build_incremental_ref_id(ORG, REPO, BRANCH)]
+
+
+def _run(es, repo, force=False):
+    command.index_incremental_in_dir(es, ORG, REPO, repo, BRANCH, force=force)
+
+
+def _file_id(path):
+    return make_doc_id(ORG, REPO, "branch", BRANCH, path)
+
+
+# --- tests ----------------------------------------------------------------------------------
+
+class TestInitialIndex:
+    def test_full_branch_view_and_ready_marker(self, repo):
+        (repo / "a.txt").write_text("one\ntwo\n")
+        (repo / "b.txt").write_text("hello\n")
+        sha = _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)
+
+        assert _file_id("a.txt") in _files(es)
+        assert _file_id("b.txt") in _files(es)
+        ref = _ref_doc(es)
+        assert ref["status"] == "ready"
+        assert ref["git"]["commit"] == sha
+        assert ref["git"]["target_commit"] is None
+        assert ref["git"]["ref_key"] == build_ref_key(ORG, REPO, BRANCH)
+        assert ref["files_count"] == 2
+        assert ref["lines_count"] == 3
+
+
+class TestTargetedUpdate:
+    def test_only_changed_paths_touched(self, repo):
+        (repo / "keep.txt").write_text("unchanged\n")
+        (repo / "mod.txt").write_text("v1\n")
+        (repo / "gone.txt").write_text("bye\n")
+        _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)
+        keep_line_id_before = make_doc_id(ORG, REPO, "branch", BRANCH, "keep.txt", "1")
+        assert keep_line_id_before in _lines(es)
+
+        (repo / "mod.txt").write_text("v1\nincremental\n")
+        (repo / "gone.txt").unlink()
+        (repo / "new.txt").write_text("fresh\n")
+        sha2 = _commit(repo, "change")
+        _run(es, repo)
+
+        # Deleted file leaves no docs.
+        assert _file_id("gone.txt") not in _files(es)
+        assert make_doc_id(ORG, REPO, "branch", BRANCH, "gone.txt", "1") not in _lines(es)
+        # New + modified files present; modified file has both lines now.
+        assert _file_id("new.txt") in _files(es)
+        assert make_doc_id(ORG, REPO, "branch", BRANCH, "mod.txt", "2") in _lines(es)
+        # Unchanged file's line doc id is stable across the update.
+        assert keep_line_id_before in _lines(es)
+        ref = _ref_doc(es)
+        assert ref["status"] == "ready" and ref["git"]["commit"] == sha2
+
+    def test_rename_leaves_no_source_docs(self, repo):
+        (repo / "old.txt").write_text("stable content\nsecond line\n")
+        _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)
+        assert _file_id("old.txt") in _files(es)
+
+        _git(repo, "mv", "old.txt", "new.txt")
+        _commit(repo, "rename")
+        _run(es, repo)
+
+        assert _file_id("old.txt") not in _files(es)
+        assert make_doc_id(ORG, REPO, "branch", BRANCH, "old.txt", "1") not in _lines(es)
+        assert _file_id("new.txt") in _files(es)
+
+
+class TestNoOp:
+    def test_unchanged_head_writes_no_content(self, repo):
+        (repo / "a.txt").write_text("one\n")
+        _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)
+        files_snapshot = dict(_files(es))
+        history_len = len(es.ref_write_history)
+
+        _run(es, repo)  # nothing changed
+
+        assert dict(_files(es)) == files_snapshot  # no content writes
+        # No new refs writes beyond the first run's indexing+ready pair.
+        assert len(es.ref_write_history) == history_len
+
+
+class TestFailureAndRetry:
+    def test_failure_never_advances_pointer_and_retry_converges(self, repo, monkeypatch):
+        (repo / "a.txt").write_text("one\n")
+        sha1 = _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)  # clean initial index at sha1
+
+        (repo / "a.txt").write_text("one\ntwo\n")
+        (repo / "b.txt").write_text("new file\n")
+        sha2 = _commit(repo, "change")
+
+        # Inject an indexing failure for the update run.
+        def boom(*a, **k):
+            raise RuntimeError("bulk exploded")
+
+        monkeypatch.setattr(command, "index_incremental_paths", boom)
+        with pytest.raises(RuntimeError):
+            _run(es, repo)
+
+        ref = _ref_doc(es)
+        assert ref["status"] == "indexing"        # not advanced past the failure
+        assert ref["git"]["commit"] == sha1        # completed pointer held at old SHA
+        assert ref["git"]["target_commit"] == sha2
+        assert ref["error"] == "bulk exploded"
+        assert ref["failed_at"] is not None
+        # No refs write during the whole failed run set git.commit to the candidate SHA.
+        assert all(commit != sha2 for _status, commit in es.ref_write_history)
+
+        # Retry (failure cleared): restore the real ingest and converge to the same view a
+        # clean run would produce. Only the boom patch is reverted -- the autouse bulk/checkout
+        # patches must stay in place.
+        monkeypatch.setattr(command, "index_incremental_paths", documents.index_incremental_paths)
+        _run(es, repo)
+        ref = _ref_doc(es)
+        assert ref["status"] == "ready"
+        assert ref["git"]["commit"] == sha2
+        assert ref["error"] is None and ref["failed_at"] is None
+        assert make_doc_id(ORG, REPO, "branch", BRANCH, "a.txt", "2") in _lines(es)
+        assert _file_id("b.txt") in _files(es)
+
+
+class TestMissingBaseReconciliation:
+    def test_unavailable_old_commit_triggers_full_rebuild(self, repo, monkeypatch):
+        (repo / "a.txt").write_text("one\n")
+        _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)
+
+        # Pretend the completed base commit is gone: force a full-namespace delete + rebuild.
+        monkeypatch.setattr(command, "base_commit_available", lambda repo_dir, sha: False, raising=False)
+        # base_commit_available is used inside git.plan_changes; patch there.
+        import sourcerer.commands.index.git as gitmod
+        monkeypatch.setattr(gitmod, "base_commit_available", lambda repo_dir, sha: False)
+
+        (repo / "a.txt").write_text("one\ntwo\n")
+        (repo / "c.txt").write_text("added\n")
+        sha2 = _commit(repo, "change")
+        _run(es, repo)
+
+        ref = _ref_doc(es)
+        assert ref["status"] == "ready" and ref["git"]["commit"] == sha2
+        # Full rebuild indexed every current file with current content.
+        assert _file_id("a.txt") in _files(es)
+        assert _file_id("c.txt") in _files(es)
+        assert make_doc_id(ORG, REPO, "branch", BRANCH, "a.txt", "2") in _lines(es)
+
+
+class TestRunConfigRouting:
+    def test_snapshot_units_use_v1_path_incremental_units_use_incremental_path(self, tmp_path, monkeypatch):
+        import contextlib
+        from unittest.mock import MagicMock
+        from sourcerer.config import RepoConfig
+        from sourcerer.progress import Unit
+
+        snap = Unit(org=ORG, repo=REPO, ref="release", kind="branch", update_mode="snapshot")
+        incr = Unit(org=ORG, repo=REPO, ref="main", kind="branch", update_mode="incremental")
+
+        v1_calls: list[str] = []
+        incremental_calls: list[str] = []
+
+        def fake_v1(es, org, repo, repo_dir, branch, tag, commit, force, reporter, unit):
+            v1_calls.append(unit.ref)
+            unit.status = "indexed"
+
+        def fake_incremental(es, org, repo, repo_dir, branch, force=False, reporter=None, unit=None):
+            incremental_calls.append(branch)
+            unit.status = "indexed"
+
+        @contextlib.contextmanager
+        def fake_prepared_repo(org, repo, cache_root, ephemeral):
+            yield tmp_path
+
+        monkeypatch.setattr(command, "make_client", lambda *a, **k: MagicMock())
+        monkeypatch.setattr(command, "_load_config",
+                            lambda p: [RepoConfig(org=ORG, repo=REPO, selectors=[])])
+        monkeypatch.setattr(command, "_resolve_entry", lambda entry: [snap, incr])
+        monkeypatch.setattr(command, "prepared_repo", fake_prepared_repo)
+        monkeypatch.setattr(command, "pre_clone_skip", lambda *a, **k: (False, "release", "sha"))
+        monkeypatch.setattr(command, "_rev_info", lambda repo_dir, ref: ("sha", None))
+        monkeypatch.setattr(command, "_effective_since_floor", lambda *a, **k: None)
+        monkeypatch.setattr(command, "plan_repo", lambda *a, **k: [])
+        monkeypatch.setattr(command, "ref_dates", lambda repo_dir: {})
+        monkeypatch.setattr(command, "index_ref_in_dir", fake_v1)
+        monkeypatch.setattr(command, "index_incremental_in_dir", fake_incremental)
+
+        command.run_config("unused.yml", "http://es", None, None, None, quiet=True)
+
+        assert v1_calls == ["release"]  # snapshot ref went through the v1 path
+        assert incremental_calls == ["main"]     # incremental ref went through the incremental path
+
+
+class TestForceFullReconciliation:
+    def test_force_rebuilds_even_when_head_matches(self, repo):
+        (repo / "a.txt").write_text("one\n")
+        _commit(repo, "init")
+        es = FakeES()
+        _run(es, repo)
+        history_len = len(es.ref_write_history)
+
+        _run(es, repo, force=True)  # HEAD unchanged, but --force rebuilds anyway
+
+        # A rebuild happened (indexing + ready), so history grew rather than a no-op skip.
+        assert len(es.ref_write_history) > history_len
+        assert _ref_doc(es)["status"] == "ready"
diff --git a/tests/test_index_orphans.py b/tests/test_index_orphans.py
index 8e013ac..f1a9f1c 100644
--- a/tests/test_index_orphans.py
+++ b/tests/test_index_orphans.py
@@ -16,6 +16,7 @@
 from sourcerer.indices import FILES_INDEX_PREFIX, LINES_INDEX_PREFIX, REFS_INDEX
 from sourcerer.queries import (
     enumerate_content_commits,
+    enumerate_ref_repositories,
     enumerate_ref_tuples,
     gather_content_commit_tuples,
     list_sourcerer_indices,
@@ -54,6 +55,31 @@ def test_queries_only_files_and_lines_prefixes(self):
 
 
 class TestCompositeTuples:
+    def test_ref_commit_scan_filters_snapshot_mode(self):
+        es = MagicMock()
+        es.search.return_value = _composite_response([], after_key=None)
+        enumerate_ref_tuples(es)
+        assert es.search.call_args.kwargs["query"] == {"term": {"update_mode": "snapshot"}}
+
+    def test_content_commit_scan_filters_snapshot_mode(self):
+        es = MagicMock()
+        es.search.return_value = _composite_response([], after_key=None)
+        enumerate_content_commits(es, "sourcerer-v1-files~acme~widgets")
+        assert es.search.call_args.kwargs["query"] == {"term": {"update_mode": "snapshot"}}
+
+    def test_ref_repositories_include_all_modes(self):
+        es = MagicMock()
+        es.search.return_value = {
+            "aggregations": {"repos": {"buckets": [
+                {"key": {"org": "acme", "repo": "snapshot"}},
+                {"key": {"org": "acme", "repo": "incremental"}},
+            ]}}
+        }
+        assert enumerate_ref_repositories(es) == {
+            ("acme", "snapshot"), ("acme", "incremental")
+        }
+        assert "query" not in es.search.call_args.kwargs
+
     def test_paginates_until_no_after_key(self):
         es = MagicMock()
         es.search.side_effect = [
diff --git a/tests/test_markers.py b/tests/test_markers.py
index e2f8016..9b3ab53 100644
--- a/tests/test_markers.py
+++ b/tests/test_markers.py
@@ -10,7 +10,18 @@
 from elasticsearch import NotFoundError
 
 # App packages
-from sourcerer.commands.index.markers import commit_prefix_indexed, pre_clone_skip
+from sourcerer.commands.index.markers import (
+    ERROR_MAX_LEN,
+    build_incremental_ref_id,
+    commit_prefix_indexed,
+    pre_clone_skip,
+    read_incremental_ref,
+    write_incremental_failed,
+    write_incremental_indexing,
+    write_incremental_ready,
+    write_ref_marker,
+)
+from sourcerer.indices import REFS_INDEX
 
 FULL_SHA = "cfefb3b2378ccbadefa7c8f4f9e21b3a1d2e5f60"
 
@@ -73,3 +84,111 @@ def test_no_matching_marker_falls_through_to_clone(self):
             es, "acme", "widgets", None, None, "cfefb3b", False,
         )
         assert (skip, ref_for_id, remote_sha) == (False, None, None)
+
+
+OLD = "1111111111111111111111111111111111111111"
+NEW = "2222222222222222222222222222222222222222"
+
+
+class TestSnapshotMarkerIdentity:
+    def test_snapshot_ref_keys_are_unique_for_refs_at_same_commit(self):
+        first = MagicMock()
+        second = MagicMock()
+        write_ref_marker(first, "Acme", "Widgets", "branch", "main", OLD, None, 1, 2)
+        write_ref_marker(second, "Acme", "Widgets", "tag", "main", OLD, None, 1, 2)
+
+        first_call = first.index.call_args.kwargs
+        second_call = second.index.call_args.kwargs
+        first_doc = first_call["document"]
+        second_doc = second_call["document"]
+        assert first_call["id"] != second_call["id"]
+        assert first_doc["update_mode"] == "snapshot"
+        assert first_doc["git"]["org"] == "acme"
+        assert first_doc["git"]["repo"] == "widgets"
+        assert first_doc["git"]["commit"] == OLD.lower()
+        assert first_doc["git"]["ref_key"] != second_doc["git"]["ref_key"]
+
+
+class TestIncrementalRefId:
+    def test_stable_across_calls_and_commit_independent(self):
+        a = build_incremental_ref_id("acme", "widgets", "main")
+        b = build_incremental_ref_id("acme", "widgets", "main")
+        assert a == b  # one document per branch, no commit folded in
+
+    def test_org_repo_case_insensitive_but_ref_case_sensitive(self):
+        assert build_incremental_ref_id("Acme", "Widgets", "main") == build_incremental_ref_id("acme", "widgets", "main")
+        assert build_incremental_ref_id("acme", "widgets", "Main") != build_incremental_ref_id("acme", "widgets", "main")
+
+
+def _indexed_doc(es):
+    return es.index.call_args.kwargs["document"]
+
+
+class TestWriteIncrementalIndexing:
+    def test_preserves_completed_commit_and_exposes_target(self):
+        es = MagicMock()
+        write_incremental_indexing(es, "acme", "widgets", "main", completed_commit=OLD, target_commit=NEW)
+        doc = _indexed_doc(es)
+        assert doc["status"] == "indexing"
+        assert doc["git"]["commit"] == OLD  # completed pointer unchanged
+        assert doc["git"]["target_commit"] == NEW  # candidate advertised
+        assert doc["update_mode"] == "incremental"
+        assert es.index.call_args.kwargs["id"] == build_incremental_ref_id("acme", "widgets", "main")
+        assert es.index.call_args.kwargs["index"] == REFS_INDEX
+
+    def test_first_index_has_no_completed_commit(self):
+        es = MagicMock()
+        write_incremental_indexing(es, "acme", "widgets", "main", completed_commit=None, target_commit=NEW)
+        assert _indexed_doc(es)["git"]["commit"] is None
+
+    def test_carries_prior_counts(self):
+        es = MagicMock()
+        prior = {"files_count": 12, "lines_count": 340, "git": {"commit_date": "2026-01-01T00:00:00+00:00"}}
+        write_incremental_indexing(es, "acme", "widgets", "main", OLD, NEW, prior=prior)
+        doc = _indexed_doc(es)
+        assert doc["files_count"] == 12 and doc["lines_count"] == 340
+        assert doc["git"]["commit_date"] == "2026-01-01T00:00:00+00:00"
+
+
+class TestWriteIncrementalReady:
+    def test_advances_commit_and_clears_target_and_error(self):
+        es = MagicMock()
+        write_incremental_ready(es, "acme", "widgets", "main", commit=NEW,
+                       commit_date_iso="2026-02-02T00:00:00+00:00", files_count=5, lines_count=99)
+        doc = _indexed_doc(es)
+        assert doc["status"] == "ready"
+        assert doc["git"]["commit"] == NEW
+        assert doc["git"]["target_commit"] is None
+        assert doc["error"] is None and doc["failed_at"] is None
+        assert doc["files_count"] == 5 and doc["lines_count"] == 99
+        assert es.index.call_args.kwargs["refresh"] is True  # publication boundary
+
+
+class TestWriteIncrementalFailed:
+    def test_keeps_status_indexing_and_retains_old_pointer(self):
+        es = MagicMock()
+        write_incremental_failed(es, "acme", "widgets", "main", completed_commit=OLD,
+                        target_commit=NEW, error="boom")
+        doc = _indexed_doc(es)
+        assert doc["status"] == "indexing"  # not advanced
+        assert doc["git"]["commit"] == OLD
+        assert doc["git"]["target_commit"] == NEW
+        assert doc["error"] == "boom"
+        assert doc["failed_at"] is not None
+
+    def test_error_text_is_bounded(self):
+        es = MagicMock()
+        write_incremental_failed(es, "acme", "widgets", "main", OLD, NEW, error="x" * 5000)
+        assert len(_indexed_doc(es)["error"]) == ERROR_MAX_LEN
+
+
+class TestReadIncrementalRef:
+    def test_returns_source(self):
+        es = MagicMock()
+        es.get.return_value = {"_source": {"status": "ready", "git": {"commit": NEW}}}
+        assert read_incremental_ref(es, "acme", "widgets", "main") == {"status": "ready", "git": {"commit": NEW}}
+
+    def test_missing_returns_none(self):
+        es = MagicMock()
+        es.get.side_effect = _not_found()
+        assert read_incremental_ref(es, "acme", "widgets", "main") is None
diff --git a/tests/test_planner_orphans.py b/tests/test_planner_orphans.py
index 58c1e6f..d0714dd 100644
--- a/tests/test_planner_orphans.py
+++ b/tests/test_planner_orphans.py
@@ -180,6 +180,30 @@ def test_skip_repos_excludes_class_a_repos(self):
 
 
 class TestPlanOrphans:
+    def test_incremental_only_repository_keeps_shared_v1_indices(self):
+        names = ["sourcerer-v1-files~acme~iac", "sourcerer-v1-lines~acme~iac"]
+        plan = plan_orphans(
+            names,
+            ref_commit_tuples=set(),
+            content_commit_tuples=set(),
+            ref_repositories={("acme", "iac")},
+        )
+        assert plan.orphan_index_names == []
+        assert plan.orphan_content == {}
+        assert plan.orphan_marker_commits == {}
+
+    def test_mixed_repository_orphans_only_unreferenced_snapshot_commits(self):
+        names = ["sourcerer-v1-files~acme~mixed", "sourcerer-v1-lines~acme~mixed"]
+        plan = plan_orphans(
+            names,
+            ref_commit_tuples={("acme", "mixed", "kept")},
+            content_commit_tuples={("acme", "mixed", "kept"), ("acme", "mixed", "orphan")},
+            ref_repositories={("acme", "mixed")},
+        )
+        assert plan.orphan_index_names == []
+        assert plan.orphan_content == {("acme", "mixed"): {"orphan"}}
+        assert plan.orphan_marker_commits == {}
+
     def test_no_orphans(self):
         names = ["sourcerer-v1-files~acme~widgets", "sourcerer-v1-lines~acme~widgets"]
         ref_tuples = {("acme", "widgets", "aaa")}
diff --git a/tests/test_prune_deletions.py b/tests/test_prune_deletions.py
index 4ec6b0a..bcd3382 100644
--- a/tests/test_prune_deletions.py
+++ b/tests/test_prune_deletions.py
@@ -4,14 +4,19 @@
 deletes, query filters, single combined refs query), not against a real cluster."""
 
 # Standard packages
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 # Third-party packages
 from elastic_transport import ApiResponseMeta, HttpHeaders
 from elasticsearch import NotFoundError
 
 # App packages
-from sourcerer.commands.prune.execute import delete_index, execute_deletions, execute_orphan_deletions
+from sourcerer.commands.prune.execute import (
+    delete_index,
+    execute_deletions,
+    execute_orphan_deletions,
+    plan_orphans_now,
+)
 from sourcerer.indices import REFS_INDEX, files_index, lines_index
 from sourcerer.planner import Decision, Marker, OrphanPlan
 
@@ -21,6 +26,22 @@ def _not_found() -> NotFoundError:
     return NotFoundError("index_not_found_exception", meta, None)
 
 
+class TestPlanOrphansNow:
+    def test_passes_all_mode_repositories_to_planner(self):
+        es = MagicMock()
+        names = ["sourcerer-v1-files~acme~iac"]
+        repos = {("acme", "iac")}
+        with (
+            patch("sourcerer.commands.prune.execute.list_sourcerer_indices", return_value=names),
+            patch("sourcerer.commands.prune.execute.enumerate_ref_tuples", return_value=set()),
+            patch("sourcerer.commands.prune.execute.enumerate_ref_repositories", return_value=repos),
+            patch("sourcerer.commands.prune.execute.gather_content_commit_tuples", return_value=set()),
+            patch("sourcerer.commands.prune.execute.plan_orphans") as planner,
+        ):
+            plan_orphans_now(es)
+        planner.assert_called_once_with(names, set(), set(), repos)
+
+
 class TestDeleteIndex:
     def test_deletes_by_exact_name_no_wildcard(self):
         es = MagicMock()
@@ -81,6 +102,21 @@ def test_marker_delete_by_query_is_a_single_call_covering_all_repos(self):
         shoulds = kwargs["query"]["bool"]["should"]
         assert len(shoulds) == 2
         assert kwargs["query"]["bool"]["minimum_should_match"] == 1
+        for clause in shoulds:
+            filters = clause["bool"]["filter"]
+            assert {"term": {"update_mode": "snapshot"}} in filters
+
+    def test_snapshot_orphan_at_incremental_commit_cannot_delete_incremental_marker(self):
+        es = MagicMock()
+        plan = OrphanPlan(
+            orphan_index_names=[],
+            orphan_content={},
+            orphan_marker_commits={("acme", "widgets"): {"shared-sha"}},
+        )
+        execute_orphan_deletions(es, plan)
+        filters = es.delete_by_query.call_args.kwargs["query"]["bool"]["should"][0]["bool"]["filter"]
+        assert {"term": {"update_mode": "snapshot"}} in filters
+        assert {"terms": {"git.commit": ["shared-sha"]}} in filters
 
     def test_missing_content_index_is_swallowed(self):
         es = MagicMock()
diff --git a/tests/test_selection.py b/tests/test_selection.py
new file mode 100644
index 0000000..e0c0f4f
--- /dev/null
+++ b/tests/test_selection.py
@@ -0,0 +1,100 @@
+"""Unit tests for sourcerer.commands.index.selection._resolve_entry, focused on the
+incremental/snapshot mode plumbing: mode propagation onto resolved Units, same-mode
+deduplication, and the mixed-mode rejection when one concrete branch is selected in both
+modes. Remote ref listing is mocked so no network is touched."""
+
+# Standard packages
+from unittest.mock import patch
+
+# Third-party packages
+import pytest
+
+# App packages
+from sourcerer.config import parse_config
+from sourcerer.commands.index import selection
+
+
+def _resolve(refs, names_by_kind):
+    cfg = parse_config([{"org": "acme", "repo": "widgets", "refs": refs}])[0]
+
+    def fake_list(org, repo, kind):
+        return names_by_kind.get(kind, [])
+
+    with patch.object(selection, "list_remote_ref_names", side_effect=fake_list):
+        return selection._resolve_entry(cfg)
+
+
+class TestUpdateModePropagation:
+    def test_snapshot_default_propagates(self):
+        units = _resolve(
+            [{"type": "branch", "match": "main"}],
+            {"heads": ["main", "dev"]},
+        )
+        main = next(u for u in units if u.ref == "main")
+        assert main.update_mode == "snapshot"
+
+    def test_incremental_propagates(self):
+        units = _resolve(
+            [{"type": "branch", "match": "main", "update": "incremental"}],
+            {"heads": ["main"]},
+        )
+        assert len(units) == 1
+        assert units[0].update_mode == "incremental"
+        assert units[0].kind == "branch"
+
+
+class TestSameModeDedup:
+    def test_two_selectors_same_branch_same_mode_dedupe(self):
+        units = _resolve(
+            [
+                {"type": "branch", "match": "main"},
+                {"type": "branch", "match": "m*"},
+            ],
+            {"heads": ["main"]},
+        )
+        assert len([u for u in units if u.ref == "main"]) == 1
+        assert units[0].update_mode == "snapshot"
+
+    def test_two_incremental_selectors_same_branch_dedupe(self):
+        units = _resolve(
+            [
+                {"type": "branch", "match": "main", "update": "incremental"},
+                {"type": "branch", "match": "m*", "update": "incremental"},
+            ],
+            {"heads": ["main"]},
+        )
+        assert len(units) == 1
+        assert units[0].update_mode == "incremental"
+
+
+class TestMixedModeRejection:
+    def test_main_matched_by_exact_snapshot_and_glob_incremental_raises(self):
+        with pytest.raises(ValueError, match="both"):
+            _resolve(
+                [
+                    {"type": "branch", "match": "main"},
+                    {"type": "branch", "match": "m*", "update": "incremental"},
+                ],
+                {"heads": ["main", "other"]},
+            )
+
+    def test_incremental_first_then_snapshot_raises(self):
+        with pytest.raises(ValueError, match="incremental"):
+            _resolve(
+                [
+                    {"type": "branch", "match": "main", "update": "incremental"},
+                    {"type": "branch", "match": "main"},
+                ],
+                {"heads": ["main"]},
+            )
+
+    def test_disjoint_branches_in_different_modes_ok(self):
+        units = _resolve(
+            [
+                {"type": "branch", "match": "main"},
+                {"type": "branch", "match": "dev", "update": "incremental"},
+            ],
+            {"heads": ["main", "dev"]},
+        )
+        modes = {u.ref: u.update_mode for u in units}
+        assert modes == {"main": "snapshot", "dev": "incremental"}
diff --git a/tests/test_setup.py b/tests/test_setup.py
new file mode 100644
index 0000000..ce2f6ef
--- /dev/null
+++ b/tests/test_setup.py
@@ -0,0 +1,109 @@
+"""Unified-v1 setup and template contract tests."""
+
+import json
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+from sourcerer.commands.setup.command import (
+    ELASTICSEARCH_INDEX_TEMPLATES_DIR,
+    ensure_refs_index,
+    load_index_templates,
+    run,
+)
+from sourcerer.indices import REFS_INDEX
+
+_TEMPLATES = ELASTICSEARCH_INDEX_TEMPLATES_DIR
+
+
+def _load(name: str) -> dict:
+    return json.loads((_TEMPLATES / f"{name}.json").read_text())
+
+
+class TestLoadIndexTemplates:
+    def test_loads_only_unified_v1_templates(self):
+        es = MagicMock()
+        loaded = load_index_templates(es)
+        assert loaded == ["sourcerer-v1-files", "sourcerer-v1-lines", "sourcerer-v1-refs"]
+        es.indices.delete.assert_not_called()
+        es.indices.delete_index_template.assert_not_called()
+
+
+class TestUnifiedRefsTemplate:
+    def test_lookup_mode_and_one_shard(self):
+        index = _load("sourcerer-v1-refs")["template"]["settings"]["index"]
+        assert index["mode"] == "lookup"
+        assert index["number_of_shards"] == 1
+
+    def test_exact_index_pattern(self):
+        assert _load("sourcerer-v1-refs")["index_patterns"] == [REFS_INDEX]
+
+    def test_has_union_fields(self):
+        props = _load("sourcerer-v1-refs")["template"]["mappings"]["properties"]
+        git = props["git"]["properties"]
+        assert set(("ref_key", "org", "repo", "ref", "ref_type", "commit",
+                    "target_commit", "commit_date")) <= set(git)
+        assert set(("status", "update_mode", "files_count", "lines_count", "indexed_at",
+                    "update_started_at", "failed_at", "error")) <= set(props)
+
+
+class TestUnifiedContentTemplates:
+    def test_patterns_match_only_real_per_repo_indices(self):
+        assert _load("sourcerer-v1-files")["index_patterns"] == ["sourcerer-v1-files~*"]
+        assert _load("sourcerer-v1-lines")["index_patterns"] == ["sourcerer-v1-lines~*"]
+
+    def test_union_mode_and_git_fields(self):
+        for name in ("sourcerer-v1-files", "sourcerer-v1-lines"):
+            props = _load(name)["template"]["mappings"]["properties"]
+            assert props["update_mode"]["type"] == "keyword"
+            git = props["git"]["properties"]
+            assert set(("org", "repo", "commit", "ref_key", "ref", "ref_type")) <= set(git)
+
+    def test_identity_fields_are_exact_keywords(self):
+        for name in ("sourcerer-v1-files", "sourcerer-v1-lines", "sourcerer-v1-refs"):
+            git = _load(name)["template"]["mappings"]["properties"]["git"]["properties"]
+            for field in ("org", "repo", "commit", "ref_key", "ref", "ref_type"):
+                if field in git:
+                    assert git[field]["type"] == "keyword"
+                    assert "normalizer" not in git[field]
+
+
+class TestEnsureRefsIndex:
+    def test_creates_only_real_refs_lookup_index(self):
+        es = MagicMock()
+        es.indices.exists.return_value = False
+        assert ensure_refs_index(es) is True
+        es.indices.create.assert_called_once_with(index=REFS_INDEX)
+
+    def test_is_idempotent_for_lookup_index(self):
+        es = MagicMock()
+        es.indices.exists.return_value = True
+        es.indices.get_settings.return_value = {
+            REFS_INDEX: {"settings": {"index": {"mode": "lookup"}}}
+        }
+        assert ensure_refs_index(es) is False
+        es.indices.create.assert_not_called()
+
+    def test_rejects_existing_non_lookup_index(self):
+        es = MagicMock()
+        es.indices.exists.return_value = True
+        es.indices.get_settings.return_value = {
+            REFS_INDEX: {"settings": {"index": {"mode": "standard"}}}
+        }
+        with pytest.raises(ValueError, match="rebuild Sourcerer indices"):
+            ensure_refs_index(es)
+
+    def test_run_reports_non_lookup_rebuild_guidance(self, capsys):
+        es = MagicMock()
+        es.indices.exists.return_value = True
+        es.indices.get_settings.return_value = {
+            REFS_INDEX: {"settings": {"index": {"mode": "standard"}}}
+        }
+        with (
+            patch("sourcerer.commands.setup.command.make_client", return_value=es),
+            patch("sourcerer.commands.setup.command.load_index_templates", return_value=[]),
+            pytest.raises(SystemExit) as exit_info,
+        ):
+            run("http://local-dev", "key", None, None, None)
+        assert exit_info.value.code == 1
+        assert "rebuild Sourcerer indices" in capsys.readouterr().err