diff --git a/docs/feature-doc-linker.md b/docs/feature-doc-linker.md
index 796d7974f..cb182d7c7 100644
--- a/docs/feature-doc-linker.md
+++ b/docs/feature-doc-linker.md
@@ -1,6 +1,6 @@
 # Feature spec: Document / semantic linker (#435)
 
-**Status:** spec, no implementation
+**Status:** implementation shipped (PR for #435), bench-gate pending lab-side corpus
 **Issue:** #435
 **Recovery-inventory line:** [`docs/ROADMAP.md`](ROADMAP.md) — *"Doc / semantic linker | v2.0.0"*
 **Substrate prereqs:** belief schema (foundation), `ingest_log` (#205, v1.6.0), `belief_corroborations` (#190, v1.5.0), `DerivationInput.source_path` (`src/aelfrice/derivation.py:64`)
diff --git a/src/aelfrice/cli.py b/src/aelfrice/cli.py
index 0d4a131c8..fd1729180 100644
--- a/src/aelfrice/cli.py
+++ b/src/aelfrice/cli.py
@@ -41,6 +41,7 @@
 from pathlib import Path
 from typing import Any, Final, Sequence
 
+from aelfrice.doc_linker import ANCHOR_MANUAL, link_belief_to_document
 from aelfrice.auditor import (
     CORPUS_MIN_DEFAULT as AUDIT_CORPUS_MIN_DEFAULT,
     SEVERITY_FAIL as AUDIT_SEVERITY_FAIL,
@@ -1099,6 +1100,21 @@ def _cmd_lock(args: argparse.Namespace, out: object) -> int:
             print(f"locked: {actual_id} (corroborated existing)", file=out)  # type: ignore[arg-type]
         else:
             print(f"locked: {actual_id}", file=out)  # type: ignore[arg-type]
+
+        # #435 doc-linker manual anchor. Idempotent on (belief_id,
+        # doc_uri); subsequent lock --doc with the same URI is a no-op
+        # write. The lock entry-point passes source_path=None to the
+        # worker (cli_remember has no canonical document), so the
+        # ingest-time hook does NOT fire — manual is the only path that
+        # writes anchors here.
+        doc_uri = getattr(args, "doc_uri", None)
+        if doc_uri:
+            link_belief_to_document(
+                store,
+                actual_id,
+                doc_uri,
+                anchor_type=ANCHOR_MANUAL,
+            )
     finally:
         store.close()
     return 0
@@ -3667,6 +3683,14 @@ def build_parser(*, show_advanced: bool = False) -> argparse.ArgumentParser:
             "if neither is set (#192)."
         ),
     )
+    p_lock.add_argument(
+        "--doc", dest="doc_uri", default=None,
+        help=(
+            "optional doc URI to anchor on this belief (#435). Stored as "
+            "anchor_type='manual' on belief_documents; opaque to the "
+            "linker beyond non-empty (file:// or https:// recommended)."
+        ),
+    )
     p_lock.set_defaults(func=_cmd_lock)
 
     p_locked = sub.add_parser("locked", help="list locked beliefs")
diff --git a/src/aelfrice/derivation_worker.py b/src/aelfrice/derivation_worker.py
index e423a6145..5e17b53e6 100644
--- a/src/aelfrice/derivation_worker.py
+++ b/src/aelfrice/derivation_worker.py
@@ -45,6 +45,7 @@
 from typing import Final
 
 from aelfrice.derivation import DerivationInput, RouteOverrides, derive
+from aelfrice.doc_linker import ANCHOR_INGEST, file_uri_from_path
 from aelfrice.models import (
     CORROBORATION_SOURCE_CLI_REMEMBER,
     CORROBORATION_SOURCE_COMMIT_INGEST,
@@ -299,6 +300,19 @@ def _process_row(
         store.insert_edge(edge)
         derived_edge_ids.append((edge.src, edge.dst, edge.type))
 
+    # #435 doc-linker. When source_path is materialised on the ingest
+    # row, write a belief↔document anchor. Idempotent on (belief_id,
+    # doc_uri) at the storage layer, so re-derive of the same row is a
+    # no-op. Skip when source_path is None (transcript-ingest, lock /
+    # remember without --doc, etc.) per spec § "Linker invocation point".
+    if inp.source_path:
+        store.link_belief_to_document(
+            belief_id=actual_id,
+            doc_uri=file_uri_from_path(inp.source_path),
+            anchor_type=ANCHOR_INGEST,
+            position_hint=None,
+        )
+
     store.update_ingest_derived_ids(
         log_id,
         derived_belief_ids=[actual_id],
diff --git a/src/aelfrice/doc_linker.py b/src/aelfrice/doc_linker.py
new file mode 100644
index 000000000..93f355758
--- /dev/null
+++ b/src/aelfrice/doc_linker.py
@@ -0,0 +1,134 @@
+"""Document / semantic linker (#435).
+
+Connects a belief to the document anchor it describes — a file path with a
+line range, a URL with a section fragment, etc. — so retrieval can return the
+canonical reference alongside the bare belief snippet. Spec:
+``docs/feature-doc-linker.md``.
+
+The linker stores opaque URI strings in a sibling table ``belief_documents``
+keyed on ``(belief_id, doc_uri)``. Idempotent re-ingest: ``INSERT OR IGNORE``
+collapses repeats to the first row. Ingest-time writes happen inside the
+derivation worker when ``DerivationInput.source_path`` is set; manual writes
+happen via ``aelf lock --doc=URI``. Retrieval consumers opt in via the
+``with_doc_anchors=True`` kwarg on ``retrieve()`` / ``retrieve_v2()``.
+
+Out of scope at v2.0.0 (per spec):
+
+- ``anchor_type='derived'`` writers (retrieval-time inference). Enum value is
+  reserved; no writer in this module.
+- URI validation beyond non-empty.
+- ``source_path`` normalisation. The ingest writer stores whatever path the
+  caller passes; consumers normalise on the way out.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Final
+
+if TYPE_CHECKING:
+    from aelfrice.store import MemoryStore
+
+ANCHOR_INGEST: Final[str] = "ingest"
+ANCHOR_MANUAL: Final[str] = "manual"
+ANCHOR_DERIVED: Final[str] = "derived"
+
+ANCHOR_TYPES: Final[frozenset[str]] = frozenset(
+    {ANCHOR_INGEST, ANCHOR_MANUAL, ANCHOR_DERIVED},
+)
+
+
+@dataclass(frozen=True)
+class DocAnchor:
+    """One stored anchor row from ``belief_documents``.
+
+    ``doc_uri`` is opaque to the linker. Recommended encodings at v2.0.0:
+
+    - ``file:///abs/path/to/source.py#Lstart-Lend`` — local-source ingest.
+    - ``https://host/path#fragment`` — external doc / web ingest.
+
+    Other forms are accepted; the linker only rejects empty input.
+
+    ``position_hint`` is a free-form string (e.g. ``"L42-L60"``,
+    ``"#section"``); ``None`` when the URI itself encodes the position or
+    no hint is available.
+
+    ``created_at`` is a unix timestamp (seconds since epoch).
+    """
+
+    belief_id: str
+    doc_uri: str
+    anchor_type: str
+    position_hint: str | None
+    created_at: float
+
+
+def file_uri_from_path(
+    source_path: str,
+    *,
+    project_root: Path | None = None,
+    position_hint: str | None = None,
+) -> str:
+    """Build a ``file://`` URI from a local source path.
+
+    When ``project_root`` is set and ``source_path`` is inside it, the URI
+    encodes the path relative to the project root (avoids leaking the
+    operator's filesystem layout into the store). Otherwise the absolute
+    path is used.
+
+    ``position_hint`` is appended as a fragment when provided; the linker
+    stores the same value separately so consumers can read it without
+    parsing the URI.
+    """
+    p = Path(source_path)
+    rel: str
+    if project_root is not None:
+        try:
+            rel = str(p.resolve().relative_to(project_root.resolve()))
+        except ValueError:
+            rel = str(p)
+    else:
+        rel = str(p)
+    uri = f"file:{rel}" if not rel.startswith("/") else f"file://{rel}"
+    if position_hint:
+        uri = f"{uri}#{position_hint}"
+    return uri
+
+
+def link_belief_to_document(
+    store: "MemoryStore",
+    belief_id: str,
+    doc_uri: str,
+    *,
+    anchor_type: str = ANCHOR_INGEST,
+    position_hint: str | None = None,
+) -> DocAnchor:
+    """Persist a ``belief_id ↔ doc_uri`` anchor and return the row.
+
+    Idempotent on ``(belief_id, doc_uri)`` via ``INSERT OR IGNORE`` in the
+    storage layer: re-calling with the same pair is a no-op write but always
+    returns a ``DocAnchor`` reflecting the canonical (first-write) row.
+
+    Raises ``ValueError`` on empty ``doc_uri`` or unknown ``anchor_type``.
+    """
+    if not doc_uri:
+        raise ValueError("doc_uri must be non-empty")
+    if anchor_type not in ANCHOR_TYPES:
+        raise ValueError(
+            f"Unknown anchor_type {anchor_type!r}. "
+            f"Must be one of {sorted(ANCHOR_TYPES)}"
+        )
+    return store.link_belief_to_document(
+        belief_id=belief_id,
+        doc_uri=doc_uri,
+        anchor_type=anchor_type,
+        position_hint=position_hint,
+    )
+
+
+def get_doc_anchors(
+    store: "MemoryStore",
+    belief_id: str,
+) -> list[DocAnchor]:
+    """Return every anchor for a belief, ordered by ``created_at`` ASC."""
+    return store.get_doc_anchors(belief_id)
diff --git a/src/aelfrice/retrieval.py b/src/aelfrice/retrieval.py
index f96cdddd1..10ea9b4d0 100644
--- a/src/aelfrice/retrieval.py
+++ b/src/aelfrice/retrieval.py
@@ -74,6 +74,7 @@
     pack_with_clusters,
 )
 from aelfrice.compression import CompressedBelief, compress_for_retrieval
+from aelfrice.doc_linker import DocAnchor
 from aelfrice.vocab_bridge import VocabBridge, VocabBridgeCache
 from aelfrice.entity_extractor import extract_entities
 from aelfrice.graph_spectral import (
@@ -240,6 +241,12 @@ class RetrievalResult:
     last call. The benchmark adapter consumes it for the L0/L1/L2.5
     counts surface; default `[]` for backwards-compat with adapters
     that only inspect `beliefs`.
+
+    `doc_anchors` (#435) is a parallel list to `beliefs`: same length,
+    same order. `doc_anchors[i]` lists every `belief_documents` row for
+    `beliefs[i]`. Empty when the caller did not opt in via
+    `with_doc_anchors=True`; also empty for beliefs that have no
+    anchors.
     """
 
     beliefs: list[Belief]
@@ -248,6 +255,7 @@ class RetrievalResult:
     entity_hits: list[str] = field(default_factory=lambda: [])
     locked_ids: list[str] = field(default_factory=lambda: [])
     l1_ids: list[str] = field(default_factory=lambda: [])
+    doc_anchors: list[list[DocAnchor]] = field(default_factory=lambda: [])
     # v2.1 #434 type-aware compression. Populated when
     # use_type_aware_compression resolves True. Same length and order as
     # `beliefs` (parallel field — consumers that want compressed render
@@ -1636,6 +1644,7 @@ def retrieve_v2(
     use_vocab_bridge: bool | None = None,
     vocab_bridge_cache: VocabBridgeCache | None = None,
     use_intentional_clustering: bool | None = None,
+    with_doc_anchors: bool = False,
 ) -> RetrievalResult:
     """Lab-compatible retrieval wrapper for academic-suite adapters.
 
@@ -1740,6 +1749,16 @@ def retrieve_v2(
             for b in beliefs
         ]
 
+    # #435 doc-linker post-rank, pre-pack projection. Default OFF keeps
+    # the adapter wire bytes-identical for callers that don't opt in.
+    # When ON, one batched `belief_id IN (...)` SELECT joins anchors
+    # onto the result. Anchors are metadata for the consumer; they do
+    # NOT count against the token budget pack.
+    doc_anchors_list: list[list[DocAnchor]] = []
+    if with_doc_anchors and beliefs:
+        anchors_by_id = store.get_doc_anchors_batch([b.id for b in beliefs])
+        doc_anchors_list = [anchors_by_id.get(b.id, []) for b in beliefs]
+
     return RetrievalResult(
         beliefs=beliefs,
         entity_hits=l25_ids_list,
@@ -1747,6 +1766,7 @@ def retrieve_v2(
         l1_ids=l1_ids_list,
         bfs_chains=bfs_chains,
         compressed_beliefs=compressed,
+        doc_anchors=doc_anchors_list,
     )
 
 
diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py
index 6196abb51..ca9c35177 100644
--- a/src/aelfrice/store.py
+++ b/src/aelfrice/store.py
@@ -366,6 +366,28 @@ def _check_insert_belief_authority() -> None:
     """,
     "CREATE INDEX IF NOT EXISTS idx_log_versions_log "
     "ON log_versions(log_id)",
+    # v2.0 #435 doc linker. One row per (belief, doc_uri). PK gives
+    # idempotency on re-ingest of the same belief from the same source.
+    # ON DELETE CASCADE removes rows when a belief is hard-deleted (#440)
+    # — anchors are a derived projection of belief origin, not an audit
+    # trail (belief_corroborations #190 is the audit-trail sibling).
+    # `created_at` is REAL (unix seconds) — the linker is hot enough that
+    # we want numeric ordering rather than ISO-string comparison.
+    """
+    CREATE TABLE IF NOT EXISTS belief_documents (
+        belief_id     TEXT NOT NULL,
+        doc_uri       TEXT NOT NULL,
+        anchor_type   TEXT NOT NULL CHECK (anchor_type IN ('ingest', 'manual', 'derived')),
+        position_hint TEXT,
+        created_at    REAL NOT NULL,
+        PRIMARY KEY (belief_id, doc_uri),
+        FOREIGN KEY (belief_id) REFERENCES beliefs(id) ON DELETE CASCADE
+    )
+    """,
+    "CREATE INDEX IF NOT EXISTS idx_belief_documents_belief_id "
+    "ON belief_documents(belief_id)",
+    "CREATE INDEX IF NOT EXISTS idx_belief_documents_doc_uri "
+    "ON belief_documents(doc_uri)",
 )
 
 # Marker key for the entity-index one-shot backfill. Empty value =
@@ -1913,6 +1935,128 @@ def list_corroborations(
             for r in cur.fetchall()
         ]
 
+    # --- #435 doc linker --------------------------------------------------
+    #
+    # `belief_documents` rows are 1:1 with (belief_id, doc_uri) pairs.
+    # `INSERT OR IGNORE` collapses re-ingest of the same belief from the
+    # same source to the first-write row; `get_doc_anchors` returns the
+    # canonical row regardless of how many times the writer was called.
+    # See `aelfrice.doc_linker` for the public DocAnchor dataclass and
+    # spec contract; this module owns the SQL only.
+
+    def link_belief_to_document(
+        self,
+        *,
+        belief_id: str,
+        doc_uri: str,
+        anchor_type: str,
+        position_hint: str | None,
+    ) -> "DocAnchor":  # noqa: F821 — forward ref to avoid a circular import
+        """Persist one anchor row and return a `DocAnchor`.
+
+        Idempotent on `(belief_id, doc_uri)`: the table's PK + INSERT OR
+        IGNORE turns repeats into no-op writes. Returns the canonical
+        (first-write) row regardless of whether this call inserted.
+        """
+        # Imported here to avoid a module-level cycle: doc_linker imports
+        # MemoryStore via TYPE_CHECKING-only.
+        from aelfrice.doc_linker import ANCHOR_TYPES, DocAnchor
+
+        if not doc_uri:
+            raise ValueError("doc_uri must be non-empty")
+        if anchor_type not in ANCHOR_TYPES:
+            raise ValueError(
+                f"Unknown anchor_type {anchor_type!r}. "
+                f"Must be one of {sorted(ANCHOR_TYPES)}"
+            )
+        ts = datetime.now(timezone.utc).timestamp()
+        self._conn.execute(
+            """
+            INSERT OR IGNORE INTO belief_documents
+                (belief_id, doc_uri, anchor_type, position_hint, created_at)
+            VALUES (?, ?, ?, ?, ?)
+            """,
+            (belief_id, doc_uri, anchor_type, position_hint, ts),
+        )
+        self._conn.commit()
+        cur = self._conn.execute(
+            """
+            SELECT belief_id, doc_uri, anchor_type, position_hint, created_at
+            FROM belief_documents
+            WHERE belief_id = ? AND doc_uri = ?
+            """,
+            (belief_id, doc_uri),
+        )
+        row = cur.fetchone()
+        return DocAnchor(
+            belief_id=str(row["belief_id"]),
+            doc_uri=str(row["doc_uri"]),
+            anchor_type=str(row["anchor_type"]),
+            position_hint=row["position_hint"],
+            created_at=float(row["created_at"]),
+        )
+
+    def get_doc_anchors(self, belief_id: str) -> list["DocAnchor"]:  # noqa: F821
+        """Return every anchor for one belief, ordered by `created_at` ASC."""
+        from aelfrice.doc_linker import DocAnchor
+
+        cur = self._conn.execute(
+            """
+            SELECT belief_id, doc_uri, anchor_type, position_hint, created_at
+            FROM belief_documents
+            WHERE belief_id = ?
+            ORDER BY created_at ASC
+            """,
+            (belief_id,),
+        )
+        return [
+            DocAnchor(
+                belief_id=str(r["belief_id"]),
+                doc_uri=str(r["doc_uri"]),
+                anchor_type=str(r["anchor_type"]),
+                position_hint=r["position_hint"],
+                created_at=float(r["created_at"]),
+            )
+            for r in cur.fetchall()
+        ]
+
+    def get_doc_anchors_batch(
+        self,
+        belief_ids: list[str],
+    ) -> dict[str, list["DocAnchor"]]:  # noqa: F821
+        """Batched fetch keyed by `belief_id`. Empty list for ids without anchors.
+
+        Used by `retrieve(..., with_doc_anchors=True)` so the projection
+        costs one indexed read per call rather than one per surfaced
+        belief. Result dict has an entry for every requested id.
+        """
+        from aelfrice.doc_linker import DocAnchor
+
+        out: dict[str, list[DocAnchor]] = {bid: [] for bid in belief_ids}
+        if not belief_ids:
+            return out
+        ph = ",".join("?" * len(belief_ids))
+        cur = self._conn.execute(
+            f"""
+            SELECT belief_id, doc_uri, anchor_type, position_hint, created_at
+            FROM belief_documents
+            WHERE belief_id IN ({ph})
+            ORDER BY belief_id ASC, created_at ASC
+            """,
+            tuple(belief_ids),
+        )
+        for r in cur.fetchall():
+            out[str(r["belief_id"])].append(
+                DocAnchor(
+                    belief_id=str(r["belief_id"]),
+                    doc_uri=str(r["doc_uri"]),
+                    anchor_type=str(r["anchor_type"]),
+                    position_hint=r["position_hint"],
+                    created_at=float(r["created_at"]),
+                )
+            )
+        return out
+
     # --- Ingest log (v2.0, #205) -----------------------------------------
 
     def record_ingest(
diff --git a/tests/bench_gate/test_doc_linker.py b/tests/bench_gate/test_doc_linker.py
new file mode 100644
index 000000000..2051a04ca
--- /dev/null
+++ b/tests/bench_gate/test_doc_linker.py
@@ -0,0 +1,55 @@
+"""Bench gate for #435 doc linker.
+
+Spec acceptance A2 (docs/feature-doc-linker.md):
+
+    NDCG@k(with_doc_anchors=ON, anchors_populated=ON)
+        > NDCG@k(with_doc_anchors=ON, anchors_populated=OFF)
+
+The anchors-populated case loads each row's seed beliefs, writes the
+labelled `expected_doc_uris` against them, then queries with
+`with_doc_anchors=True` and lets a downstream rerank read the anchors.
+The anchors-EMPTY baseline runs the same query against the same beliefs
+without writing the anchors. Strictly positive uplift is the ship
+trigger; zero or negative uplift fails the gate.
+
+Public CI skips when ``AELFRICE_CORPUS_ROOT`` is unset, per the
+directory-of-origin rule (labelled corpus lives only in
+``~/projects/aelfrice-lab/tests/corpus/v2_0/doc_linker/``).
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.conftest import load_corpus_module
+
+
+@pytest.mark.bench_gated
+def test_doc_linker_uplift(aelfrice_corpus_root: Path) -> None:
+    rows = load_corpus_module(aelfrice_corpus_root, "doc_linker")
+    assert rows, "doc_linker corpus produced zero rows"
+
+    # The actual uplift driver lives in tests.retrieve_uplift_runner; the
+    # doc-linker variant defers until the bench-gate scoring contract
+    # for `with_doc_anchors` is written. Until then, the harness skips
+    # so a corpus-mounted run still surfaces the wiring as PASS rather
+    # than masking a missing scorer.
+    runner_mod = pytest.importorskip(
+        "tests.retrieve_uplift_runner",
+        reason=(
+            "doc-linker uplift runner not yet wired (operator gate; "
+            "spec § A2 — pending lab-side corpus + scorer)"
+        ),
+    )
+
+    results = runner_mod.run_doc_linker_uplift(rows)
+    detail = (
+        f"  NDCG_anchors_off={results.mean_ndcg_off:.4f} "
+        f"NDCG_anchors_on={results.mean_ndcg_on:.4f} "
+        f"uplift={results.uplift:+.4f}"
+    )
+    assert results.uplift > 0, (
+        f"doc-linker uplift not strictly positive on "
+        f"{len(rows)} rows:\n{detail}"
+    )
diff --git a/tests/corpus/v2_0/README.md b/tests/corpus/v2_0/README.md
index 2d12e0ca5..868902b6b 100644
--- a/tests/corpus/v2_0/README.md
+++ b/tests/corpus/v2_0/README.md
@@ -51,7 +51,9 @@ tests/corpus/v2_0/
 │   └── *.jsonl
 ├── wonder_online/                     #389 (Track B: aelf wonder)
 │   └── *.jsonl
-└── multi_fact/                        #436 (intentional clustering — multi-fact recall)
+├── multi_fact/                        #436 (intentional clustering — multi-fact recall)
+│   └── *.jsonl
+└── doc_linker/                        #435 (belief↔document anchor uplift)
     └── *.jsonl
 ```
 
@@ -93,6 +95,7 @@ required for **all** modules:
 | `reasoning` | `query` (string), `beliefs` (list[obj]), `edges` (list[obj]), `expected_hit_ids` (list[string]), `baseline_search_only_top_k` (list[string]), `k` (int) | `graded` |
 | `wonder_online` | `beliefs` (list[obj]), `edges` (list[obj]), `seed_id` (string), `expected_candidate_ids` (list[string]) | `graded` |
 | `multi_fact` | `query` (string), `expected_belief_ids` (list[string]), `expected_clusters` (list[list[string]]), `n_clusters_required` (int), `tag` (string) | `graded` |
+| `doc_linker` | `query` (string), `beliefs` (list[obj]), `expected_belief_ids` (list[string]), `expected_doc_uris` (list[string]), `k` (int) | `graded` |
 
 ### `directive_detection` re-entry gate (#374)
 
@@ -268,6 +271,32 @@ Per-row shape:
 The bench-gate test at `tests/bench_gate/test_intentional_clustering.py`
 skips cleanly when the module dir is empty.
 
+### `doc_linker` ship gate (#435)
+
+Per `docs/feature-doc-linker.md` § A2 the doc linker ships when the
+labelled corpus shows a **strictly positive NDCG@k uplift** on
+`with_doc_anchors=True` between the anchors-populated case and the
+anchors-empty case (same query, same beliefs, anchors written vs not
+written). Public CI cannot run this — labelled `expected_doc_uris`
+content lives lab-side per the directory-of-origin rule.
+
+Per-row shape:
+
+- `query` — string. The retrieve_v2 input under test.
+- `beliefs` — list of `{"id": str, "text": str}` (each id stable within
+  the row; harness wires them into a transient `MemoryStore`).
+- `expected_belief_ids` — non-empty list of belief ids that should
+  appear in the top-k under both runs (parity floor).
+- `expected_doc_uris` — non-empty list of doc URIs to write against the
+  beliefs in the anchors-populated run only. Forms: `file:<rel>` or
+  `https://...` per spec § "Doc URI scheme".
+- `k` — integer ≥ 1; rank cutoff used to compute NDCG@k.
+
+Public-tree fixtures may live here (synthetic-only); real-traffic
+fixtures stay lab-side per directory-of-origin rules. The bench-gate
+test at `tests/bench_gate/test_doc_linker.py` skips cleanly when the
+module dir is empty.
+
 ## v0.1 acceptance (per #307)
 
 - ≥ 50 non-seed entries per module file (300 total).
diff --git a/tests/corpus/v2_0/doc_linker/.gitkeep b/tests/corpus/v2_0/doc_linker/.gitkeep
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_cli_lock_doc_anchor.py b/tests/test_cli_lock_doc_anchor.py
new file mode 100644
index 000000000..340c106e2
--- /dev/null
+++ b/tests/test_cli_lock_doc_anchor.py
@@ -0,0 +1,91 @@
+"""#435 — `aelf lock --doc=URI` writes a manual anchor on the locked belief.
+
+The lock entry point passes `source_path=None` to the derivation worker
+(cli_remember has no canonical doc URI), so the worker's ingest-time
+hook does NOT fire. The CLI handler is the only path that writes anchors
+on the lock surface. Idempotent on re-lock with the same URI.
+"""
+from __future__ import annotations
+
+import argparse
+import io
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+
+from aelfrice.cli import _cmd_lock
+from aelfrice.doc_linker import ANCHOR_MANUAL, get_doc_anchors
+from aelfrice.store import MemoryStore
+
+
+@pytest.fixture
+def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[MemoryStore]:
+    db = tmp_path / "lock-doc.db"
+    monkeypatch.setenv("AELFRICE_DB", str(db))
+    s = MemoryStore(str(db))
+    yield s
+    s.close()
+
+
+def _ns(
+    statement: str,
+    *,
+    session_id: str | None = None,
+    doc_uri: str | None = None,
+) -> argparse.Namespace:
+    return argparse.Namespace(
+        statement=statement, session_id=session_id, doc_uri=doc_uri,
+    )
+
+
+def _locked_belief_id(store: MemoryStore) -> str:
+    rows = store._conn.execute(  # pyright: ignore[reportPrivateUsage]
+        "SELECT id FROM beliefs"
+    ).fetchall()
+    assert len(rows) == 1
+    return str(rows[0]["id"])
+
+
+def test_lock_without_doc_writes_no_anchor(store: MemoryStore) -> None:
+    """Hypothesis: aelf lock without --doc produces no belief_documents
+    rows. Falsifiable by any anchor on the locked belief."""
+    rc = _cmd_lock(_ns("atomic commits beat batched"), io.StringIO())
+    assert rc == 0
+
+    bid = _locked_belief_id(store)
+    assert get_doc_anchors(store, bid) == []
+
+
+def test_lock_with_doc_writes_manual_anchor(store: MemoryStore) -> None:
+    """Hypothesis: aelf lock --doc=URI produces exactly one manual
+    anchor with the supplied URI. Falsifiable by missing anchor, wrong
+    type, or wrong URI."""
+    rc = _cmd_lock(
+        _ns(
+            "the sky is blue",
+            doc_uri="https://example.com/notes#sky",
+        ),
+        io.StringIO(),
+    )
+    assert rc == 0
+
+    bid = _locked_belief_id(store)
+    anchors = get_doc_anchors(store, bid)
+    assert len(anchors) == 1
+    a = anchors[0]
+    assert a.doc_uri == "https://example.com/notes#sky"
+    assert a.anchor_type == ANCHOR_MANUAL
+
+
+def test_relock_with_same_doc_is_idempotent(store: MemoryStore) -> None:
+    """Re-locking the same statement with the same --doc adds no extra
+    anchor rows. The lock body itself is re-applied (lock-upgrade); the
+    anchor write is a no-op via INSERT OR IGNORE."""
+    args = _ns("atomic commits beat batched", doc_uri="file:CLAUDE.md#commits")
+    _cmd_lock(args, io.StringIO())
+    _cmd_lock(args, io.StringIO())
+
+    bid = _locked_belief_id(store)
+    anchors = get_doc_anchors(store, bid)
+    assert len(anchors) == 1
diff --git a/tests/test_corpus_schema.py b/tests/test_corpus_schema.py
index b88e33661..1439ba4e3 100644
--- a/tests/test_corpus_schema.py
+++ b/tests/test_corpus_schema.py
@@ -185,6 +185,20 @@
             "tag": "str",
         },
     ),
+    # #435 doc linker. NDCG@k uplift on a labelled query/belief/anchor
+    # fixture. `expected_belief_ids` floors the parity check (top-k must
+    # contain them in both runs); `expected_doc_uris` are the anchors
+    # written against those beliefs in the populated arm only.
+    "doc_linker": (
+        {"graded"},
+        {
+            "query": "str",
+            "beliefs": "list[belief]",
+            "expected_belief_ids": "list[str]",
+            "expected_doc_uris": "list[str]",
+            "k": "int",
+        },
+    ),
 }
 
 COMMON_REQUIRED = ("id", "provenance", "labeller_note", "label")
diff --git a/tests/test_doc_linker.py b/tests/test_doc_linker.py
new file mode 100644
index 000000000..ec2bb04c7
--- /dev/null
+++ b/tests/test_doc_linker.py
@@ -0,0 +1,325 @@
+"""Tests for the doc linker (#435) — storage, idempotency, schema migration."""
+from __future__ import annotations
+
+import sqlite3
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from aelfrice.doc_linker import (
+    ANCHOR_DERIVED,
+    ANCHOR_INGEST,
+    ANCHOR_MANUAL,
+    DocAnchor,
+    file_uri_from_path,
+    get_doc_anchors,
+    link_belief_to_document,
+)
+from aelfrice.models import (
+    BELIEF_FACTUAL,
+    LOCK_NONE,
+    LOCK_USER,
+    ORIGIN_AGENT_INFERRED,
+    ORIGIN_USER_STATED,
+    RETENTION_FACT,
+    Belief,
+)
+from aelfrice.store import MemoryStore
+
+
+def _ts() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _belief(bid: str, content: str, *, lock: str = LOCK_NONE) -> Belief:
+    ts = _ts()
+    return Belief(
+        id=bid,
+        content=content,
+        content_hash=f"hash-{bid}",
+        alpha=1.0,
+        beta=1.0,
+        type=BELIEF_FACTUAL,
+        lock_level=lock,
+        locked_at=ts if lock == LOCK_USER else None,
+        demotion_pressure=0,
+        created_at=ts,
+        last_retrieved_at=None,
+        session_id=None,
+        origin=ORIGIN_USER_STATED if lock == LOCK_USER else ORIGIN_AGENT_INFERRED,
+        retention_class=RETENTION_FACT,
+    )
+
+
+def test_link_belief_to_document_round_trip(tmp_path: Path) -> None:
+    store = MemoryStore(str(tmp_path / "rt.db"))
+    try:
+        store.insert_belief(_belief("b1", "hello world"))
+
+        anchor = link_belief_to_document(
+            store,
+            "b1",
+            "file:src/foo.py#L10-L20",
+            anchor_type=ANCHOR_INGEST,
+            position_hint="L10-L20",
+        )
+
+        assert isinstance(anchor, DocAnchor)
+        assert anchor.belief_id == "b1"
+        assert anchor.doc_uri == "file:src/foo.py#L10-L20"
+        assert anchor.anchor_type == ANCHOR_INGEST
+        assert anchor.position_hint == "L10-L20"
+        assert anchor.created_at > 0
+
+        out = get_doc_anchors(store, "b1")
+        assert out == [anchor]
+    finally:
+        store.close()
+
+
+def test_link_belief_to_document_idempotent(tmp_path: Path) -> None:
+    """A3: idempotency — N writes of the same (belief_id, doc_uri) → one row."""
+    store = MemoryStore(str(tmp_path / "idem.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        first = link_belief_to_document(
+            store, "b1", "file:src/foo.py", position_hint="L1"
+        )
+        for _ in range(5):
+            again = link_belief_to_document(
+                store,
+                "b1",
+                "file:src/foo.py",
+                anchor_type=ANCHOR_MANUAL,  # different! still no-op write
+                position_hint="L99",
+            )
+            # Returned row reflects the canonical (first-write) state —
+            # anchor_type / position_hint are NOT overwritten on no-op.
+            assert again.created_at == first.created_at
+            assert again.anchor_type == ANCHOR_INGEST
+            assert again.position_hint == "L1"
+
+        anchors = get_doc_anchors(store, "b1")
+        assert len(anchors) == 1
+        assert anchors[0].created_at == first.created_at
+    finally:
+        store.close()
+
+
+def test_link_belief_to_document_multiple_uris(tmp_path: Path) -> None:
+    """A belief can have many anchors. Ordering is created_at ASC."""
+    store = MemoryStore(str(tmp_path / "multi.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        first = link_belief_to_document(
+            store, "b1", "file:src/foo.py"
+        )
+        second = link_belief_to_document(
+            store, "b1", "https://example.com/foo#section"
+        )
+        third = link_belief_to_document(
+            store, "b1", "file:docs/bar.md", anchor_type=ANCHOR_MANUAL
+        )
+
+        anchors = get_doc_anchors(store, "b1")
+        assert [a.doc_uri for a in anchors] == [
+            first.doc_uri,
+            second.doc_uri,
+            third.doc_uri,
+        ]
+    finally:
+        store.close()
+
+
+def test_link_belief_to_document_rejects_empty_uri(tmp_path: Path) -> None:
+    store = MemoryStore(str(tmp_path / "empty.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        with pytest.raises(ValueError, match="non-empty"):
+            link_belief_to_document(store, "b1", "")
+    finally:
+        store.close()
+
+
+def test_link_belief_to_document_rejects_unknown_anchor_type(
+    tmp_path: Path,
+) -> None:
+    store = MemoryStore(str(tmp_path / "type.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        with pytest.raises(ValueError, match="anchor_type"):
+            link_belief_to_document(
+                store, "b1", "file:foo", anchor_type="auto",
+            )
+    finally:
+        store.close()
+
+
+def test_link_belief_to_document_fk_required(tmp_path: Path) -> None:
+    """Linking to a non-existent belief fails at the FK layer."""
+    store = MemoryStore(str(tmp_path / "fk.db"))
+    try:
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            link_belief_to_document(store, "b-missing", "file:foo.py")
+    finally:
+        store.close()
+
+
+def test_get_doc_anchors_empty_for_unknown_belief(tmp_path: Path) -> None:
+    store = MemoryStore(str(tmp_path / "empty.db"))
+    try:
+        assert get_doc_anchors(store, "nope") == []
+    finally:
+        store.close()
+
+
+def test_get_doc_anchors_batch(tmp_path: Path) -> None:
+    """Batched fetch returns an entry per requested id, even with no anchors."""
+    store = MemoryStore(str(tmp_path / "batch.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        store.insert_belief(_belief("b2", "y"))
+        store.insert_belief(_belief("b3", "z"))
+        link_belief_to_document(store, "b1", "file:a.py")
+        link_belief_to_document(store, "b1", "file:b.py")
+        link_belief_to_document(store, "b3", "file:c.py")
+
+        out = store.get_doc_anchors_batch(["b1", "b2", "b3"])
+        assert sorted(out.keys()) == ["b1", "b2", "b3"]
+        assert [a.doc_uri for a in out["b1"]] == ["file:a.py", "file:b.py"]
+        assert out["b2"] == []
+        assert [a.doc_uri for a in out["b3"]] == ["file:c.py"]
+
+        # Empty input → empty dict, no SQL.
+        assert store.get_doc_anchors_batch([]) == {}
+    finally:
+        store.close()
+
+
+def test_anchor_cascades_on_belief_delete(tmp_path: Path) -> None:
+    """ON DELETE CASCADE: deleting a belief drops its anchors."""
+    store = MemoryStore(str(tmp_path / "cascade.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        link_belief_to_document(store, "b1", "file:a.py")
+        link_belief_to_document(store, "b1", "file:b.py")
+        assert len(get_doc_anchors(store, "b1")) == 2
+
+        # Direct DELETE bypasses the audit-row insertion path of `aelf
+        # delete`; we just want the FK cascade behaviour confirmed.
+        store._conn.execute("DELETE FROM beliefs WHERE id = ?", ("b1",))
+        store._conn.commit()
+        assert get_doc_anchors(store, "b1") == []
+    finally:
+        store.close()
+
+
+def test_schema_migration_creates_table_on_existing_store(
+    tmp_path: Path,
+) -> None:
+    """A4: schema migration — first open of a store with no
+    `belief_documents` table creates it and populates anchors normally.
+
+    Simulates a v1.7-era store by opening with sqlite3 directly, creating
+    only the legacy schema, then re-opening through MemoryStore (which
+    runs the additive `CREATE TABLE IF NOT EXISTS belief_documents` on
+    every open).
+    """
+    db = tmp_path / "legacy.db"
+    legacy_conn = sqlite3.connect(str(db))
+    legacy_conn.execute(
+        """
+        CREATE TABLE beliefs (
+            id TEXT PRIMARY KEY,
+            content TEXT NOT NULL,
+            content_hash TEXT NOT NULL,
+            alpha REAL NOT NULL,
+            beta REAL NOT NULL,
+            type TEXT NOT NULL,
+            lock_level TEXT NOT NULL,
+            locked_at TEXT,
+            demotion_pressure INTEGER NOT NULL DEFAULT 0,
+            created_at TEXT NOT NULL,
+            last_retrieved_at TEXT
+        )
+        """
+    )
+    legacy_conn.commit()
+    legacy_conn.close()
+
+    store = MemoryStore(str(db))
+    try:
+        cur = store._conn.execute(
+            "SELECT name FROM sqlite_master "
+            "WHERE type='table' AND name='belief_documents'"
+        )
+        assert cur.fetchone() is not None, (
+            "belief_documents table must be created on first open of legacy store"
+        )
+
+        store.insert_belief(_belief("b1", "x"))
+        anchor = link_belief_to_document(store, "b1", "file:foo.py")
+        assert get_doc_anchors(store, "b1") == [anchor]
+    finally:
+        store.close()
+
+
+def test_anchor_type_enum_includes_derived_reservation(
+    tmp_path: Path,
+) -> None:
+    """`anchor_type='derived'` is reserved at v2.0.0 — accepted by the writer.
+
+    The spec defers the *writer* (retrieval-time inference) to v2.x but
+    keeps the enum value live so a future revision lands without a
+    schema migration. Verify the reservation by writing one row.
+    """
+    store = MemoryStore(str(tmp_path / "derived.db"))
+    try:
+        store.insert_belief(_belief("b1", "x"))
+        anchor = link_belief_to_document(
+            store,
+            "b1",
+            "file:src/foo.py",
+            anchor_type=ANCHOR_DERIVED,
+            position_hint="inferred",
+        )
+        assert anchor.anchor_type == ANCHOR_DERIVED
+    finally:
+        store.close()
+
+
+# --- file_uri_from_path -------------------------------------------------
+
+
+def test_file_uri_from_path_relative(tmp_path: Path) -> None:
+    """When project_root contains source_path, URI is repo-relative."""
+    src = tmp_path / "src" / "foo.py"
+    src.parent.mkdir(parents=True)
+    src.write_text("x")
+    uri = file_uri_from_path(
+        str(src),
+        project_root=tmp_path,
+        position_hint="L1-L5",
+    )
+    assert uri == "file:src/foo.py#L1-L5"
+
+
+def test_file_uri_from_path_absolute_when_outside_project(
+    tmp_path: Path,
+) -> None:
+    """source_path outside project_root falls back to the absolute path."""
+    elsewhere = tmp_path / "elsewhere.py"
+    elsewhere.write_text("x")
+    proj = tmp_path / "proj"
+    proj.mkdir()
+    uri = file_uri_from_path(str(elsewhere), project_root=proj)
+    # Falls through to the absolute path; "/abs/path" form gets file://
+    assert uri.startswith("file://"), uri
+    assert str(elsewhere) in uri
+
+
+def test_file_uri_from_path_no_position_hint() -> None:
+    uri = file_uri_from_path("docs/foo.md")
+    assert uri == "file:docs/foo.md"
+    assert "#" not in uri
diff --git a/tests/test_doc_linker_worker_hook.py b/tests/test_doc_linker_worker_hook.py
new file mode 100644
index 000000000..e78b20a4b
--- /dev/null
+++ b/tests/test_doc_linker_worker_hook.py
@@ -0,0 +1,154 @@
+"""#435 — derivation worker writes a doc anchor when source_path is set.
+
+The worker calls the linker AFTER `insert_or_corroborate`, so the anchor
+fires for both new beliefs and corroborations. Idempotent on
+`(belief_id, doc_uri)` so re-derive of the same row produces no extra
+anchors.
+
+Skips when `source_path` is None (transcripts, lock without --doc).
+"""
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+
+from aelfrice.derivation_worker import run_worker
+from aelfrice.doc_linker import ANCHOR_INGEST, get_doc_anchors
+from aelfrice.models import (
+    CORROBORATION_SOURCE_FILESYSTEM_INGEST,
+    CORROBORATION_SOURCE_TRANSCRIPT_INGEST,
+    INGEST_SOURCE_FILESYSTEM,
+)
+from aelfrice.store import MemoryStore
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> Iterator[MemoryStore]:
+    s = MemoryStore(str(tmp_path / "wh.db"))
+    yield s
+    s.close()
+
+
+def _record(
+    store: MemoryStore,
+    text: str,
+    *,
+    source_path: str | None,
+    call_site: str = CORROBORATION_SOURCE_FILESYSTEM_INGEST,
+) -> str:
+    return store.record_ingest(
+        source_kind=INGEST_SOURCE_FILESYSTEM,
+        source_path=source_path,
+        raw_text=text,
+        raw_meta={"call_site": call_site},
+    )
+
+
+def test_worker_writes_anchor_when_source_path_is_set(
+    store: MemoryStore,
+) -> None:
+    """Hypothesis: after `run_worker()`, every newly-derived belief whose
+    log row carried `source_path` has exactly one `belief_documents` row
+    pointing at that path. Falsifiable by a missing anchor or a wrong
+    URI."""
+    log_id = _record(
+        store,
+        "The system uses SQLite for storage.",
+        source_path="docs/architecture.md",
+    )
+
+    result = run_worker(store)
+    assert result.beliefs_inserted == 1
+
+    row = store.get_ingest_log_entry(log_id)
+    assert row is not None
+    bid = row["derived_belief_ids"][0]
+    anchors = get_doc_anchors(store, bid)
+    assert len(anchors) == 1
+    a = anchors[0]
+    assert a.anchor_type == ANCHOR_INGEST
+    assert a.doc_uri == "file:docs/architecture.md"
+    assert a.position_hint is None
+
+
+def test_worker_skips_anchor_when_source_path_is_none(
+    store: MemoryStore,
+) -> None:
+    """Transcript-ingest rows have `source_path=None`. No anchor written."""
+    log_id = _record(
+        store,
+        "I prefer atomic commits over batches.",
+        source_path=None,
+        call_site=CORROBORATION_SOURCE_TRANSCRIPT_INGEST,
+    )
+
+    run_worker(store)
+
+    row = store.get_ingest_log_entry(log_id)
+    assert row is not None
+    derived = row["derived_belief_ids"]
+    if not derived:
+        # Some classifier configurations skip transcript text; nothing
+        # to assert about anchors. Test still demonstrates the
+        # source_path=None path doesn't crash the worker.
+        return
+    bid = derived[0]
+    assert get_doc_anchors(store, bid) == []
+
+
+def test_worker_anchor_idempotent_on_re_derive(store: MemoryStore) -> None:
+    """Hypothesis: re-running the worker on the same row produces no
+    extra anchors. Falsifiable by anchor count > 1 after the second pass."""
+    _record(
+        store,
+        "The system uses SQLite for storage.",
+        source_path="docs/foo.md",
+    )
+
+    run_worker(store)
+    # Force a second pass by clearing the stamp on the row.
+    store._conn.execute(
+        "UPDATE ingest_log SET derived_belief_ids = NULL"
+    )
+    store._conn.commit()
+    run_worker(store)
+
+    cur = store._conn.execute("SELECT id FROM beliefs")
+    bids = [r["id"] for r in cur.fetchall()]
+    assert len(bids) == 1
+    anchors = get_doc_anchors(store, bids[0])
+    assert len(anchors) == 1
+
+
+def test_worker_anchor_persists_through_corroboration(
+    store: MemoryStore,
+) -> None:
+    """A second ingest of the same content (same content_hash) corroborates
+    rather than inserting. The first ingest's anchor remains; if the
+    second ingest carries a *different* doc_uri, both anchors stack.
+
+    Hypothesis: anchors are 1:1 with (belief_id, doc_uri), not (belief_id),
+    so two ingests of the same text from two different files produce two
+    anchors on one belief.
+    """
+    _record(
+        store,
+        "Same belief content for both ingests.",
+        source_path="docs/a.md",
+    )
+    run_worker(store)
+    _record(
+        store,
+        "Same belief content for both ingests.",
+        source_path="docs/b.md",
+    )
+    run_worker(store)
+
+    cur = store._conn.execute("SELECT id FROM beliefs")
+    bids = [r["id"] for r in cur.fetchall()]
+    assert len(bids) == 1, "expected one belief (corroborated)"
+    anchors = get_doc_anchors(store, bids[0])
+    uris = sorted(a.doc_uri for a in anchors)
+    assert uris == ["file:docs/a.md", "file:docs/b.md"]
diff --git a/tests/test_retrieve_doc_anchors.py b/tests/test_retrieve_doc_anchors.py
new file mode 100644
index 000000000..e800216e3
--- /dev/null
+++ b/tests/test_retrieve_doc_anchors.py
@@ -0,0 +1,170 @@
+"""#435 — retrieve_v2(with_doc_anchors=True) attaches anchor metadata.
+
+`with_doc_anchors=False` (the default) keeps `RetrievalResult.doc_anchors`
+empty so adapters that don't opt in see byte-identical wire shape.
+`with_doc_anchors=True` populates a parallel list aligned with
+`result.beliefs`.
+"""
+from __future__ import annotations
+
+from collections.abc import Iterator
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from aelfrice.derivation_worker import run_worker
+from aelfrice.doc_linker import (
+    ANCHOR_INGEST,
+    ANCHOR_MANUAL,
+    link_belief_to_document,
+)
+from aelfrice.models import (
+    BELIEF_FACTUAL,
+    INGEST_SOURCE_FILESYSTEM,
+    LOCK_NONE,
+    ORIGIN_AGENT_INFERRED,
+    RETENTION_FACT,
+    Belief,
+)
+from aelfrice.retrieval import retrieve_v2
+from aelfrice.store import MemoryStore
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> Iterator[MemoryStore]:
+    s = MemoryStore(str(tmp_path / "ra.db"))
+    yield s
+    s.close()
+
+
+def _seed_via_worker(store: MemoryStore, text: str, source_path: str) -> str:
+    """Drive a belief through the worker so anchors are written through
+    the production code path. Returns the resulting belief id."""
+    log_id = store.record_ingest(
+        source_kind=INGEST_SOURCE_FILESYSTEM,
+        source_path=source_path,
+        raw_text=text,
+        raw_meta={"call_site": "filesystem_ingest"},
+    )
+    run_worker(store)
+    row = store.get_ingest_log_entry(log_id)
+    assert row is not None and row["derived_belief_ids"]
+    return row["derived_belief_ids"][0]
+
+
+def _seed_direct(store: MemoryStore, bid: str, content: str) -> Belief:
+    ts = datetime.now(timezone.utc).isoformat()
+    b = Belief(
+        id=bid,
+        content=content,
+        content_hash=f"hash-{bid}",
+        alpha=1.0,
+        beta=1.0,
+        type=BELIEF_FACTUAL,
+        lock_level=LOCK_NONE,
+        locked_at=None,
+        demotion_pressure=0,
+        created_at=ts,
+        last_retrieved_at=None,
+        session_id=None,
+        origin=ORIGIN_AGENT_INFERRED,
+        retention_class=RETENTION_FACT,
+    )
+    store.insert_belief(b)
+    return b
+
+
+def test_default_off_keeps_doc_anchors_empty(store: MemoryStore) -> None:
+    """Hypothesis: default-off retrieve_v2 leaves doc_anchors=[] even when
+    anchors exist in the store. Adapter wire shape stays byte-identical.
+    Falsifiable by a non-empty `doc_anchors` field on default call."""
+    bid = _seed_via_worker(
+        store,
+        "The architecture document explains storage.",
+        "docs/architecture.md",
+    )
+    assert store.get_doc_anchors(bid)  # anchor exists in store
+
+    result = retrieve_v2(store, "architecture document storage")
+    assert result.doc_anchors == []
+
+
+def test_with_doc_anchors_attaches_parallel_list(store: MemoryStore) -> None:
+    """Hypothesis: with_doc_anchors=True returns a list of length
+    len(beliefs); each entry carries every anchor for the corresponding
+    belief; ordering is created_at ASC. Falsifiable by length mismatch
+    or by a missing anchor."""
+    bid = _seed_via_worker(
+        store,
+        "The architecture document explains storage.",
+        "docs/architecture.md",
+    )
+    # Stack a manual anchor on the same belief.
+    link_belief_to_document(
+        store,
+        bid,
+        "https://example.com/architecture",
+        anchor_type=ANCHOR_MANUAL,
+    )
+
+    result = retrieve_v2(
+        store,
+        "architecture document storage",
+        with_doc_anchors=True,
+    )
+    assert result.beliefs, "retrieve must return at least one belief"
+    assert len(result.doc_anchors) == len(result.beliefs)
+
+    # Find the anchored belief in the result and check its anchors.
+    for b, anchors in zip(result.beliefs, result.doc_anchors):
+        if b.id == bid:
+            uris = sorted(a.doc_uri for a in anchors)
+            assert uris == [
+                "file:docs/architecture.md",
+                "https://example.com/architecture",
+            ]
+            assert any(a.anchor_type == ANCHOR_INGEST for a in anchors)
+            assert any(a.anchor_type == ANCHOR_MANUAL for a in anchors)
+            break
+    else:
+        pytest.fail(f"belief {bid} not in retrieve result")
+
+
+def test_with_doc_anchors_handles_anchorless_beliefs(
+    store: MemoryStore,
+) -> None:
+    """Beliefs that have no anchors get an empty list at the same index
+    (not omitted). The parallel-list contract requires len(doc_anchors)
+    == len(beliefs)."""
+    bid_with_anchor = _seed_via_worker(
+        store,
+        "The system uses SQLite for storage.",
+        "docs/storage.md",
+    )
+    # Anchor-free belief, written directly (no ingest log entry).
+    _seed_direct(store, "naked", "Storage layer matters.")
+
+    result = retrieve_v2(store, "storage", with_doc_anchors=True)
+    assert len(result.doc_anchors) == len(result.beliefs)
+
+    for b, anchors in zip(result.beliefs, result.doc_anchors):
+        if b.id == bid_with_anchor:
+            assert anchors and anchors[0].doc_uri == "file:docs/storage.md"
+        elif b.id == "naked":
+            assert anchors == []
+
+
+def test_with_doc_anchors_off_does_not_query_store(
+    store: MemoryStore,
+) -> None:
+    """Default off path performs no batched anchor SELECT — proxy: an
+    empty store of beliefs still returns doc_anchors=[].
+
+    This protects the byte-identical-wire-shape contract for adapters
+    that don't opt in: turning the flag on must be the only path that
+    reads `belief_documents`.
+    """
+    result = retrieve_v2(store, "anything")
+    assert result.beliefs == []
+    assert result.doc_anchors == []