diff --git a/benchmark/robustness.py b/benchmark/robustness.py
new file mode 100644
index 0000000..a66b0a6
--- /dev/null
+++ b/benchmark/robustness.py
@@ -0,0 +1,421 @@
+"""Reproducible local fault experiments (fixtures are never quality results).
+
+    python -m benchmark.robustness query --fixture --output /tmp/query-faults
+    python -m benchmark.robustness build --fixture --after 2 --output /tmp/build-faults
+
+Live mode uses Settings/environment and requires --source and --questions for
+queries. JSONL tasks may contain question, task_id, filters, expected_sources,
+and reference_answer. Labels are used only by the external scoring code below.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import hashlib
+import json
+import math
+import os
+import platform
+import random
+import shutil
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+from datamind.config import Settings
+from datamind.workspace import Workspace
+
+REPO = Path(__file__).resolve().parents[1]
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()
+
+
+def write_json(path, value):
+    path.write_text(json.dumps(value, indent=2, ensure_ascii=False, default=str) + "\n")
+
+
+def append_row(path, value):
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(value, ensure_ascii=False, default=str) + "\n")
+        handle.flush()
+        os.fsync(handle.fileno())
+
+
+def provenance(args, source):
+    def git(*cmd):
+        return subprocess.run(["git", *cmd], cwd=REPO, capture_output=True).stdout
+    return {
+        "schema_version": 1, "command": args.command, "fixture": args.fixture,
+        "result_kind": "fixture_smoke" if args.fixture else "live_experiment",
+        "git_commit": git("rev-parse", "HEAD").decode().strip(),
+        "tracked_diff_hash": hashlib.sha256(git("diff", "HEAD")).hexdigest(),
+        "code_hash": digest([(str(p.relative_to(REPO)), hashlib.sha256(p.read_bytes()).hexdigest())
+                             for package in ("datamind", "benchmark")
+                             for p in sorted((REPO / package).rglob("*.py"))]),
+        "python": platform.python_version(), "seed": getattr(args, "seed", None),
+        "source_hash": digest([(str(p.relative_to(source)), hashlib.sha256(p.read_bytes()).hexdigest())
+                               for p in sorted(source.rglob("*")) if p.is_file()]),
+        "questions_hash": hashlib.sha256(Path(args.questions).read_bytes()).hexdigest() if args.questions else None,
+        # Only harness arguments, never Settings secrets/endpoints.
+        "harness": vars(args),
+    }
+
+
+def setup_fixture():
+    """Deterministic CPU-only providers with real local snapshot storage."""
+    from datamind.agent import options
+
+    class Embed:
+        name = "robustness-fixture"
+        dimension = 8
+
+        async def embed_texts(self, texts):
+            return [[float(i == int(hashlib.sha256(t.encode()).hexdigest()[:4], 16) % 8)
+                     for i in range(8)] for t in texts]
+
+        async def embed_query(self, text):
+            return (await self.embed_texts([text]))[0]
+
+    class Model:
+        async def generate_text(self, prompt, **kwargs):
+            return '[{"subject":"Fixture","relation":"owns","object":"Demo"}]'
+
+        async def complete(self, **kwargs):
+            raise RuntimeError("Fixture mode evaluates retrieval, not LLM answer quality")
+
+        async def aclose(self):
+            pass
+
+    options.build_embedding = lambda *a, **k: Embed()
+    options.build_model_client = lambda *a, **k: Model()
+
+
+def settings_for(runtime, fixture, args=None):
+    settings = (Settings(_env_file=None, llm={"api_key": "fixture", "model": "fixture"})
+                if fixture else Settings())
+    settings.data.base_dir = Path(runtime).resolve()
+    settings.retrieval.strategy = "hybrid"
+    settings.retrieval.vector_retries = 0  # Reader owns the one shared retry budget.
+    settings.retrieval.vector_failure_fallback = "error"
+    if fixture:
+        settings.llm.api_key = "fixture"
+        settings.llm.model = "fixture"
+        settings.llm.fallback_model = None
+        settings.embedding.dimension = 8
+        settings.embedding.model = "fixture"
+        settings.kb.vector_store = "chroma"
+        settings.kb.ingest_mode = "direct"
+        settings.graph.backend = "networkx"
+        settings.db.dialect = "sqlite"
+        settings.db.dsn = None
+        setup_fixture()
+    if args is not None and getattr(args, "deadline_s", None):
+        settings.agent.wall_clock_timeout_s = args.deadline_s
+    return settings
+
+
+def source_and_tasks(args, output):
+    if args.source:
+        source = Path(args.source).resolve()
+        if not source.is_dir():
+            raise ValueError("--source must be a directory")
+        if output == source or output.is_relative_to(source):
+            raise ValueError("--output must be outside the source workspace")
+    elif args.fixture:
+        source = output / "inputs"
+        source.mkdir()
+        for i, term in enumerate(("nebula", "orchid", "cobalt", "quartz", "maple", "harbor")):
+            (source / f"doc{i}.md").write_text(f"{term} project budget is {i + 10} units.\n")
+        (source / "orders.csv").write_text("item,amount\nA,12\nB,24\n")
+    else:
+        raise ValueError("Live experiments require --source")
+    if args.questions:
+        tasks = [json.loads(line) for line in Path(args.questions).read_text().splitlines() if line.strip()]
+        for i, task in enumerate(tasks):
+            if not isinstance(task.get("question"), str):
+                raise ValueError(f"task {i} needs a question")
+            task.setdefault("task_id", str(i))
+    elif args.fixture:
+        tasks = [{"task_id": "0", "question": "nebula", "expected_sources": ["doc0.md"]},
+                 {"task_id": "1", "question": "orchid", "expected_sources": ["doc1.md"]}]
+    else:
+        tasks = []
+    return source, tasks
+
+
+def score(task, result, execution):
+    """No LLM judge or heuristic pretending to be official benchmark accuracy."""
+    expected = task.get("expected_sources")
+    observed = [r.get("source") for r in result.get("results", [])] if execution == "retrieval" else [
+        e.get("locator", {}).get("source") or e.get("source_id") for e in result.get("evidence", [])]
+    source_hit = None if not expected else any(
+        s and (s in expected or Path(s).name in expected) for s in observed)
+    reference = task.get("reference_answer")
+    exact = None if execution != "agent" or reference is None else str(result.get("answer", "")).strip() == str(reference).strip()
+    return {"source_hit": source_hit, "exact_answer_match": exact}
+
+
+class Faults:
+    """Per-task schedules are paired across policies; retries consume later events."""
+    def __init__(self, seed, task, probability, outage_ms, delay_ms):
+        self.seed, self.task, self.p = seed, task, probability
+        self.started = time.monotonic()
+        self.outage = outage_ms / 1000
+        self.delay = delay_ms / 1000
+        self.attempts, self.injected = 0, 0
+
+    async def before(self):
+        attempt = self.attempts
+        self.attempts += 1
+        rng = random.Random(f"{self.seed}:{self.task}:{attempt}")
+        if time.monotonic() - self.started < self.outage or rng.random() < self.p:
+            self.injected += 1
+            await asyncio.sleep(self.delay)
+            raise TimeoutError("injected transient fault")
+
+
+def install_fault(reader, faults, component):
+    retriever = reader._system.services.kb.retriever
+    if component == "embedding":
+        original = retriever._embed.embed_query
+        async def failing(query):
+            await faults.before()
+            return await original(query)
+        retriever._embed.embed_query = failing
+    else:
+        original = retriever._store.query
+        async def failing(*a, **k):
+            await faults.before()
+            return await original(*a, **k)
+        retriever._store.query = failing
+
+
+async def query_experiment(args, output, source, tasks):
+    if not tasks:
+        raise ValueError("Query experiments require --questions (or --fixture)")
+    if args.fixture and args.execution != "retrieval":
+        raise ValueError("Fixture mode cannot evaluate LLM answers; use --execution retrieval")
+    settings = settings_for(output / "runtime", args.fixture, args)
+    workspace = Workspace(settings)
+    built = await workspace.update(str(source), surfaces=args.surfaces.split(","))
+    manifest = provenance(args, source)
+    manifest.update(workspace_revision=built["workspace_revision"], pipeline_hash=built["pipeline_hash"],
+                    retrieval=settings.retrieval.model_dump(),
+                    model=settings.llm.model, embedding_model=settings.embedding.model,
+                    scoring="source-hit / exact string match only; use official scorers for paper results")
+    write_json(output / "manifest.json", manifest)
+    write_json(output / "build.json", built)
+    rows = []
+    # Seeded task/policy ordering limits systematic cold/warm and order effects.
+    jobs = [(repeat, task, policy, p) for repeat in range(args.repeats) for task in tasks
+            for policy in ("none", "retry", "fallback") for p in args.failure_probabilities]
+    random.Random(args.seed).shuffle(jobs)
+    for repeat, task, policy, probability in jobs:
+        workspace.settings.retrieval.vector_failure_fallback = "bm25" if policy == "fallback" else "error"
+        reader_started = time.monotonic()
+        row = {"task_id": task["task_id"], "repeat": repeat, "policy": policy,
+               "failure_probability": probability, "component": args.component,
+               "workspace_revision": built["workspace_revision"], "fixture": args.fixture,
+               "source_hit": False if task.get("expected_sources") else None,
+               "exact_answer_match": False if task.get("reference_answer") is not None and args.execution == "agent" else None}
+        try:
+            async with workspace.reader(built["workspace_revision"], read_retries=0 if policy == "none" else args.retries) as reader:
+                row["reader_setup_ms"] = (time.monotonic() - reader_started) * 1000
+                faults = Faults(args.seed + repeat, str(task["task_id"]), probability,
+                                args.outage_ms, args.fault_delay_ms)
+                install_fault(reader, faults, args.component)
+                started = time.monotonic()
+                try:
+                    async with asyncio.timeout(args.deadline_s):
+                        result = await (reader.call("kb_search", query=task["question"], filters=task.get("filters"))
+                                        if args.execution == "retrieval" else reader.query(task["question"]))
+                    row.update(outcome="ok", result=result, **score(task, result, args.execution))
+                except Exception as exc:
+                    row.update(outcome="error", error_type=type(exc).__name__, recovery=getattr(exc, "datamind_recovery", {}), source_hit=False if task.get("expected_sources") else None,
+                               exact_answer_match=False if task.get("reference_answer") is not None and args.execution == "agent" else None)
+                row.update(latency_ms=(time.monotonic() - started) * 1000,
+                           injected_faults=faults.injected, component_attempts=faults.attempts)
+        except Exception as exc:
+            row.update(outcome="reader_error", error_type=type(exc).__name__)
+        rows.append(row)
+        append_row(output / "rows.jsonl", row)
+    groups = []
+    for probability in args.failure_probabilities:
+        for policy in ("none", "retry", "fallback"):
+            batch = [r for r in rows if r["policy"] == policy and r["failure_probability"] == probability]
+            latencies = sorted(r["latency_ms"] for r in batch if "latency_ms" in r)
+            groups.append({"policy": policy, "failure_probability": probability, "tasks": len(batch),
+                           "completion_rate": sum(r["outcome"] == "ok" for r in batch) / len(batch),
+                           "p95_latency_ms": latencies[math.ceil(.95 * len(latencies)) - 1] if latencies else None,
+                           **{key + "_rate": (sum(r.get(key) is True for r in batch) / sum(r.get(key) is not None for r in batch))
+                              if any(r.get(key) is not None for r in batch) else None for key in ("source_hit", "exact_answer_match")}})
+    write_json(output / "summary.json", {"fixture": args.fixture, "groups": groups})
+
+
+async def snapshot_content(workspace, revision):
+    """Logical output fingerprint, excluding physical index IDs and timings."""
+    manifest = workspace.manifest(revision)
+    contents = {"sources": [(s["uri"], s["version"]) for s in manifest["sources"]]}
+    async with workspace.reader(revision) as reader:
+        services = reader._system.services
+        if services.kb:
+            contents["kb"] = sorted(await services.kb.vector_store.get_all_texts(), key=lambda x: x[0])
+        if services.graph:
+            graph = json.loads(services.graph.store._path.read_text())
+            contents["graph"] = {k: sorted(v, key=lambda x: json.dumps(x, sort_keys=True)) for k, v in graph.items()}
+        if services.db:
+            # Local SQLite workspace API: read all rows, not a row-limit sample.
+            import sqlite3
+            with sqlite3.connect(services.db.engine.url.database) as conn:
+                contents["db"] = {}
+                for table, in conn.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"):
+                    quoted = '"' + table.replace('"', '""') + '"'
+                    contents["db"][table] = sorted(conn.execute(f"SELECT * FROM {quoted}").fetchall(), key=str)
+    return digest(contents)
+
+
+async def build_worker(args):
+    import datamind.workspace as module
+    settings = settings_for(args.runtime, args.fixture)
+    workspace = Workspace(settings)
+    original = module._atomic_json
+    completed = 0
+    def injected(path, value, **kwargs):
+        nonlocal completed
+        original(path, value, **kwargs)
+        if path.parent.name == "checkpoints":
+            completed += 1
+            if args.crash_stage == "checkpoint" and completed == args.after:
+                os._exit(23)
+        if args.crash_stage == "publication" and path.name == "manifest.json":
+            os._exit(23)  # Snapshot prepared, head has not moved.
+    module._atomic_json = injected
+    result = await workspace.update(args.source, surfaces=args.surfaces.split(","))
+    fingerprint = await snapshot_content(workspace, result["workspace_revision"])
+    probes = []
+    if args.questions:
+        tasks = [json.loads(line) for line in Path(args.questions).read_text().splitlines() if line.strip()]
+        async with workspace.reader(result["workspace_revision"]) as reader:
+            for i, task in enumerate(tasks):
+                row = {"task_id": task.get("task_id", str(i))}
+                try:
+                    execution = "retrieval" if args.fixture else "agent"
+                    response = (await reader.call("kb_search", query=task["question"], filters=task.get("filters"))
+                                if args.fixture else await reader.query(task["question"]))
+                    row.update(outcome="ok", result=response, **score(task, response, execution))
+                except Exception as exc:
+                    row.update(outcome="error", error_type=type(exc).__name__)
+                probes.append(row)
+    write_json(Path(args.worker_result), {"build": result, "content_fingerprint": fingerprint,
+                                         "probes": probes})
+
+
+def run_worker(args, runtime, result, crash_stage="none"):
+    command = [sys.executable, "-m", "benchmark.robustness", "_build_worker",
+               "--source", args.source, "--runtime", str(runtime), "--worker-result", str(result),
+               "--surfaces", args.surfaces, "--crash-stage", crash_stage, "--after", str(args.after)]
+    if args.fixture:
+        command.append("--fixture")
+    if args.questions:
+        command.extend(["--questions", args.questions])
+    with result.with_suffix(".log").open("w") as log:
+        return subprocess.run(command, cwd=REPO, env={**os.environ, "PYTHONPATH": str(REPO)},
+                              stdout=log, stderr=log, timeout=args.worker_timeout_s).returncode
+
+
+def build_experiment(args, output, source):
+    args.source = str(source)
+    settings = settings_for(output / "crashed", args.fixture)
+    count = len(Workspace(settings)._inventory(source))
+    if args.crash_stage == "checkpoint" and not 1 <= args.after <= count:
+        raise ValueError(f"--after must be between 1 and {count} supported sources")
+    write_json(output / "manifest.json", provenance(args, source))
+    crashed = output / "crashed"
+    code = run_worker(args, crashed, output / "interrupted.json", args.crash_stage)
+    if code != 23:
+        raise RuntimeError(f"expected injected process exit 23, got {code}; inspect interrupted.log")
+    assert Workspace(settings).current_revision is None, "partial build was published"
+    results = {}
+    for policy in ("cold", "resume"):
+        runtime = output / policy
+        shutil.copytree(crashed, runtime)
+        if policy == "cold":
+            shutil.rmtree(runtime / "storage" / settings.data.profile / "workspace" / "checkpoints", ignore_errors=True)
+        result_path = output / f"{policy}.json"
+        code = run_worker(args, runtime, result_path)
+        if code:
+            raise RuntimeError(f"{policy} recovery failed with {code}; inspect {policy}.log")
+        results[policy] = json.loads(result_path.read_text())
+    write_json(output / "summary.json", {
+        "fixture": args.fixture, "crash_stage": args.crash_stage, "after_sources": args.after,
+        "cold_metrics": results["cold"]["build"]["metrics"],
+        "resume_metrics": results["resume"]["build"]["metrics"],
+        "logical_content_equal": results["cold"]["content_fingerprint"] == results["resume"]["content_fingerprint"],
+        "cold_probes": results["cold"]["probes"], "resume_probes": results["resume"]["probes"],
+        "note": "Recovery-only costs. Live model extraction can be nondeterministic; assess quality with downstream tasks too.",
+    })
+
+
+def parser():
+    root = argparse.ArgumentParser(description=__doc__)
+    sub = root.add_subparsers(dest="command", required=True)
+    for command in ("query", "build", "_build_worker"):
+        p = sub.add_parser(command)
+        p.add_argument("--fixture", action="store_true", help="CPU-only smoke fixtures, never paper quality results")
+        p.add_argument("--source")
+        p.add_argument("--questions")
+        p.add_argument("--surfaces", default="kb,graph,db" if command != "query" else "kb")
+        if command == "_build_worker":
+            p.add_argument("--runtime", required=True)
+            p.add_argument("--worker-result", required=True)
+        else:
+            p.add_argument("--output", required=True, help="new directory; never overwrite an experiment")
+        if command in {"build", "_build_worker"}:
+            p.add_argument("--after", type=int, default=2)
+            p.add_argument("--crash-stage", choices=["none", "checkpoint", "publication"], default="checkpoint" if command == "build" else "none")
+        if command == "build":
+            p.add_argument("--worker-timeout-s", type=float, default=300.)
+        if command == "query":
+            p.add_argument("--seed", type=int, default=0)
+            p.add_argument("--execution", choices=["retrieval", "agent"], default="retrieval")
+            p.add_argument("--component", choices=["embedding", "vector"], default="embedding")
+            p.add_argument("--failure-probabilities", type=float, nargs="+", default=[0., .1, .5, 1.])
+            p.add_argument("--outage-ms", type=float, default=0.)
+            p.add_argument("--fault-delay-ms", type=float, default=0.)
+            p.add_argument("--retries", type=int, default=1)
+            p.add_argument("--repeats", type=int, default=3)
+            p.add_argument("--deadline-s", type=float, default=30.)
+    return root
+
+
+def main():
+    args = parser().parse_args()
+    if args.questions:
+        args.questions = str(Path(args.questions).resolve())
+    if args.command == "_build_worker":
+        asyncio.run(build_worker(args))
+        return
+    if args.command == "query":
+        if any(not 0 <= p <= 1 for p in args.failure_probabilities) or not 0 <= args.retries <= 5:
+            raise ValueError("probabilities must be in [0,1]; retries must be in [0,5]")
+        if args.repeats < 1 or args.deadline_s <= 0 or min(args.outage_ms, args.fault_delay_ms) < 0:
+            raise ValueError("invalid repeats/deadline/fault duration")
+        if "kb" not in args.surfaces.split(","):
+            raise ValueError("query fault experiments require the kb surface")
+    elif args.crash_stage == "none":
+        raise ValueError("build experiment requires checkpoint or publication fault")
+    output = Path(args.output).resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    source, tasks = source_and_tasks(args, output)
+    if args.command == "query":
+        asyncio.run(query_experiment(args, output, source, tasks))
+    else:
+        build_experiment(args, output, source)
+    print(output / "summary.json")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmark/run.py b/benchmark/run.py
index 561bcf3..aab7684 100644
--- a/benchmark/run.py
+++ b/benchmark/run.py
@@ -228,6 +228,7 @@ async def _run_one(
             "surfaces_used": (result or {}).get("surfaces_used", []),
             "tool_trace": (result or {}).get("tool_trace", []),
             "evidence": (result or {}).get("evidence", []),
+            "recovery": (result or {}).get("recovery", {}),
             "completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
         }
         for passthrough in ("reference_answer", "metadata"):
diff --git a/datamind/agent/options.py b/datamind/agent/options.py
index 6c92a1c..6a0490b 100644
--- a/datamind/agent/options.py
+++ b/datamind/agent/options.py
@@ -9,6 +9,8 @@
 """
 from __future__ import annotations
 
+import asyncio
+import time
 from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any
@@ -42,6 +44,7 @@
 from datamind.core.model_clients import build_model_client
 from datamind.core.protocols import EmbeddingProvider, TextModelClient, ToolCallingModelClient
 from datamind.core.tools import ToolRegistry
+from datamind.core.recovery import recovery_summary
 
 from .base import AgentLoopConfig, AgentLoopProtocol
 from .loop_native import NativeAgentLoop
@@ -137,9 +140,22 @@ async def query(
         history: list[dict] | None = None,
         final_contract: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        return await self.loop.run_turn(
-            user_message=message, history=history, final_contract=final_contract,
-        )
+        parent = current_context()
+        ctx = (replace(parent, extra=dict(parent.extra)) if parent else
+               RequestContext.new(profile=self.ledger._profile if self.ledger else "default"))
+        ctx.extra["recovery_events"] = []
+        deadline = time.monotonic() + getattr(getattr(self.loop, "_cfg", None), "wall_clock_timeout_s", 300.)
+        ctx.extra["query_deadline"] = min(deadline, ctx.extra.get("query_deadline", deadline))
+        with bind_context(ctx):
+            try:
+                async with asyncio.timeout(max(0., ctx.extra["query_deadline"] - time.monotonic())):
+                    result = await self.loop.run_turn(
+                        user_message=message, history=history, final_contract=final_contract,
+                    )
+            finally:
+                if parent is not None:
+                    parent.extra.setdefault("recovery_events", []).extend(ctx.extra["recovery_events"])
+        return {**result, "recovery": recovery_summary(ctx.extra["recovery_events"])}
 
 
 @dataclass
diff --git a/datamind/capabilities/embedding/providers/openai_compatible.py b/datamind/capabilities/embedding/providers/openai_compatible.py
index 42a91a0..ce0c89e 100644
--- a/datamind/capabilities/embedding/providers/openai_compatible.py
+++ b/datamind/capabilities/embedding/providers/openai_compatible.py
@@ -23,6 +23,7 @@
 from datamind.core.errors import ExternalServiceError
 from datamind.core.logging import get_logger
 from datamind.core.registry import embedding_registry
+from datamind.core.recovery import read_retry_budget_active
 
 _log = get_logger("embedding.openai_compatible")
 
@@ -212,6 +213,17 @@ async def _call(self, inputs: list[str]) -> list[list[float]]:
                         "embedding_dimension_detected",
                         extra={"model": self._model, "dim": self.dimension},
                     )
+                # Provider-reported usage only. Never infer tokens from text
+                # length or label an unavailable usage report as zero cost.
+                from datamind.core.logging import current_context
+                ctx = current_context()
+                usage = body.get("usage")
+                if ctx is not None and isinstance(usage, dict):
+                    tokens = usage.get("total_tokens")
+                    if isinstance(tokens, int) and not isinstance(tokens, bool) and tokens >= 0:
+                        ctx.extra.setdefault("embedding_usage", []).append({
+                            "model": self._model, "total_tokens": tokens,
+                        })
                 return vecs
             except httpx.HTTPStatusError as exc:
                 # Ordinary 4xx (especially 401/403) are deterministic.
@@ -229,6 +241,10 @@ async def _call(self, inputs: list[str]) -> list[list[float]]:
                 )
                 if not retriable:
                     raise
+                if read_retry_budget_active():
+                    # Let the enclosing reader/retriever spend its one budget;
+                    # don't multiply HTTP retries by tool/vector retries.
+                    raise
                 if attempt < self._max_retries:
                     retry_after = None
                     if resp is not None and resp.headers.get("retry-after"):
diff --git a/datamind/capabilities/kb/providers/hybrid_retriever.py b/datamind/capabilities/kb/providers/hybrid_retriever.py
index bcca7a8..66a5f4a 100644
--- a/datamind/capabilities/kb/providers/hybrid_retriever.py
+++ b/datamind/capabilities/kb/providers/hybrid_retriever.py
@@ -18,6 +18,7 @@
 from datamind.core.logging import get_logger
 from datamind.core.protocols import EmbeddingProvider, RetrievedChunk, VectorStore
 from datamind.core.registry import retriever_registry
+from datamind.core.recovery import recover_read
 from datamind.capabilities.kb.filters import matches_metadata, validate_metadata_filter
 
 _log = get_logger("retriever.hybrid")
@@ -54,6 +55,9 @@ def __init__(
         vector_weight: float = 1.0,
         bm25_weight: float = 1.0,
         candidate_multiplier: int = 3,
+        vector_failure_fallback: str = "error",
+        vector_retries: int = 0,
+        vector_timeout_s: float | None = None,
     ) -> None:
         self._store = vector_store
         self._embed = embedding
@@ -61,6 +65,15 @@ def __init__(
         self._vw = vector_weight
         self._bw = bm25_weight
         self._cm = candidate_multiplier
+        if vector_failure_fallback not in {"error", "bm25"}:
+            raise ValueError("vector_failure_fallback must be error or bm25")
+        if not 0 <= vector_retries <= 5:
+            raise ValueError("vector_retries must be between 0 and 5")
+        if vector_timeout_s is not None and vector_timeout_s <= 0:
+            raise ValueError("vector_timeout_s must be positive")
+        self._fallback = vector_failure_fallback
+        self._vector_retries = vector_retries
+        self._vector_timeout = vector_timeout_s
         # Lazy-built on first call; rebuildable.
         self._bm25: BM25Okapi | None = None
         self._ids: list[str] = []
@@ -109,157 +122,85 @@ async def rebuild_lexical(self) -> None:
         await self._ensure_lexical()
 
     async def aretrieve(
-        self,
-        query: str,
-        *,
-        top_k: int = 5,
+        self, query: str, *, top_k: int = 5,
         filters: dict[str, Any] | None = None,
     ) -> list[RetrievedChunk]:
-        validate_metadata_filter(filters)
-        await self._ensure_lexical()
-        k_inner = top_k * self._cm
-
-        # Vector side
-        vec = await self._embed.embed_query(query)
-        vec_hits = await self._store.query(vec, top_k=k_inner, where=filters)
-        vec_ranked = {ch.id: (rank, ch) for rank, ch in enumerate(vec_hits)}
-
-        # BM25 must use the exact same candidate scope as the vector branch.
-        # Filtering after fusion would allow an out-of-scope lexical hit to
-        # displace a valid vector candidate.
-        if self._bm25 is not None and self._ids:
-            scores = self._bm25.get_scores(_tokenize(query))
-            eligible = [
-                i for i in range(len(scores))
-                if matches_metadata(self._metas[i], filters)
-            ]
-            order = sorted(eligible, key=lambda i: -scores[i])[:k_inner]
-            bm25_ranked = {
-                self._ids[i]: (
-                    rank,
-                    RetrievedChunk(
-                        id=self._ids[i],
-                        text=self._texts[i],
-                        score=float(scores[i]),
-                        source=self._metas[i].get("source"),
-                        metadata=self._metas[i],
-                    ),
-                )
-                for rank, i in enumerate(order)
-                if scores[i] > 0
-            }
-        else:
-            bm25_ranked = {}
-
-        # Reciprocal Rank Fusion
-        fused: dict[str, tuple[float, RetrievedChunk]] = {}
-        for idmap, weight in ((vec_ranked, self._vw), (bm25_ranked, self._bw)):
-            for cid, (rank, ch) in idmap.items():
-                contrib = weight / (self._rrf_k + rank + 1)
-                if cid in fused:
-                    prev_score, prev_ch = fused[cid]
-                    fused[cid] = (prev_score + contrib, prev_ch)
-                else:
-                    fused[cid] = (contrib, ch)
-
-        top = sorted(fused.values(), key=lambda sc: -sc[0])[:top_k]
-        # Preserve fused score so callers can reason about it.
-        out = [
-            RetrievedChunk(
-                id=ch.id,
-                text=ch.text,
-                score=float(score),
-                source=ch.source,
-                metadata=ch.metadata,
-            )
-            for score, ch in top
-        ]
-        _log.info(
-            "retrieved",
-            extra={
-                "top_k": top_k,
-                "vec_hits": len(vec_ranked),
-                "bm25_hits": len(bm25_ranked),
-                "fused": len(out),
-            },
-        )
-        return out
+        return await self._retrieve(query, top_k=top_k, filters=filters, version=None)
 
     async def aretrieve_at_version(
-        self,
-        query: str,
-        *,
-        top_k: int = 5,
-        filters: dict[str, Any] | None = None,
-        version: int = None,
+        self, query: str, *, top_k: int = 5,
+        filters: dict[str, Any] | None = None, version: int | None = None,
     ) -> list[RetrievedChunk]:
-        validate_metadata_filter(filters)
-        _bm25, _ids, _texts, _metas = await self._ensure_lexical_at_version(version=version)
-        k_inner = top_k * self._cm
-
-        # Vector side
-        vec = await self._embed.embed_query(query)
-        vec_hits = await self._store.query(vec, top_k=k_inner, where=filters, version=version)
-        vec_ranked = {ch.id: (rank, ch) for rank, ch in enumerate(vec_hits)}
+        return await self._retrieve(query, top_k=top_k, filters=filters, version=version)
 
-        # BM25 must use the exact same candidate scope as the vector branch.
-        # Filtering after fusion would allow an out-of-scope lexical hit to
-        # displace a valid vector candidate.
-        if _bm25 is not None and _ids:
-            scores = _bm25.get_scores(_tokenize(query))
-            eligible = [
-                i for i in range(len(scores))
-                if matches_metadata(_metas[i], filters)
-            ]
-            order = sorted(eligible, key=lambda i: -scores[i])[:k_inner]
-            bm25_ranked = {
-                _ids[i]: (
-                    rank,
-                    RetrievedChunk(
-                        id=_ids[i],
-                        text=_texts[i],
-                        score=float(scores[i]),
-                        source=_metas[i].get("source"),
-                        metadata=_metas[i],
-                    ),
-                )
-                for rank, i in enumerate(order)
-                if scores[i] > 0
-            }
+    async def _retrieve(self, query, *, top_k, filters, version):
+        validate_metadata_filter(filters)
+        if top_k < 1:
+            raise ValueError("top_k must be positive")
+        if version is None:
+            await self._ensure_lexical()
+            bm25, ids, texts, metas = self._bm25, self._ids, self._texts, self._metas
         else:
-            bm25_ranked = {}
-
-        # Reciprocal Rank Fusion
+            # Lexical and vector reads must use exactly the same historical view.
+            bm25, ids, texts, metas = await self._ensure_lexical_at_version(version)
+        k_inner = top_k * self._cm
+        stage = "kb.embedding"
+
+        async def vector_search():
+            nonlocal stage
+            stage = "kb.embedding"
+            vec = await self._embed.embed_query(query)
+            stage = "kb.vector_query"
+            kwargs = {"top_k": k_inner, "where": filters}
+            if version is not None:
+                kwargs["version"] = version
+            return await self._store.query(vec, **kwargs)
+
+        def rank_lexical():
+            if bm25 is not None and ids:
+                scores = bm25.get_scores(_tokenize(query))
+                eligible = [i for i in range(len(scores)) if matches_metadata(metas[i], filters)]
+                order = sorted(eligible, key=lambda i: -scores[i])[:k_inner]
+                return {
+                    ids[i]: (rank, RetrievedChunk(
+                        id=ids[i], text=texts[i], score=float(scores[i]),
+                        source=metas[i].get("source"), metadata=metas[i],
+                    )) for rank, i in enumerate(order) if scores[i] > 0
+                }
+            else:
+                return {}
+
+        fallback_ranked = {}
+        async def lexical_fallback():
+            nonlocal fallback_ranked
+            fallback_ranked = rank_lexical()
+            return None  # Distinct from a successful vector search returning [].
+
+        vec_hits = await recover_read(
+            vector_search, component=lambda: stage, retries=self._vector_retries,
+            attempt_timeout_s=self._vector_timeout,
+            fallback=lexical_fallback if self._fallback == "bm25" else None,
+            fallback_method="bm25",
+        )
+        degraded = vec_hits is None
+        vec_ranked = {ch.id: (rank, ch) for rank, ch in enumerate(vec_hits or [])}
+        bm25_ranked = fallback_ranked if degraded else rank_lexical()
         fused: dict[str, tuple[float, RetrievedChunk]] = {}
-        for idmap, weight in ((vec_ranked, self._vw), (bm25_ranked, self._bw)):
-            for cid, (rank, ch) in idmap.items():
-                contrib = weight / (self._rrf_k + rank + 1)
+        # Lexical fallback remains useful even if lexical fusion was disabled.
+        lexical_weight = 1.0 if degraded else self._bw
+        for ranked, weight in ((vec_ranked, self._vw), (bm25_ranked, lexical_weight)):
+            for cid, (rank, ch) in ranked.items():
+                score = weight / (self._rrf_k + rank + 1)
                 if cid in fused:
-                    prev_score, prev_ch = fused[cid]
-                    fused[cid] = (prev_score + contrib, prev_ch)
+                    previous, old = fused[cid]
+                    fused[cid] = (previous + score, old)
                 else:
-                    fused[cid] = (contrib, ch)
-
-        top = sorted(fused.values(), key=lambda sc: -sc[0])[:top_k]
-        # Preserve fused score so callers can reason about it.
-        out = [
-            RetrievedChunk(
-                id=ch.id,
-                text=ch.text,
-                score=float(score),
-                source=ch.source,
-                metadata=ch.metadata,
-            )
-            for score, ch in top
-        ]
-        _log.info(
-            "retrieved",
-            extra={
-                "top_k": top_k,
-                "vec_hits": len(vec_ranked),
-                "bm25_hits": len(bm25_ranked),
-                "fused": len(out),
-                "version": version,
-            },
-        )
+                    fused[cid] = (score, ch)
+        out = [RetrievedChunk(
+            id=ch.id, text=ch.text, score=float(score), source=ch.source,
+            metadata={**ch.metadata, **({"degraded": True, "retrieval_method": "bm25"} if degraded else {})},
+        ) for score, ch in sorted(fused.values(), key=lambda item: -item[0])[:top_k]]
+        _log.info("retrieved", extra={"top_k": top_k, "vec_hits": len(vec_ranked),
+                  "bm25_hits": len(bm25_ranked), "fused": len(out),
+                  "version": version, "degraded": degraded})
         return out
diff --git a/datamind/capabilities/kb/service.py b/datamind/capabilities/kb/service.py
index da97784..c31add2 100644
--- a/datamind/capabilities/kb/service.py
+++ b/datamind/capabilities/kb/service.py
@@ -309,6 +309,9 @@ def build_kb_service(
     elif strategy == "hybrid":
         retriever = retriever_registry.create(
             "hybrid", vector_store=vector_store, embedding=embedding,
+            vector_failure_fallback=settings.retrieval.vector_failure_fallback,
+            vector_retries=settings.retrieval.vector_retries,
+            vector_timeout_s=settings.retrieval.vector_timeout_s,
         )
     else:
         raise ConfigError(
diff --git a/datamind/capabilities/kb/tools.py b/datamind/capabilities/kb/tools.py
index da323c6..831d35b 100644
--- a/datamind/capabilities/kb/tools.py
+++ b/datamind/capabilities/kb/tools.py
@@ -14,8 +14,12 @@
 from __future__ import annotations
 
 from typing import Any
+from dataclasses import replace
 
 from datamind.core.tools import ToolSpec, tool_provider_registry
+from datamind.core.context import RequestContext
+from datamind.core.logging import current_context, bind_context
+from datamind.core.recovery import recovery_summary
 
 from .service import KBService
 
@@ -23,8 +27,18 @@
 def build_kb_tools(kb: KBService) -> list[ToolSpec]:
     """Return the concrete ToolSpecs bound to this KBService instance."""
 
-    async def _search(query: str, top_k: int = 5, filters: dict | None = None) -> dict:
-        chunks = await kb.search(query, top_k=top_k, filters=filters)
+    async def _search_result(query, top_k, filters, version=None, historical=False):
+        parent = current_context()
+        ctx = replace(parent, extra={**parent.extra, "recovery_events": []}) if parent else RequestContext.new()
+        with bind_context(ctx):
+            try:
+                if not historical:
+                    chunks = await kb.search(query, top_k=top_k, filters=filters)
+                else:
+                    chunks = await kb.search_at_version(query, top_k=top_k, filters=filters, version=version)
+            finally:
+                if parent is not None:
+                    parent.extra.setdefault("recovery_events", []).extend(ctx.extra.get("recovery_events", []))
         return {
             "query": query,
             "top_k": top_k,
@@ -33,19 +47,14 @@ async def _search(query: str, top_k: int = 5, filters: dict | None = None) -> di
             "total_count": None,
             "truncated": len(chunks) >= top_k,
             "next_cursor": None,
+            "retrieval": recovery_summary(ctx.extra.get("recovery_events", [])),
         }
 
+    async def _search(query: str, top_k: int = 5, filters: dict | None = None) -> dict:
+        return await _search_result(query, top_k, filters)
+
     async def _search_at_version(query: str, top_k: int = 5, filters: dict | None = None, version: int = None) -> dict:
-        chunks = await kb.search_at_version(query, top_k=top_k, filters=filters, version=version, )
-        return {
-            "query": query,
-            "top_k": top_k,
-            "results": chunks,
-            "count": len(chunks),
-            "total_count": None,
-            "truncated": len(chunks) >= top_k,
-            "next_cursor": None,
-        }
+        return await _search_result(query, top_k, filters, version, historical=True)
 
     async def _list_documents() -> dict:
         items = await kb.list_documents()
@@ -127,7 +136,7 @@ async def _rollback(version: int) -> dict:
         ),
     ]
 
-    if kb is not None and hasattr(kb, "vector_store") and kb.vector_store.name == "dolt":
+    if kb is not None and hasattr(kb, "vector_store") and kb.vector_store.name == "lancedb":
         # Versioned KB Tools
         tools.extend([
             ToolSpec(
diff --git a/datamind/config.py b/datamind/config.py
index aa35e3b..a5bf552 100644
--- a/datamind/config.py
+++ b/datamind/config.py
@@ -89,6 +89,10 @@ class RetrievalConfig(BaseModel):
     chunk_overlap: int = 64
     rerank: bool = False
     rerank_model: str | None = None
+    # Opt-in read degradation; no source/metadata/validation errors are hidden.
+    vector_failure_fallback: Literal["error", "bm25"] = "error"
+    vector_retries: int = Field(default=0, ge=0, le=5)
+    vector_timeout_s: float | None = Field(default=None, gt=0)
 
     @field_validator("chunk_overlap")
     @classmethod
diff --git a/datamind/core/recovery.py b/datamind/core/recovery.py
new file mode 100644
index 0000000..58575a5
--- /dev/null
+++ b/datamind/core/recovery.py
@@ -0,0 +1,168 @@
+"""Bounded, observable read recovery. State is task/request scoped, never global."""
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import time
+from dataclasses import dataclass
+from typing import Any, Awaitable, Callable, TypeVar
+
+import httpx
+
+from .errors import CapabilityError, ConfigError, ExternalServiceError
+from .logging import current_context, get_logger
+
+T = TypeVar("T")
+_log = get_logger("read_recovery")
+
+
+def is_transient(exc: Exception) -> bool:
+    """Do not mask auth, invalid responses, bad filters, or local path errors."""
+    seen: set[int] = set()
+    while id(exc) not in seen:
+        seen.add(id(exc))
+        if isinstance(exc, (ConfigError, ValueError, PermissionError, FileNotFoundError)):
+            return False
+        if isinstance(exc, httpx.HTTPStatusError):
+            return exc.response.status_code in {408, 429} or exc.response.status_code >= 500
+        if isinstance(exc, ExternalServiceError) and exc.status_code is not None:
+            return exc.status_code in {408, 429} or exc.status_code >= 500
+        if isinstance(exc, (TimeoutError, ConnectionError, httpx.TransportError)):
+            return True
+        if isinstance(exc, OSError) and not isinstance(exc, (CapabilityError, ExternalServiceError)):
+            return True
+        cause = exc.__cause__
+        if not isinstance(cause, Exception):
+            return False
+        exc = cause
+    return False
+
+
+@dataclass
+class _Budget:
+    limit: int
+    used: int = 0
+    managed: bool = True
+
+
+_BUDGET: contextvars.ContextVar[_Budget | None] = contextvars.ContextVar("read_retry_budget", default=None)
+
+
+def read_retry_budget_active() -> bool:
+    budget = _BUDGET.get()
+    return budget is not None and budget.managed
+
+
+def recovery_summary(events: list[dict[str, Any]]) -> dict[str, Any]:
+    return {
+        "degraded": any(e.get("degraded", False) for e in events),
+        "retries": sum(e.get("retries", 0) for e in events),
+        "fallbacks": sum(e.get("degraded", False) for e in events),
+        "events": list(events),
+    }
+
+
+async def recover_read(
+    operation: Callable[[], Awaitable[T]],
+    *,
+    component: str | Callable[[], str],
+    retries: int = 0,
+    attempt_timeout_s: float | None = None,
+    fallback: Callable[[], Awaitable[T]] | None = None,
+    fallback_method: str = "fallback",
+    manage_provider_retries: bool = False,
+) -> T:
+    """Nested readers share a retry budget; the outer query supplies a deadline.
+
+    A vector attempt timeout may trigger lexical fallback. Cancellation and the
+    overall query deadline never trigger extra work after the budget has expired.
+    Events contain types/status only, not exception strings or user data.
+    """
+    budget = _BUDGET.get()
+    token = None
+    if budget is None:
+        budget = _Budget(retries, managed=manage_provider_retries or retries > 0 or fallback is not None)
+        token = _BUDGET.set(budget)
+    ctx = current_context()
+    deadline = ctx.extra.get("query_deadline") if ctx else None
+    started = time.monotonic()
+    event: dict[str, Any] = {"attempts": 0, "retries": 0, "degraded": False,
+                             "method": "primary", "outcome": "error", "retry_wait_ms": 0.0}
+    failure = None
+
+    def remaining() -> float | None:
+        if deadline is None:
+            return None
+        left = deadline - time.monotonic()
+        if left <= 0:
+            raise TimeoutError("query deadline exhausted")
+        return left
+
+    try:
+        while True:
+            left = remaining()
+            timeout = attempt_timeout_s
+            if left is not None:
+                timeout = left if timeout is None else min(timeout, left)
+            event["attempts"] += 1
+            try:
+                async with asyncio.timeout(timeout):
+                    result = await operation()
+                remaining()
+                event["outcome"] = "ok"
+                return result
+            except Exception as exc:
+                event["error_type"] = type(exc).__name__
+                event["status_code"] = getattr(exc, "status_code", None)
+                if not is_transient(exc):
+                    raise
+                remaining()
+                if budget.used < budget.limit:
+                    delay = 0.05 * 2 ** budget.used
+                    left = remaining()
+                    if left is not None and delay >= left:
+                        raise TimeoutError("query deadline exhausted") from exc
+                    budget.used += 1
+                    event["retries"] += 1
+                    await asyncio.sleep(delay)
+                    event["retry_wait_ms"] += delay * 1000
+                    continue
+                if fallback is None:
+                    raise
+                fallback_started = time.monotonic()
+                event["method"] = fallback_method
+                try:
+                    async with asyncio.timeout(remaining()):
+                        result = await fallback()
+                    remaining()
+                except BaseException:
+                    # The fallback has failed too; don't retry the whole nested
+                    # operation and don't report this as successful degradation.
+                    budget.used = budget.limit
+                    raise
+                finally:
+                    event["fallback_ms"] = round((time.monotonic() - fallback_started) * 1000, 3)
+                event.update(degraded=True, outcome="ok")
+                return result
+    except asyncio.CancelledError:
+        event["outcome"] = "cancelled"
+        raise
+    except Exception as exc:
+        failure = exc
+        raise
+    finally:
+        event["component"] = component() if callable(component) else component
+        event["elapsed_ms"] = round((time.monotonic() - started) * 1000, 3)
+        if ctx is not None:
+            event["workspace_revision"] = ctx.extra.get("workspace_revision")
+            ctx.extra.setdefault("recovery_events", []).append(event)
+            if failure is not None:
+                # Preserve the original exception type while exposing sanitized
+                # diagnostics to callers/experiment runners on failed reads.
+                try:
+                    failure.datamind_recovery = recovery_summary(ctx.extra["recovery_events"])
+                except (AttributeError, TypeError):
+                    pass
+        _log.info("read_recovery", extra={"recovery": event})
+        if token is not None:
+            _BUDGET.reset(token)
diff --git a/datamind/tests/test_read_recovery.py b/datamind/tests/test_read_recovery.py
new file mode 100644
index 0000000..f52d6be
--- /dev/null
+++ b/datamind/tests/test_read_recovery.py
@@ -0,0 +1,218 @@
+"""Degradation must preserve scope/version and must not hide deterministic errors."""
+import asyncio
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from datamind.capabilities.kb.providers.hybrid_retriever import HybridRetriever
+from datamind.capabilities.kb.tools import build_kb_tools
+from datamind.core.context import RequestContext
+from datamind.core.errors import ExternalServiceError
+from datamind.core.logging import bind_context
+from datamind.core.recovery import recover_read, recovery_summary
+
+
+class Embedding:
+    def __init__(self, error=None):
+        self.error = error
+        self.calls = 0
+
+    async def embed_query(self, query):
+        self.calls += 1
+        if self.error:
+            raise self.error
+        return [1., 0.]
+
+
+class Store:
+    def __init__(self, error=None):
+        self.error = error
+        self.versions = []
+        self.rows = {
+            1: [("old", "needle old value", {"tenant": "a", "source": "a.md"})],
+            2: [("new", "needle new value", {"tenant": "a", "source": "a.md"})],
+        }
+        for rows in self.rows.values():
+            rows.extend((str(i), f"filler unrelated document {i}", {"tenant": "b"}) for i in range(5))
+
+    async def get_all_texts(self, version=None):
+        self.versions.append(version)
+        return self.rows[version or 2]
+
+    async def query(self, vec, **kwargs):
+        if self.error:
+            raise self.error
+        return []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stage", ["embedding", "vector"])
+async def test_transient_failure_falls_back_after_bounded_retries(stage):
+    embed = Embedding(TimeoutError() if stage == "embedding" else None)
+    store = Store(TimeoutError() if stage == "vector" else None)
+    retriever = HybridRetriever(vector_store=store, embedding=embed,
+        vector_failure_fallback="bm25", vector_retries=1)
+    ctx = RequestContext.new()
+    with bind_context(ctx):
+        hits = await retriever.aretrieve("needle", filters={"tenant": "a"})
+    assert [h.id for h in hits] == ["new"]
+    assert hits[0].metadata["degraded"] is True
+    assert embed.calls == 2
+    summary = recovery_summary(ctx.extra["recovery_events"])
+    assert summary["retries"] == 1 and summary["fallbacks"] == 1
+    assert summary["events"][0]["component"] == f"kb.{'embedding' if stage == 'embedding' else 'vector_query'}"
+
+
+@pytest.mark.asyncio
+async def test_historical_fallback_never_reads_current_lexical_data():
+    store = Store()
+    retriever = HybridRetriever(vector_store=store, embedding=Embedding(TimeoutError()),
+                                vector_failure_fallback="bm25")
+    old, new = await asyncio.gather(
+        retriever.aretrieve_at_version("needle", version=1),
+        retriever.aretrieve_at_version("needle", version=2),
+    )
+    assert old[0].id == "old" and new[0].id == "new"
+    assert sorted(store.versions) == [1, 2]
+    assert await retriever.aretrieve_at_version("needle", version=1, filters={"tenant": "b"}) == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("error", [
+    ValueError("bad input"), ExternalServiceError("embedding", "auth", status_code=401),
+    ExternalServiceError("embedding", "malformed response"), FileNotFoundError("missing index"),
+])
+async def test_deterministic_errors_are_not_retried_or_hidden(error):
+    embed = Embedding(error)
+    retriever = HybridRetriever(vector_store=Store(), embedding=embed,
+                                vector_failure_fallback="bm25", vector_retries=2)
+    with pytest.raises(type(error)):
+        await retriever.aretrieve("needle")
+    assert embed.calls == 1
+
+
+@pytest.mark.asyncio
+async def test_disabled_fallback_propagates_and_nested_retry_budget_is_shared():
+    embed = Embedding(TimeoutError())
+    retriever = HybridRetriever(vector_store=Store(), embedding=embed, vector_retries=3)
+    ctx = RequestContext.new()
+    with bind_context(ctx), pytest.raises(TimeoutError):
+        await recover_read(lambda: retriever.aretrieve("needle"), component="kb_search", retries=1)
+    assert embed.calls == 2  # Not (outer retries + 1) * (inner retries + 1).
+    assert recovery_summary(ctx.extra["recovery_events"])["retries"] == 1
+
+
+@pytest.mark.asyncio
+async def test_successful_empty_vector_result_is_not_a_fault():
+    retriever = HybridRetriever(vector_store=Store(), embedding=Embedding(), vector_failure_fallback="bm25")
+    hits = await retriever.aretrieve("needle")
+    assert hits and "degraded" not in hits[0].metadata
+
+
+@pytest.mark.asyncio
+async def test_empty_fallback_is_observable_in_tool_result():
+    retriever = HybridRetriever(vector_store=Store(), embedding=Embedding(TimeoutError()),
+                                vector_failure_fallback="bm25")
+
+    async def search(query, **kwargs):
+        return [h.model_dump() for h in await retriever.aretrieve(query, **kwargs)]
+
+    tool = next(t for t in build_kb_tools(SimpleNamespace(
+        search=search, data_dir=".", vector_store=SimpleNamespace(name="test"),
+    )) if t.name == "kb_search")
+    result = await tool.handler(query="no_matching_term")
+    assert result["results"] == []
+    assert result["retrieval"]["degraded"] is True
+    assert result["retrieval"]["events"][0]["method"] == "bm25"
+
+
+@pytest.mark.asyncio
+async def test_unavailable_lexical_source_is_not_silently_empty():
+    store = Store()
+    async def unavailable(**kwargs):
+        raise TimeoutError("text storage offline")
+    store.get_all_texts = unavailable
+    retriever = HybridRetriever(vector_store=store, embedding=Embedding(TimeoutError()),
+                                vector_failure_fallback="bm25")
+    with pytest.raises(TimeoutError):
+        await retriever.aretrieve("needle")
+
+
+@pytest.mark.asyncio
+async def test_attempt_timeout_can_fallback_but_query_deadline_cannot():
+    embed = Embedding()
+    async def slow(query):
+        await asyncio.sleep(.1)
+        return [1., 0.]
+    embed.embed_query = slow
+    retriever = HybridRetriever(vector_store=Store(), embedding=embed,
+                                vector_failure_fallback="bm25", vector_timeout_s=.005)
+    assert (await retriever.aretrieve("needle"))[0].metadata["degraded"]
+    ctx = RequestContext.new()
+    ctx.extra["query_deadline"] = time.monotonic() + .005
+    with bind_context(ctx), pytest.raises(TimeoutError):
+        await retriever.aretrieve("needle")
+    assert not recovery_summary(ctx.extra["recovery_events"])["degraded"]
+
+
+@pytest.mark.asyncio
+async def test_cancellation_never_triggers_fallback():
+    embed = Embedding(asyncio.CancelledError())
+    retriever = HybridRetriever(vector_store=Store(), embedding=embed, vector_failure_fallback="bm25")
+    with pytest.raises(asyncio.CancelledError):
+        await retriever.aretrieve("needle")
+
+
+@pytest.mark.asyncio
+async def test_request_recovery_records_do_not_leak():
+    retriever = HybridRetriever(vector_store=Store(), embedding=Embedding(TimeoutError()),
+                                vector_failure_fallback="bm25")
+    async def run():
+        ctx = RequestContext.new()
+        with bind_context(ctx):
+            await retriever.aretrieve("needle")
+        return ctx.extra["recovery_events"]
+    first, second = await asyncio.gather(run(), run())
+    assert first is not second and len(first) == len(second) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("retries", [0, 1])
+async def test_http_provider_does_not_multiply_outer_retry_budget(retries):
+    import httpx
+    from datamind.capabilities.embedding.providers.openai_compatible import OpenAICompatibleEmbedding
+    calls = 0
+    def response(request):
+        nonlocal calls
+        calls += 1
+        return httpx.Response(503, text="offline")
+    async with httpx.AsyncClient(transport=httpx.MockTransport(response)) as client:
+        embed = OpenAICompatibleEmbedding(api_key="test", model="test", dimension=2,
+                                         max_retries=3, client=client)
+        with pytest.raises(ExternalServiceError):
+            await recover_read(lambda: embed.embed_query("query"), component="test",
+                               retries=retries, manage_provider_retries=True)
+    assert calls == retries + 1
+
+
+@pytest.mark.asyncio
+async def test_unmanaged_http_provider_retains_existing_retry_behavior_and_usage():
+    import httpx
+    from datamind.capabilities.embedding.providers.openai_compatible import OpenAICompatibleEmbedding
+    calls = 0
+    def response(request):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return httpx.Response(503, text="offline")
+        return httpx.Response(200, json={"data": [{"index": 0, "embedding": [1., 0.]}],
+                                        "usage": {"total_tokens": 7}})
+    ctx = RequestContext.new()
+    async with httpx.AsyncClient(transport=httpx.MockTransport(response)) as client:
+        embed = OpenAICompatibleEmbedding(api_key="test", model="test", dimension=2,
+            max_retries=1, backoff_base_s=.001, client=client)
+        with bind_context(ctx):
+            assert await embed.embed_query("query") == [1., 0.]
+    assert calls == 2
+    assert ctx.extra["embedding_usage"] == [{"model": "test", "total_tokens": 7}]
diff --git a/datamind/tests/test_robustness_runner.py b/datamind/tests/test_robustness_runner.py
new file mode 100644
index 0000000..de95e79
--- /dev/null
+++ b/datamind/tests/test_robustness_runner.py
@@ -0,0 +1,44 @@
+"""Exercise the public experiment CLI, including a real child process exit."""
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def run_cli(*args):
+    result = subprocess.run([sys.executable, "-m", "benchmark.robustness", *args],
+        cwd=REPO, env={**os.environ, "PYTHONPATH": str(REPO)}, capture_output=True, timeout=60)
+    assert result.returncode == 0, result.stderr.decode()
+
+
+def test_query_experiment_pairs_policies_and_labels_fixture_results(tmp_path):
+    output = tmp_path / "query"
+    run_cli("query", "--fixture", "--output", str(output),
+            "--failure-probabilities", "0", "1", "--repeats", "1")
+    manifest = json.loads((output / "manifest.json").read_text())
+    summary = json.loads((output / "summary.json").read_text())
+    assert manifest["result_kind"] == "fixture_smoke" and manifest["code_hash"]
+    groups = {(g["policy"], g["failure_probability"]): g for g in summary["groups"]}
+    assert groups["none", 1]["completion_rate"] == 0
+    assert groups["retry", 1]["completion_rate"] == 0
+    assert groups["fallback", 1]["source_hit_rate"] == 1
+    assert groups["fallback", 1]["exact_answer_match_rate"] is None
+    rows = [json.loads(line) for line in (output / "rows.jsonl").read_text().splitlines()]
+    assert len(rows) == 12
+    assert len({r["workspace_revision"] for r in rows}) == 1
+    degraded = [r for r in rows if r["policy"] == "fallback" and r["failure_probability"] == 1]
+    assert all(r["result"]["retrieval"]["degraded"] for r in degraded)
+
+
+def test_process_exit_recovery_experiment_matches_cold_content(tmp_path):
+    output = tmp_path / "build"
+    run_cli("build", "--fixture", "--after", "2", "--output", str(output))
+    result = json.loads((output / "summary.json").read_text())
+    assert result["fixture"] and result["logical_content_equal"]
+    assert result["resume_metrics"]["sources_reused"] == 2
+    assert result["cold_metrics"]["sources_reused"] == 0
+    assert result["resume_metrics"]["graph_extraction_calls"] < result["cold_metrics"]["graph_extraction_calls"]
+    assert result["resume_metrics"]["materialization_ms"] > 0
diff --git a/datamind/tests/test_workspace_revisions.py b/datamind/tests/test_workspace_revisions.py
index 57cd6aa..225c44d 100644
--- a/datamind/tests/test_workspace_revisions.py
+++ b/datamind/tests/test_workspace_revisions.py
@@ -127,6 +127,14 @@ async def test_unchanged_sources_reuse_embeddings_and_extraction(setup_workspace
     second = await workspace.update(str(source))
     assert second['reused_sources'] == 1
     assert (model.calls, embedding.calls) == calls
+    assert second['metrics']['sources_reused'] == 1
+    assert second['metrics']['sources_computed'] == 0
+    assert second['metrics']['embedding_calls'] == second['metrics']['graph_extraction_calls'] == 0
+    assert second['metrics']['materialization_ms'] > 0
+    report = workspace._settings_for(second['workspace_revision']).data.base_dir / 'build_metrics.json'
+    saved = json.loads(report.read_text())
+    assert saved['total_ms'] > 0 and saved['status'] == 'published'
+    assert saved['reported_embedding_tokens'] is None  # Fixture doesn't report usage.
 
 
 @pytest.mark.asyncio
@@ -166,6 +174,27 @@ async def fail_once(*args, **kwargs):
         assert workspace.current_revision != reader.revision
 
 
+@pytest.mark.asyncio
+async def test_lexical_fallback_keeps_old_sources_after_new_publication(setup_workspace):
+    workspace, source, _, embedding = setup_workspace
+    workspace.settings.retrieval.strategy = 'hybrid'
+    workspace.settings.retrieval.vector_failure_fallback = 'bm25'
+    for i in range(5):
+        (source / f'filler{i}.md').write_text(f'unrelated document number {i}')
+    first = await workspace.update(str(source), surfaces=['kb'])
+    async with workspace.reader(first['workspace_revision'], read_retries=1) as reader:
+        (source/'a.md').write_text('Bob owns Project')
+        await workspace.update(str(source), surfaces=['kb'])
+        async def unavailable(query):
+            raise TimeoutError('embedding unavailable')
+        embedding.embed_query = unavailable
+        result = await reader.call('kb_search', query='Alice')
+        assert 'Alice owns' in str(result['results']) and 'Bob owns' not in str(result['results'])
+        assert result['retrieval']['degraded'] is True
+        assert result['retrieval']['retries'] == 1
+        assert result['retrieval']['events'][0]['workspace_revision'] == first['workspace_revision']
+
+
 @pytest.mark.asyncio
 async def test_inspection_reads_pinned_sources_after_deletion(setup_workspace):
     workspace, source, _, _ = setup_workspace
diff --git a/datamind/workspace.py b/datamind/workspace.py
index 76483b6..81a498b 100644
--- a/datamind/workspace.py
+++ b/datamind/workspace.py
@@ -13,6 +13,7 @@
 import json
 import os
 import sqlite3
+import time
 import uuid
 from dataclasses import replace
 from contextlib import asynccontextmanager
@@ -25,6 +26,7 @@
 from datamind.core.errors import CapabilityError, ConfigError, ExternalServiceError
 from datamind.core.logging import bind_context, current_context
 from datamind.core.protocols import GraphTriple
+from datamind.core.recovery import recover_read, recovery_summary
 
 
 def _digest(value: bytes) -> str:
@@ -130,6 +132,52 @@ def _inventory(self, source: Path) -> list[dict[str, str]]:
         return inventory
 
     async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_types=None, max_triples_per_file=200) -> dict[str, Any]:
+        """Publish a generation and report logical compute/materialization costs.
+
+        Metrics are per attempt, including unsuccessful attempts. Provider usage
+        is reported only when available; absent token usage is not a zero cost.
+        """
+        metrics = {"sources_total": 0, "sources_reused": 0, "sources_computed": 0,
+                   "sources_completed": 0, "embedding_calls": 0, "embedding_texts": 0,
+                   "graph_extraction_calls": 0, "parsed_table_files": 0,
+                   "inventory_ms": 0., "setup_ms": 0., "source_copy_ms": 0.,
+                   "checkpoint_read_ms": 0., "checkpoint_write_ms": 0.,
+                   "source_compute_ms": 0., "materialization_ms": 0., "publication_ms": 0.}
+        parent = current_context()
+        ctx = replace(parent, extra=dict(parent.extra)) if parent else RequestContext.new(profile=self.settings.data.profile)
+        ctx.extra.update(nested_model_usage=[], embedding_usage=[])
+        started = time.monotonic()
+        try:
+            with bind_context(ctx):
+                result = await self._update(path, surfaces=surfaces, relation_types=relation_types,
+                    max_triples_per_file=max_triples_per_file, metrics=metrics)
+            metrics["status"] = "published"
+            return {**result, "metrics": metrics}
+        except BaseException as exc:
+            metrics.update(status="failed", error_type=type(exc).__name__)
+            raise
+        finally:
+            metrics["total_ms"] = round((time.monotonic() - started) * 1000, 3)
+            usage = ctx.extra.get("nested_model_usage", [])
+            embed_usage = ctx.extra.get("embedding_usage", [])
+            metrics["model_usage_records"] = usage
+            metrics["embedding_usage_records"] = embed_usage
+            metrics["reported_model_input_tokens"] = sum(x.get("input_tokens", 0) for x in usage) if usage else None
+            metrics["reported_model_output_tokens"] = sum(x.get("output_tokens", 0) for x in usage) if usage else None
+            metrics["reported_embedding_tokens"] = sum(x.get("total_tokens", 0) for x in embed_usage) if embed_usage else None
+            if parent is not None:
+                parent.extra.setdefault("nested_model_usage", []).extend(usage)
+                parent.extra.setdefault("embedding_usage", []).extend(embed_usage)
+            if metrics.get("workspace_revision"):
+                report = self._settings_for(metrics["workspace_revision"]).data.base_dir / "build_metrics.json"
+                try:
+                    _atomic_json(report, metrics)
+                except OSError:
+                    # An observability failure must not undo a published head or
+                    # replace the original build error.
+                    metrics["metrics_persisted"] = False
+
+    async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_file, metrics) -> dict[str, Any]:
         """Prepare and publish one generation, retaining the previous one on failure.
 
         Cached complete-source results are reused. Index materialization is currently
@@ -145,7 +193,10 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
         ):
             raise ConfigError('workspace snapshots currently require local NetworkX and SQLite without an external DSN')
         source = Path(path).expanduser().resolve()
+        stage_started = time.monotonic()
         inventory = await asyncio.to_thread(self._inventory, source)
+        metrics['inventory_ms'] = (time.monotonic() - stage_started) * 1000
+        metrics['sources_total'] = len(inventory)
         pipeline = {'version': 'workspace-v1', 'surfaces': sorted(selected),
                     'embedding': self.settings.embedding.model_dump(mode='json', exclude={'api_key'}),
                     'model': self.settings.llm.fallback_model or self.settings.llm.model,
@@ -158,6 +209,7 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
         # Persist only the fingerprint; configuration can include private endpoints.
         pipeline_hash = _digest(json.dumps(pipeline, sort_keys=True, default=str).encode())
         revision = 'revision-' + uuid.uuid4().hex
+        metrics['workspace_revision'] = revision
         with self._connect() as conn:
             parent, parent_epoch = conn.execute('SELECT revision, epoch FROM head WHERE singleton=1').fetchone()
             conn.execute('INSERT INTO revisions VALUES (?, ?, ?, NULL)', (revision, parent, 'BUILDING'))
@@ -168,11 +220,14 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
         truncated_sources = []
         bindings = {}
         try:
+            stage_started = time.monotonic()
             system = await self._system(settings, selected)
+            metrics['setup_ms'] = (time.monotonic() - stage_started) * 1000
             ingest = system.services.ingest
             (settings.data.data_dir / 'uploads').mkdir(parents=True, exist_ok=True)
             graph_triples = []
             for item in inventory:
+                stage_started = time.monotonic()
                 original = Path(item['uri'])
                 snapshot = settings.data.data_dir / 'uploads' / item['relative_path']
                 snapshot.parent.mkdir(parents=True, exist_ok=True)
@@ -180,12 +235,16 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                 if 'sha256:' + _digest(raw) != item['version']:
                     raise CapabilityError('workspace', 'source changed during build; retry')
                 snapshot.write_bytes(raw)
+                metrics['source_copy_ms'] += (time.monotonic() - stage_started) * 1000
                 source_id = 'file:' + _digest(item['uri'].encode())
                 cache_key = _digest(json.dumps([source_id, item['version'], pipeline_hash]).encode())
                 cache_path = self.root / 'checkpoints' / (cache_key + '.json')
+                stage_started = time.monotonic()
                 if cache_path.is_file():
                     prepared = json.loads(cache_path.read_text())
                     reused += 1
+                    metrics['sources_reused'] += 1
+                    metrics['checkpoint_read_ms'] += (time.monotonic() - stage_started) * 1000
                 else:
                     prepared = {'chunks': [], 'vectors': [], 'triples': [], 'tables': [], 'truncated': False}
                     is_table = original.suffix.lower() in TABLE_EXTS
@@ -200,6 +259,8 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                             for i, part in enumerate(parts)
                         ]
                         if parts:
+                            metrics['embedding_calls'] += 1
+                            metrics['embedding_texts'] += len(parts)
                             prepared['vectors'] = await system.services.kb.embedding.embed_texts(parts)
                     if 'graph' in selected and text and not is_table:
                         parts = _split_text(text, chunk_size=settings.retrieval.chunk_size, chunk_overlap=0)
@@ -208,6 +269,7 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                             if remaining <= 0:
                                 prepared['truncated'] = True
                                 break
+                            metrics['graph_extraction_calls'] += 1
                             triples = await ingest._extract_graph_triples(
                                 text=part, max_triples=min(30, remaining), source=item['uri'],
                                 source_version=item['version'], locator={'part': i},
@@ -217,9 +279,15 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                             if len(triples) >= min(30, remaining):
                                 prepared['truncated'] = True
                     if 'db' in selected and is_table:
+                        metrics['parsed_table_files'] += 1
                         prepared['tables'] = await asyncio.to_thread(extract_tabular, snapshot)
+                    metrics['source_compute_ms'] += (time.monotonic() - stage_started) * 1000
+                    metrics['sources_computed'] += 1
+                    stage_started = time.monotonic()
                     _atomic_json(cache_path, prepared, overwrite=False)
                     prepared = json.loads(cache_path.read_text())
+                    metrics['checkpoint_write_ms'] += (time.monotonic() - stage_started) * 1000
+                stage_started = time.monotonic()
                 if prepared.get('truncated'):
                     truncated_sources.append(item['uri'])
                 if prepared['chunks']:
@@ -237,6 +305,9 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                                               if_exists='replace', source_file=item['uri'])
                 manifest_sources.append({**item, 'source_id': source_id,
                     'snapshot': str(snapshot.relative_to(settings.data.base_dir))})
+                metrics['sources_completed'] += 1
+                metrics['materialization_ms'] += (time.monotonic() - stage_started) * 1000
+            stage_started = time.monotonic()
             if 'graph' in selected:
                 await system.services.graph.store.upsert_triples(graph_triples)
                 await system.services.graph.store.persist()
@@ -257,6 +328,8 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                     'path': str(db_path.relative_to(settings.data.base_dir))}
             await system.aclose()
             system = None
+            metrics['materialization_ms'] += (time.monotonic() - stage_started) * 1000
+            stage_started = time.monotonic()
             if await asyncio.to_thread(self._inventory, source) != inventory:
                 raise CapabilityError('workspace', 'workspace changed during build; retry')
             manifest = {'workspace_revision': revision, 'parent_revision': parent, 'source_root': str(source), 'parent_epoch': parent_epoch,
@@ -272,6 +345,7 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
                     raise CapabilityError('workspace', 'publication conflict; another build published first')
                 conn.execute('UPDATE revisions SET state=? WHERE id=?', ('PUBLISHED', revision))
                 conn.execute('UPDATE head SET revision=?, epoch=epoch+1 WHERE singleton=1', (revision,))
+            metrics['publication_ms'] = (time.monotonic() - stage_started) * 1000
             return manifest
         except BaseException as exc:
             with self._connect() as conn:
@@ -291,7 +365,8 @@ async def reader(self, revision: str | None = None, *, read_retries: int = 1):
         settings = self._settings_for(revision)
         system = await self._system(settings, manifest['surfaces'])
         try:
-            reader = WorkspaceReader(system, manifest, settings.data.base_dir, read_retries=read_retries)
+            reader = WorkspaceReader(system, manifest, settings.data.base_dir, read_retries=read_retries,
+                                     timeout_s=settings.agent.wall_clock_timeout_s)
             yield reader
         finally:
             await system.aclose()
@@ -304,10 +379,11 @@ async def query(self, message: str, *, revision: str | None = None, read_retries
 class WorkspaceReader:
     """Read-only facade fixed to a retained generation, including raw evidence."""
 
-    def __init__(self, system, manifest, root, *, read_retries=1):
+    def __init__(self, system, manifest, root, *, read_retries=1, timeout_s=300.):
         if not 0 <= read_retries <= 5:
             raise ValueError("read_retries must be between 0 and 5")
         self._system = system
+        self._timeout = timeout_s
         self.manifest = manifest
         self.revision = manifest['workspace_revision']
         mapping = {s['uri']: root / s['snapshot'] for s in manifest['sources']}
@@ -353,34 +429,26 @@ async def search_pinned(version: int, **kwargs):
             spec = system.retrieve.tools.get(name)
             original_handler = spec.handler
 
-            async def retry_read(_handler=original_handler, **kwargs):
-                for attempt in range(read_retries + 1):
-                    try:
-                        return await _handler(**kwargs)
-                    except Exception as exc:
-                        cause = exc.__cause__ if isinstance(exc, CapabilityError) else exc
-                        transient = isinstance(cause, (OSError, TimeoutError)) or (
-                            isinstance(cause, ExternalServiceError) and
-                            (cause.status_code is None or cause.status_code in {408, 429} or cause.status_code >= 500))
-                        if not transient or attempt == read_retries:
-                            raise
-                        context = current_context()
-                        if context is not None:
-                            context.extra['workspace_read_retries'] = context.extra.get('workspace_read_retries', 0) + 1
-                        await asyncio.sleep(0.05 * 2 ** attempt)
+            async def retry_read(_handler=original_handler, _name=name, **kwargs):
+                return await recover_read(lambda: _handler(**kwargs), component=_name,
+                                          retries=read_retries, manage_provider_retries=True)
             system.retrieve.tools.replace(replace(spec, handler=retry_read))
 
     async def call(self, name: str, **kwargs):
         context = RequestContext.new(profile=self._system.profile)
         context.extra['workspace_revision'] = self.revision
+        context.extra['query_deadline'] = time.monotonic() + self._timeout
         with bind_context(context):
             return await self._system.retrieve.tools.get(name).handler(**kwargs)
 
     async def query(self, message: str, **kwargs):
         context = RequestContext.new(profile=self._system.profile)
         context.extra['workspace_revision'] = self.revision
+        context.extra['query_deadline'] = time.monotonic() + self._timeout
         with bind_context(context):
-            result = await self._system.query(message, **kwargs)
+            async with asyncio.timeout(self._timeout):
+                result = await self._system.query(message, **kwargs)
         return {**result, 'workspace_revision': self.revision,
                 'truncated_sources': self.manifest.get('truncated_sources', []),
-                'workspace_read_retries': context.extra.get('workspace_read_retries', 0)}
+                'workspace_read_retries': recovery_summary(context.extra.get('recovery_events', []))['retries'],
+                'recovery': recovery_summary(context.extra.get('recovery_events', []))}
diff --git a/docs/BENCHMARK_RUNNER.md b/docs/BENCHMARK_RUNNER.md
index 318bb91..06f7fcb 100644
--- a/docs/BENCHMARK_RUNNER.md
+++ b/docs/BENCHMARK_RUNNER.md
@@ -1,5 +1,8 @@
 # Benchmark runner
 
+For paired read-fault and actual process-interruption experiments, see
+[Read recovery and robustness experiments](READ_RECOVERY.md).
+
 DataMind provides a dataset-neutral benchmark entry point for repeatable,
 long-running evaluations. Dataset adapters, official scorers, expected answers,
 and task-specific anomaly rules belong to the evaluation project that owns
diff --git a/docs/READ_RECOVERY.md b/docs/READ_RECOVERY.md
new file mode 100644
index 0000000..e6160bd
--- /dev/null
+++ b/docs/READ_RECOVERY.md
@@ -0,0 +1,152 @@
+# Read recovery and robustness experiments
+
+DataMind supports opt-in lexical fallback for hybrid retrieval and source-level
+build checkpoint reuse. These are execution mechanisms, not guarantees of answer
+correctness. Research quality, latency, and cost must be evaluated separately.
+
+## Configure query recovery
+
+```bash
+export DATAMIND__RETRIEVAL__STRATEGY=hybrid
+export DATAMIND__RETRIEVAL__VECTOR_FAILURE_FALLBACK=bm25
+export DATAMIND__RETRIEVAL__VECTOR_RETRIES=1
+export DATAMIND__RETRIEVAL__VECTOR_TIMEOUT_S=2
+```
+
+Defaults preserve the existing behavior: fallback is `error`, vector retries are
+zero, and no additional vector-attempt timeout is imposed. Without managed read
+recovery, the embedding provider retains its configured HTTP retry policy.
+
+- A successful vector call returning no hits is not a fault.
+- Only transient connectivity/timeouts and retryable service statuses trigger
+  read recovery. Auth failures, invalid filters/configuration, missing local
+  files, and malformed embedding responses remain errors.
+- After the retry budget is exhausted, an enabled fallback searches the same
+  lexical corpus with the same metadata filters. Historical queries use the
+  requested historical text, never latest text. Published workspace readers use
+  their private immutable source/index generation.
+- If lexical data cannot be loaded, the error is propagated. An empty lexical
+  result is explicitly reported as a degraded empty result, not fabricated facts.
+- There is no automatic DB-to-Python or KG-to-document replacement.
+
+For `Workspace.reader(..., read_retries=N)`, the reader's per-tool retry budget
+takes precedence over `vector_retries`. Nested vector/tool recovery and the
+built-in HTTP embedding provider share that budget rather than multiplying their
+retry counts. Standalone hybrid reads use `vector_retries`. Third-party clients
+may still have their own retry implementations; configure them explicitly for
+fair experiments.
+
+`vector_timeout_s` bounds each vector attempt. `RetrieveAgent.query` and
+`Workspace.query` also enforce the configured `agent.wall_clock_timeout_s` across
+the whole query, including retries, sleeps and fallback. Direct `KBService`
+callers remain responsible for setting an overall deadline. Cancellation and an
+exhausted overall deadline never initiate fallback.
+
+## Observe reads and builds
+
+`kb_search` / `kb_search_at_version` return a `retrieval` field even for zero hits.
+It includes `degraded`, retry/fallback counts and events (component, attempts,
+error type/status, method, outcome and duration). Degraded chunks also carry
+`metadata.degraded=true` and `metadata.retrieval_method=bm25`.
+
+Non-streaming query results expose a `recovery` summary; benchmark rows preserve
+it. Events are request scoped, and workspace events include the pinned revision.
+Nested events overlap in time: do not sum their elapsed durations as query cost.
+Streaming consumers can inspect the KB tool result's `retrieval` field; there is
+not yet a backend-neutral streaming final recovery summary.
+
+`Workspace.update()` returns `metrics` and saves `build_metrics.json` alongside
+the revision. Metrics include completed/reused/computed sources, logical
+embedding/extraction calls, checkpoint load/write, compute, source-copy,
+materialization, publication and total durations. Completed phase timings do not
+include unfinished work in a failed source; total duration includes failures.
+Provider-reported token records are retained when available; unavailable usage
+is `null`, not an estimate or a claim of zero billed cost. Logical call counters
+are not HTTP request counters, and provider-reported usage may omit failed calls.
+
+Build metrics are observability sidecars, not mutable snapshot contents.
+Caught failures write a metrics sidecar when possible. An abrupt process kill
+cannot finalize metrics; use the external experiment record for that event.
+Re-running `update()` creates a new candidate and reuses complete-source
+checkpoints. The unfinished source is retried in full. Index materialization
+still builds a complete generation; this is compute reuse, not sub-file
+maintenance or automatic background task resumption.
+
+## CPU-only smoke experiments
+
+From a checkout with normal DataMind dependencies installed:
+
+```bash
+python -m benchmark.robustness query --fixture \
+  --failure-probabilities 0 0.1 0.5 1 --repeats 3 --seed 7 \
+  --output /tmp/datamind-query-faults
+
+python -m benchmark.robustness build --fixture --after 2 \
+  --output /tmp/datamind-build-recovery
+
+python -m benchmark.robustness build --fixture --crash-stage publication \
+  --output /tmp/datamind-publication-recovery
+```
+
+Output directories must be new. Fixtures replace model/embedding services with
+deterministic CPU-only doubles but use actual local snapshot storage. Every
+output is marked `fixture_smoke`; these results are not LLM accuracy or model
+cost measurements. Fixture extraction reports no token usage.
+
+## Query fault workload
+
+Live mode uses normal Settings/environment. A task file is JSONL, for example:
+
+```json
+{"task_id":"q1","question":"What is the budget for Project A?","expected_sources":["budget.md"],"reference_answer":"120"}
+```
+
+Labels are never passed to the agent. `expected_sources` measures retrieval
+source hit rate; `reference_answer` enables exact string matching only in agent
+mode. They are optional. Official benchmark scoring should be applied externally.
+
+```bash
+python -m benchmark.robustness query \
+  --source /absolute/workspace --questions /absolute/tasks.jsonl \
+  --execution agent --surfaces kb,graph,db --component embedding \
+  --failure-probabilities 0 0.05 0.1 0.2 --repeats 3 --seed 7 \
+  --retries 1 --deadline-s 60 --output /absolute/new-experiment
+```
+
+The runner builds one revision and compares `none`, `retry`, and `fallback`
+against it. Faults target embedding or vector queries. `--outage-ms` sets an
+initial per-query outage; `--fault-delay-ms` sets failed-call latency. Schedules
+are paired by seed/task/repeat/attempt across policies. Order is seeded/shuffled.
+Reader setup time is separate from execution latency. JSONL rows retain outputs,
+fault/attempt counts, errors, usage and recovery metadata; summary completion is
+operation completion, not automatically task accuracy. Unlabeled quality rates
+are null. Fault timing can interact with model speed; report this explicitly.
+
+The manifest records the revision, pipeline/configuration, code hash (including
+untracked Python source), input hashes, seed and fixture/live mode. It omits
+credentials and endpoint configuration. Pin dependencies and retain the source
+checkout/environment for reproduction. This runner doesn't overwrite or append
+to a prior experiment and does not automatically make API-priced claims.
+
+## Build interruption workload
+
+```bash
+python -m benchmark.robustness build --source /absolute/workspace \
+  --surfaces kb,graph,db --after 10 --worker-timeout-s 600 \
+  --questions /absolute/tasks.jsonl --output /absolute/new-build-experiment
+```
+
+The worker really exits (`os._exit`) after N source checkpoints or immediately
+before publication. The parent checks that no partial revision was published.
+It copies the same interrupted state for both policies: `cold` removes source
+checkpoints; `resume` retains them. Both rebuild private indexes normally.
+Only experiment-owned copies are removed. Source input is never edited.
+
+Results compare recovery-only costs and logical KB/DB/KG content fingerprints.
+Optional downstream probes run after each successful recovery. Fixture probes
+measure retrieval; live probes invoke the agent. Live extraction is potentially
+nondeterministic, so a fingerprint difference is diagnostic, not sufficient proof
+of incorrect recovery. Run downstream gold tasks and repeated trials. To measure
+crash positions at 25/50/75%, select source counts from the supported inventory
+and run each in a fresh output directory. These process-exit tests do not claim
+power-loss durability, distributed recovery, or exactly-once external writes.