ship: v0.2 P2 research engine + trust loop (#11)

2026-09-27 11:48:27 +00:00
parent 968eeb9102
commit 558f21a851
24 changed files with 1938 additions and 68 deletions
@@ -1,81 +1,77 @@
-# PLAN — Milestone v0.2 Phase 1: Methodology Framework + Adapter Layer
+# PLAN — Milestone v0.2 Phase 2: Research Engine + Trust Loop

 ---
-phase: 1
+phase: 2
 milestone: v0.2
 status: planned
 must_haves:
-  - "backend/ uv project scaffolds with one command (uv sync) and gates green from the first commit (ruff, mypy strict, pytest)"
-  - "12-dial typed schema with safe defaults resolves; every methodology = named dial bundle + planner/writer templates + output shape; registry resolves by slug (rapid-answer, deep-report)"
-  - "Resolved dial values persist to run config and are auditable (REQ-04) — schema-level contract proven by test"
-  - "Adapter contracts (model/search/storage/memory/secrets) typed; local implementations + fake model/search adapters enable zero-API-key engine tests (REQ-31)"
-  - "Probe/warning CONTRACTS typed and proven with fakes: capability-probe result (context cap min(model claim, 8192), embedding batch 8) and structured-output warning surfaces [REQ-07/REQ-12 contracts; LIVE endpoint probing executes in P2 per grill G-001]"
-  - "Model adapter: Ollama cloud-first with keyless localhost fallback (REQ-08); search: Tavily default + DDG keyless fallback (REQ-09); storage: local FS Markdown/JSON (REQ-10); memory stubbed off, secrets env-backed (REQ-11)"
-  - "Structured-output weakness warning CONTRACT (REQ-12): typed warning surface + graceful-degrade path proven with fakes"
+  - "Asyncio state machine walks PLANNING → AWAITING_APPROVAL → SEARCHING → EXTRACTING → REFLECTING → (depth loop) → SYNTHESIZING → VERIFYING → DONE with typed transitions; pause/cancel/resume survive process restarts via durable JSON state (atomic rewrite per transition + cursor)"
+  - "Plan = sub-questions + source universe + outline; run halts at AWAITING_APPROVAL until approved (fake-model plan proven end-to-end)"
+  - "Typed event bus → trace file per run + SSE-ready event stream (query_fired, source_visited, passage_extracted, state_changed, run_done, run_error)"
+  - "Evidence store: visited pages with stable passage IDs; passage extraction; writer context = stored passages only (visited-only constraint by construction)"
+  - "Trust loop: post-run URL liveness pass (live/stale/dead); claim-to-passage quote anchoring (normalize + exact short-circuit + SequenceMatcher ≥ 0.85, weak_match flags); citation ledger JSON per run; exportable verification checklist; confidence labels (single-source vs corroborated)"
+  - "SSRF guard at the fetch boundary (private/loopback/link-local + scheme allowlist) before the engine consumes search results (verify P1 flag, now mandatory)"
+  - "Run-id generator consolidated to the storage adapter's (reduced alphabet)"
+  - "Live capability probing against real endpoints when keys exist (REQ-07/12 complete) — probes surfaced as typed results, never crash the run"
+  - "Zero-API-key: every engine + trust test runs on fakes"
 personas:
-  - data-engineer
  - backend-engineer
+  - data-engineer
+  - security-engineer
  - ci-lead-developer
-requirements: [REQ-01, REQ-02, REQ-03, REQ-04, REQ-05, REQ-06, REQ-08, REQ-09, REQ-10, REQ-11, REQ-31]
+requirements: [REQ-07, REQ-12, REQ-13, REQ-14, REQ-15, REQ-16, REQ-18, REQ-19, REQ-20, REQ-21, REQ-22, REQ-23, REQ-24]
 ---

 ## User-Facing Surface

-No new UI in P1 — the frontend is untouched (mock-backed, still v0.1-complete). User-facing surface for P1 is the **backend package itself**: `uv sync && uv run uvicorn researchx.api.app:app` boots a healthless dev server skeleton (routes land in P3), and the methodology registry + adapters are exercisable from tests and a documented README quickstart section. The deliverable P1 surface ships to the user in P3 when routes + the client swap go live.
+Still no UI change in P2 (API + client swap land in P3). The P2 surface is the engine itself, exercisable via tests and a documented `uv run python -m researchx.engine.demo` happy-path script (fake adapters): question → plan → approve → search → extract → reflect → synthesize → verify → DONE, artifacts written to a temp runs dir (state.json, trace.jsonl, report.md, ledger.json, checklist.md).

-## Happy Path (P1 — test/README script, written before execute)
+## Happy Path (P2 — test + demo script, written before execute)

-1. `cd backend && uv sync` — installs Python 3.12 + deps, editable researchx, dev tools
-2. `uv run pytest` — full suite green with zero API keys (fake adapters drive everything, REQ-31)
-3. `uv run python -c "from researchx.methodologies import registry; m = registry.get('deep-report'); print(m.dials.depth)"` → `3` — registry resolves by slug, dial bundle typed
-4. `uv run python -c "from researchx.config.dials import Dials; print(Dials().model_dump())"` → all 12 dials with safe defaults
-5. `uv run ruff check src tests && uv run mypy src` → clean
-6. Frontend gates still green (untouched): `corepack pnpm typecheck && corepack pnpm test` from frontend/
+1. `cd backend && uv sync && uv run pytest` — all green, zero API keys
+2. `uv run python -m researchx.engine.demo` — fake-adapter run walks every state, artifacts land in a temp dir
+3. `uv run python -c "..."` — load state.json: state DONE; ledger.json: citations with health + authority tier; checklist.md exportable
+4. Pause mid-depth-loop test: cancel task, restart engine from state file, run completes (REQ-15/32 mid-loop seed planted)
+5. Adversarial seed: fake writer emits a hallucinated URL → verifier flags it dead; quote mismatch → weak_match/mismatch flagged in ledger (REQ-32 seeds, full suite in P4)
+6. Frontend gates still green (untouched)

-## UX Acceptance Criteria (P1)
+## UX Acceptance Criteria (P2)

- One-command backend setup (uv sync) with zero manual steps beyond uv itself
- Every adapter contract is typed; fakes implement the full interface — no stub methods, no TODOs
- No API key is ever required to run the test suite; no secret value ever reaches a log line, error message, or fixture
- Degradation is honest: probe results and structured-output warnings are typed data, not console noise
+- Trust is visible by construction: every claim in report.md carries a citation marker; the ledger names every cited URL with visited_at, passages, health, authority tier
+- Degradation honest: liveness failures, weak matches, and fallback probes surface in typed data, never silent
+- No secret value ever reaches a trace, artifact, or error message
+- A restarted run resumes, never re-plans from scratch (unless depth cursor says so)

-## Tasks (waves — dependency-ordered, persona-assigned)
+## Tasks (waves — dependency-ordered)

-### Wave 1 — Schema core (data-engineer)
- **T-1.1** `backend/pyproject.toml` + src-layout skeleton: packages config, deps, dev group, ruff/mypy/pytest config (hatchling packages key, pythonpath src, asyncio_mode auto) [REQ-33 seed — tooling; full monorepo story completes in P3/P4]
- **T-1.2** Dial schema: `researchx/config/dials.py` — 12 typed dials (source universe, search routing, query strategy, breadth, depth, quality gates, citation rigor, verification passes, synthesis style, output structure, memory scope, checkpoints), each with safe default; pydantic v2 [REQ-01]
- **T-1.3** Methodology model + registry: `researchx/methodologies/` — methodology = named dial bundle + planner/writer prompt templates + output shape; registry resolves by slug; ships Rapid Answer (depth 1, breadth 1, memo, per-claim citations) + Deep Report (depth 3, breadth 2, executive report, full verification); behavior-difference contract documented + asserted (different query sets/depth/output shape) [REQ-02, REQ-03, REQ-05]
- **T-1.4** Run-config contract: resolved dial values snapshot model (persisted per run — storage wiring lands with the engine in P2; schema + round-trip test here) [REQ-04]
+### Wave 1 — Evidence store (data-engineer)
+- **T-2.1** Evidence models: `researchx/evidence/store.py` — Passage (stable ID: `p-<n>` sequence, url, quote, text), VisitedPage, EvidenceStore (add passage/page, get context bundle for writer: ONLY stored passages, by construction REQ-18); passage extraction from PageContent (paragraph split, min length, dedupe)
+- **T-2.2** Evidence tests: stable IDs, dedupe, visited-only context (a passage not stored can never appear in writer context)

-### Wave 2 — Adapter contracts + fakes (backend-engineer)
- **T-1.5** Adapter contract types: `researchx/adapters/` — model (OpenAI-compatible, planner/writer roles), search (Tavily/DDG), storage (local FS), memory (stub), secrets (env-file); all I/O behind interfaces; httpx confined to adapters [REQ-06, REQ-10, REQ-11]
- **T-1.6** Fake adapters: FakeModelAdapter (canned planner/writer), FakeSearchAdapter (fixture corpus) — full interface implementations; zero API keys [REQ-31]
- **T-1.7** Storage adapter: local FS write of run artifacts (Markdown/JSON under user-chosen dir); run ID generation (8-char slug) [REQ-10, D-010]
- **T-1.8** Secrets adapter: env-file resolution (.ciagent/.env.secrets + .env scopes per config.json); never logs values [REQ-11]
+### Wave 2 — Engine (backend-engineer)
+- **T-2.3** State machine core: `researchx/engine/states.py` (enum + legal transitions table) + `run_state.py` (RunState pydantic: state, cursor, plan, counters; JSON round-trip) — persisted via storage adapter atomically after every transition (REQ-13, REQ-15)
+- **T-2.4** Event bus: `researchx/engine/events.py` — typed pydantic events (discriminated union on type), EventBus with subscribe/publish, trace writer (append-only .jsonl per run) (REQ-16)
+- **T-2.5** RunActor: `researchx/engine/actor.py` — one actor per run; walks states; planner call (model adapter, plan = sub-questions + source universe + outline) → AWAITING_APPROVAL halt → approved → SEARCHING (search adapter per dials; query_fired events) → EXTRACTING (fetch + evidence store + SSRF guard; source_visited/passage_extracted events) → REFLECTING depth loop (planner; depth from dials; cursor checkpoint) → SYNTHESIZING (writer; context = evidence bundle ONLY; per-claim markers) → VERIFYING (trust loop) → DONE (persist report + ledger + checklist) (REQ-13, REQ-14, REQ-16, REQ-18)
+- **T-2.6** Pause/cancel/resume: asyncio.Event pause gate; cancel persists terminal state; resume from state file re-enters at cursor; CancelledError never swallowed (re-raised after shielded persist) (REQ-15)
+- **T-2.7** Engine tests on fakes: full walk, AWAITING_APPROVAL halt + approve, pause/resume mid-depth-loop, cancel, restart-from-state-file, event ordering, visited-only writer context, zero API keys

-### Wave 3 — Real adapters + probe contracts (backend-engineer)
- **T-1.9** Model adapter: OpenAI-compatible chat completions against OLLAMA_BASE_URL (default https://ollama.com/v1, Bearer OLLAMA_API_KEY) with keyless http://localhost:11434/v1 fallback on auth/connection failure; planner/writer roles [REQ-06, REQ-08]
- **T-1.10** Capability-probe contract + implementation: GET /models + structured-output smoke test; conservative caps: context min(claim, 8192), embedding batch 8; typed probe result with overrides; proven with fakes here — LIVE probing against real endpoints executes in P2 [REQ-07 contract, G-001]
- **T-1.11** Structured-output warning contract: smoke-test failure → typed warning surfaced for UI; degrade to prompt-embedded JSON contract — never cryptic failure [REQ-12 contract, G-001]
- **T-1.12** Search adapter: Tavily (POST api.tavily.com/search, key from secrets) default; DDG keyless fallback (html.duckduckgo.com/html conservative scrape, max 8, one backoff, typed degradation warning — degraded-mode contract per G-008); registry selects by key presence [REQ-09, D-012]
+### Wave 3 — Trust loop (backend-engineer + security-engineer)
+- **T-2.8** Liveness pass: `researchx/trust/liveness.py` — GET with redirect-follow (10s, size cap from P1 flag), live/stale/dead classification; runs post-run over cited URLs (REQ-19)
+- **T-2.9** Quote anchoring: `researchx/trust/anchoring.py` — normalize (NFC/casefold/whitespace collapse) + exact-substring short-circuit + SequenceMatcher ≥ 0.85 per-passage; near-threshold (0.80–0.90) flagged weak_match; mismatches flagged never dropped (REQ-20)
+- **T-2.10** Citation ledger + checklist: `researchx/trust/ledger.py` — per-run ledger JSON (url, visited_at, passages, claims supported, health, authority tier); verification checklist export (markdown); confidence labels single-source vs corroborated (REQ-21, REQ-22, REQ-24)
+- **T-2.11** SSRF guard: `researchx/trust/url_guard.py` (or adapters/fetch guard) — scheme http/https only; resolve host; refuse private/loopback/link-local/0.0.0.0/IPv6 ULA; wired into every fetch path (search adapter fetch + engine extract) (P1 P1-flag, security persona)
+- **T-2.12** Trust tests: liveness classification (MockTransport), anchoring thresholds + weak_match + shared-corpus sentence, ledger schema, checklist export, confidence labels, SSRF refusals (private IP, link-local metadata 169.254.169.254, non-http scheme)

-### Wave 4 — Integration proofs (ci-lead-developer)
- **T-1.13** Cross-cutting test: registry → dial resolution → fake-model plan fixture → dial snapshot round-trip; methodology behavior-difference assertion (Rapid Answer vs Deep Report query sets/depth/output shape) [REQ-05]
- **T-1.14** README + VERSIONING updates: backend quickstart (uv sync/uv run uvicorn), adapter config table (env vars, no values), monorepo command map [REQ-33 partial]
+### Wave 4 — Live probing + integration (backend-engineer + ci-lead-developer)
+- **T-2.13** Live probe wiring: engine run pre-flight probes model adapter when OLLAMA_API_KEY or local endpoint reachable — probe result + structured-output warning stored in run state + trace; probe failure degrades honestly, never crashes the run (REQ-07, REQ-12 complete)
+- **T-2.14** Demo script + integration proofs: researchx.engine.demo (fake adapters, temp dir); end-to-end integration test pinning artifacts (state DONE, report with markers, ledger health, checklist); run-id consolidation (storage adapter's generator everywhere)
+- **T-2.15** README: engine architecture section (state machine diagram in text, artifact layout runs/<id>/)

-## Out of scope for P1 (later phases)
- Engine state machine, evidence store, trust loop (P2); API routes, SSE, OpenAPI client swap (P3); adversarial/E2E suite (P4)
-
-## Milestone-level constraints promoted from grill G-001..G-010 (binding, absorbed by phase plans)
- **Live-mode surface (G-002)**: v0.2 live mode = core research loop only (runs, methodologies, lifecycle, SSE, config status). Library persistence, Settings persistence, Suggestions, Refine, graph/inbox, profile stay mock-bound until v0.3+ — owner ratifies at the P5 gate (escalation G-009)
- **ApiClient extension (G-003)**: P3 adds pauseRun/resumeRun as the SINGLE planned interface extension; client.ts endpoint-mapped surfaces beyond the core loop stay mock-backed
- **Liveness gate (G-004)**: REQ-23 ≥99% is a runtime per-run export gate + an opt-in keyed live-URL test tier — never a zero-key CI assertion on the fixture corpus
- **OpenAPI snapshot discipline (G-005)**: P3 must commit an openapi.json snapshot; regen only from the snapshot; schema-drift gate; stable operation-ids
- **P2 split lever (G-006)**: if P2 slips beyond one extension cycle it may ship as two patches (P2a engine/SSE, P2b trust/probe); P3 swap endpoint plan drafted at P2 start
- **Cost counters (G-007)**: per-run advisory request/token/search counters surface via GET /api/config (P3)
- **Contingency lever (G-010)**: if P2/P4 slip beyond one extension cycle, Deep Report defers to v0.3 and the MVP ships Rapid Answer end-to-end — owner sign-off required to pull this lever
+## Out of scope for P2
+- API routes/SSE transport/OpenAPI client (P3); adversarial suite expansion + E2E (P4); ETA/cost heuristics (P3, REQ-25 adjacent)

 ## Verification hooks (consumed by VERIFY)
- Gates: `uv run ruff check src tests`, `uv run mypy src`, `uv run pytest` (backend); `corepack pnpm typecheck/lint/test/build` (frontend, must stay green)
- REQ coverage assert: REQ-01..06, 08..11, 31 each mapped to ≥1 test or contract proof
- Security: no secret values in code/fixtures/logs; SSRF not yet applicable (no page fetching until P2 evidence store)
+- Gates: backend ruff/mypy strict/pytest; frontend untouched-green
+- REQ coverage: REQ-07, 12, 13..24 each mapped to ≥1 test or artifact proof
+- Security: SSRF refusals test-enforced; no value leaks in traces/artifacts; CancelledError discipline
+- G-006 note: if this phase slips beyond one extension cycle, split into P2a (engine/SSE artifacts) + P2b (trust) patches — requires orchestrator decision commit
@@ -28,6 +28,26 @@ Three clicks from cold start to initiated research.
 | backend (uv, Python 3.12) | `uv sync` | `uv run uvicorn researchx.api.app:app --reload --port 8000` | `uv run ruff check src tests` · `uv run mypy src` · `uv run pytest` |
 | frontend (pnpm 12.6) | `corepack pnpm install` | `corepack pnpm dev` | `corepack pnpm typecheck` · `corepack pnpm lint` · `corepack pnpm test` · `corepack pnpm build` |

+## Engine architecture (v0.2 P2)
+
+The research engine is a hand-rolled asyncio state machine:
+
+```
+PLANNING → AWAITING_APPROVAL → SEARCHING → EXTRACTING → REFLECTING ─┐
+                (halt until             ↑ depth loop (dials.depth)  │
+                 approved)              └────────────────────────────┘
+→ SYNTHESIZING → VERIFYING → DONE  (+ CANCELLED / FAILED terminals)
+```
+
+- Durable state: `runs/<id>/state.json` rewritten atomically after every transition; resume re-enters at the recorded cursor with the persisted plan
+- Event bus: typed events (`run_started`, `query_fired`, `source_visited`, `passage_extracted`, `state_changed`, `run_done`, `run_error`) — trace per run
+- Visited-only constraint: the writer's context is the evidence store and nothing else
+- Trust loop: SSRF-guarded liveness (live/stale/dead), quote anchoring (normalize + fuzzy ≥ 0.85, weak matches flagged), citation ledger JSON, exportable verification checklist, single-source vs corroborated confidence labels
+
+Run artifacts per run: `state.json`, `report.md`, `ledger.json`, `checklist.md`.
+
+Try it on fakes (no keys): `uv run python -m researchx.engine.demo`
+
 ## Adapter configuration (v0.2)

 All keys resolve from env files only (`.env`, `.env.secrets`, `.env.*` — never the shell environment, never git):
@@ -7,7 +7,8 @@ import httpx

 from researchx.adapters.secrets import SecretsProvider
 from researchx.adapters.tavily import TavilySearchAdapter, extract_text, extract_title
-from researchx.adapters.types import PageContent, SearchResult
+from researchx.adapters.types import AdapterError, PageContent, SearchResult
+from researchx.adapters.url_guard import url_allowed

 _DDG_URL = "https://html.duckduckgo.com/html"
 _SEARCH_TIMEOUT = 10.0
@@ -92,6 +93,8 @@ class DuckDuckGoSearchAdapter:
        return re.sub(r"\s+", " ", text).strip()

    async def fetch(self, url: str) -> PageContent:
+        if not url_allowed(url):
+            raise AdapterError("fetch refused: url not allowed (private or non-http scheme)")
        async with httpx.AsyncClient(
            timeout=_FETCH_TIMEOUT,
            follow_redirects=True,
@@ -58,4 +58,16 @@ class LocalStorageAdapter:
        try:
            return target.read_text(encoding="utf-8")
        except FileNotFoundError as exc:
-            raise AdapterError(f"artifact not found: runs/{run_id}/{name}") from exc
+            raise AdapterError(f"artifact not found: runs/{run_id}/{name}") from exc
+
+    async def append_run_artifact(self, run_id: str, name: str, content: str) -> Path:
+        _validate_artifact_name(name)
+        target = self.run_dir(run_id) / name
+        target.parent.mkdir(parents=True, exist_ok=True)
+
+        def _append() -> None:
+            with open(target, "a", encoding="utf-8") as handle:
+                handle.write(content)
+
+        await asyncio.to_thread(_append)
+        return target
@@ -5,6 +5,7 @@ import httpx

 from researchx.adapters.secrets import SecretsProvider
 from researchx.adapters.types import AdapterError, PageContent, SearchResult
+from researchx.adapters.url_guard import url_allowed

 _TAVILY_URL = "https://api.tavily.com/search"
 _SEARCH_TIMEOUT = 10.0
@@ -55,6 +56,8 @@ class TavilySearchAdapter:
        ][:max_results]

    async def fetch(self, url: str) -> PageContent:
+        if not url_allowed(url):
+            raise AdapterError("fetch refused: url not allowed (private or non-http scheme)")
        async with httpx.AsyncClient(
            timeout=_FETCH_TIMEOUT,
            follow_redirects=True,
@@ -0,0 +1,44 @@
+import socket
+from ipaddress import IPv4Address, IPv6Address
+from urllib.parse import urlparse
+
+
+def _host_is_private(hostname: str) -> bool:
+    try:
+        addr_info = socket.getaddrinfo(hostname, None)
+    except (socket.gaierror, UnicodeError):
+        return True
+    for family, _, _, _, sockaddr in addr_info:
+        try:
+            if family == socket.AF_INET6:
+                ip: IPv4Address | IPv6Address = IPv6Address(sockaddr[0])
+            else:
+                ip = IPv4Address(sockaddr[0])
+        except ValueError:
+            return True
+        if (
+            ip.is_private
+            or ip.is_loopback
+            or ip.is_link_local
+            or ip.is_reserved
+            or ip.is_multicast
+            or ip.is_unspecified
+        ):
+            return True
+    return False
+
+
+def url_allowed(url: str) -> bool:
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        return False
+    if not parsed.hostname:
+        return False
+    return not _host_is_private(parsed.hostname)
+
+
+class FetchRefusedError(Exception):
+    pass
+
+
+__all__ = ["FetchRefusedError", "url_allowed"]
@@ -0,0 +1,348 @@
+import asyncio
+from datetime import UTC, datetime
+from typing import Any
+
+import httpx
+
+from researchx.adapters.local_storage import LocalStorageAdapter
+from researchx.adapters.model import ModelAdapter
+from researchx.adapters.search import SearchAdapter
+from researchx.adapters.types import AdapterError, ModelMessage
+from researchx.adapters.url_guard import url_allowed
+from researchx.config.dials import Dials
+from researchx.config.run_config import RunConfig, resolve_dials
+from researchx.engine.events import EventBus, RunEvent
+from researchx.engine.run_state import Plan, RunState
+from researchx.engine.states import RunStateName
+from researchx.evidence.store import EvidenceStore
+from researchx.methodologies import MethodologyRegistry
+from researchx.trust.ledger import build_ledger, export_checklist
+from researchx.trust.liveness import liveness_report
+
+
+class EngineError(Exception):
+    pass
+
+
+class RunActor:
+    def __init__(
+        self,
+        *,
+        run_config: RunConfig,
+        model: ModelAdapter,
+        search: SearchAdapter,
+        storage: LocalStorageAdapter,
+        registry: MethodologyRegistry,
+        event_bus: EventBus,
+    ) -> None:
+        self._config = run_config
+        self._model = model
+        self._search = search
+        self._storage = storage
+        self._registry = registry
+        self._bus = event_bus
+        self._store = EvidenceStore()
+        self.state = RunState(
+            run_id=run_config.run_id,
+            question=run_config.question,
+            context=run_config.context,
+            methodology_slug=run_config.methodology_slug,
+        )
+        methodology = registry.get(run_config.methodology_slug)
+        self._dials: Dials = run_config.resolved_dials
+        self.state.depth_target = self._dials.depth
+        self._methodology = methodology
+        self._pause = asyncio.Event()
+        self._pause.set()
+        self._cancelled = False
+        self._approved = asyncio.Event()
+
+    async def _publish(self, event_type: str, data: dict[str, str] | None = None) -> None:
+        event = RunEvent(
+            type=event_type,
+            run_id=self._config.run_id,
+            data=data or {},
+        )
+        await self._bus.publish(event)
+        try:
+            await self._storage.append_run_artifact(
+                self._config.run_id,
+                "trace.jsonl",
+                event.model_dump_json() + "\n",
+            )
+        except AdapterError:
+            pass
+
+    async def _persist(self) -> None:
+        content = self._snapshot()
+        await self._storage.write_run_artifact(self._config.run_id, "state.json", content)
+
+    def _snapshot(self) -> str:
+        return self.state.model_dump_json()
+
+    async def _checkpoint(self, target: RunStateName) -> None:
+        if self.state.state != target:
+            self.state.transition(target)
+        await self._persist()
+
+    async def _wait_if_paused(self) -> None:
+        await self._pause.wait()
+
+    def pause(self) -> None:
+        self._pause.clear()
+
+    def resume(self) -> None:
+        self._pause.set()
+
+    def cancel(self) -> None:
+        self._cancelled = True
+
+    async def approve(self, plan: Plan | None = None) -> None:
+        if plan is not None:
+            self.state.plan = plan
+        self._approved.set()
+
+    async def run(self) -> RunState:
+        await self._publish("run_started", {"question": self._config.question})
+        try:
+            await self._execute()
+        except asyncio.CancelledError:
+            await asyncio.shield(self._persist_cancelled())
+            raise
+        except AdapterError as exc:
+            await self._fail(str(exc))
+            raise EngineError(str(exc)) from exc
+        return self.state
+
+    async def attach_probe(self, probe_data: dict[str, Any]) -> None:
+        self.state.probe = probe_data
+        await self._persist()
+        await self._publish("probe_attached", {"endpoint": str(probe_data.get("endpoint", ""))})
+
+    async def _persist_cancelled(self) -> None:
+        self.state.state = RunStateName.CANCELLED
+        await self._persist()
+
+    async def _fail(self, message: str) -> None:
+        await self._publish("run_error", {"message": message})
+        self.state.state = RunStateName.FAILED
+        await asyncio.shield(self._persist())
+
+    async def _execute(self) -> None:
+        current = self.state.state
+        if current == RunStateName.PLANNING:
+            await self._planning_phase()
+        elif current == RunStateName.AWAITING_APPROVAL:
+            pass
+        elif current in (
+            RunStateName.SEARCHING,
+            RunStateName.EXTRACTING,
+            RunStateName.REFLECTING,
+            RunStateName.SYNTHESIZING,
+        ):
+            await self._research_phase()
+            await self._verify_phase()
+            return
+        elif current == RunStateName.VERIFYING:
+            await self._verify_phase()
+            return
+        else:
+            raise EngineError(f"cannot resume from state {current}")
+        await self._approval_phase()
+        await self._research_phase()
+        await self._verify_phase()
+
+    async def _planning_phase(self) -> None:
+        await self._checkpoint(RunStateName.PLANNING)
+        await self._wait_if_paused()
+        if self._cancelled:
+            await self._persist_cancelled()
+            raise asyncio.CancelledError
+        prompt = self._methodology.planner_prompt.format(
+            question=self._config.question, context=self._config.context
+        )
+        result = await self._model.chat(
+            [ModelMessage(role="user", content=prompt)],
+            model=self._model.roles().planner,
+            json_mode=True,
+        )
+        plan = self._parse_plan(result.content)
+        self.state.plan = plan
+        await self._publish("state_changed", {"state": "AWAITING_APPROVAL"})
+
+    def _parse_plan(self, raw: str) -> Plan:
+        import json
+
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise AdapterError("planner returned non-JSON plan") from exc
+        return Plan(
+            sub_questions=[str(q) for q in data.get("sub_questions", [])],
+            source_universe=[str(u) for u in data.get("source_universe", [])],
+            outline=[str(o) for o in data.get("outline", [])],
+        )
+
+    async def _approval_phase(self) -> None:
+        await self._checkpoint(RunStateName.AWAITING_APPROVAL)
+        while not self._approved.is_set():
+            if self._cancelled:
+                await self._persist_cancelled()
+                raise asyncio.CancelledError
+            try:
+                await asyncio.wait_for(self._approved.wait(), timeout=0.1)
+            except TimeoutError:
+                continue
+        await self._checkpoint(RunStateName.SEARCHING)
+
+    async def _research_phase(self) -> None:
+        while True:
+            await self._check_cancelled()
+            await self._searching_step()
+            await self._extracting_step()
+            if self.state.depth_cursor + 1 >= self.state.depth_target:
+                break
+            await self._reflecting_step()
+            self.state.depth_cursor += 1
+            await self._persist()
+        await self._synthesizing_step()
+
+    async def _check_cancelled(self) -> None:
+        if self._cancelled:
+            await self._persist_cancelled()
+            raise asyncio.CancelledError
+
+    async def _searching_step(self) -> None:
+        if not self.state.plan or not self.state.plan.sub_questions:
+            await self._checkpoint(RunStateName.SEARCHING)
+            return
+        await self._checkpoint(RunStateName.SEARCHING)
+        await self._wait_if_paused()
+        for sub_question in self.state.plan.sub_questions[: self._dials.breadth]:
+            await self._publish("query_fired", {"query": sub_question})
+            self.state.bump("queries")
+            results = await self._search.search(sub_question, max_results=3)
+            for result in results:
+                await self._publish(
+                    "source_found",
+                    {"url": result.url, "title": result.title},
+                )
+
+    async def _extracting_step(self) -> None:
+        await self._checkpoint(RunStateName.EXTRACTING)
+        await self._wait_if_paused()
+        if not self.state.plan:
+            return
+        for sub_question in self.state.plan.sub_questions[: self._dials.breadth]:
+            results = await self._search.search(sub_question, max_results=3)
+            for result in results:
+                if not url_allowed(result.url):
+                    await self._publish("source_refused", {"url": result.url})
+                    continue
+                try:
+                    page = await self._search.fetch(result.url)
+                except (AdapterError, httpx.HTTPError):
+                    continue
+                self._store.add_page(page.url, page.title, page.text)
+                await self._publish(
+                    "source_visited",
+                    {"url": page.url, "title": page.title},
+                )
+                self.state.bump("pages")
+                for passage in self._store.passages_for_url(page.url):
+                    await self._publish(
+                        "passage_extracted",
+                        {"passage_id": passage.passage_id, "url": passage.url},
+                    )
+                    self.state.bump("passages")
+
+    async def _reflecting_step(self) -> None:
+        await self._checkpoint(RunStateName.REFLECTING)
+        await self._wait_if_paused()
+        prompt = self._methodology.planner_prompt.format(
+            question=self._config.question, context=self._config.context
+        )
+        result = await self._model.chat(
+            [ModelMessage(role="user", content=prompt)],
+            model=self._model.roles().planner,
+            json_mode=True,
+        )
+        if self.state.plan:
+            refined = self._parse_plan(result.content)
+            self.state.plan = refined
+
+    async def _synthesizing_step(self) -> None:
+        await self._checkpoint(RunStateName.SYNTHESIZING)
+        await self._wait_if_paused()
+        await self._check_cancelled()
+        context_block = "\n\n".join(
+            f"[{p.passage_id}] ({p.url}) {p.text}" for p in self._store.writer_context()
+        )
+        prompt = self._methodology.render_writer_prompt(
+            question=self._config.question,
+            context=self._config.context or "(none)",
+            plan=self.state.plan.model_dump_json() if self.state.plan else "{}",
+            evidence=context_block or "(no evidence collected)",
+        )
+        result = await self._model.chat(
+            [ModelMessage(role="user", content=prompt)],
+            model=self._model.roles().writer,
+        )
+        report = result.content
+        await self._storage.write_run_artifact(self._config.run_id, "report.md", report)
+        self.state.bump("writer_calls")
+
+    async def _verify_phase(self) -> None:
+        await self._checkpoint(RunStateName.VERIFYING)
+        report = await self._storage.read_run_artifact(self._config.run_id, "report.md")
+        liveness = await liveness_report(self._cited_urls(report))
+        ledger = build_ledger(
+            run_id=self._config.run_id,
+            report_text=report,
+            evidence=self._store,
+            liveness=liveness,
+            visited_at=datetime.now(UTC),
+        )
+        await self._storage.write_run_artifact(
+            self._config.run_id, "ledger.json", ledger.model_dump_json(indent=2)
+        )
+        checklist = export_checklist(ledger)
+        await self._storage.write_run_artifact(self._config.run_id, "checklist.md", checklist)
+        await self._checkpoint(RunStateName.DONE)
+        await self._publish("run_done", {"run_id": self._config.run_id})
+
+    def _cited_urls(self, report_text: str) -> list[str]:
+        import re
+
+        pattern = re.compile(r"\[p-(\d+)\]")
+        cited: list[str] = []
+        for match in pattern.finditer(report_text):
+            passage = self._store.get(f"p-{match.group(1)}")
+            if passage and passage.url not in cited:
+                cited.append(passage.url)
+        return cited
+
+
+def make_run_config(
+    *,
+    question: str,
+    context: str,
+    methodology_slug: str,
+    overrides: dict[str, object],
+    registry: MethodologyRegistry,
+    run_id: str | None = None,
+) -> tuple[str, RunConfig]:
+    from researchx.adapters.local_storage import generate_run_id
+
+    rid = run_id or generate_run_id()
+    dials = resolve_dials(methodology_slug, overrides, registry)
+    config = RunConfig(
+        run_id=rid,
+        methodology_slug=methodology_slug,
+        question=question,
+        context=context,
+        resolved_dials=dials,
+        overrides=dict(overrides),
+        created_at=datetime.now(UTC),
+    )
+    return rid, config
@@ -0,0 +1,68 @@
+import asyncio
+import tempfile
+from pathlib import Path
+
+from researchx.adapters.fakes import CORPUS, FakeModelAdapter, FakeSearchAdapter
+from researchx.adapters.local_storage import LocalStorageAdapter
+from researchx.adapters.types import ChatResult
+from researchx.engine.actor import RunActor, make_run_config
+from researchx.engine.events import EventBus
+from researchx.methodologies import registry
+
+
+async def main() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        base = Path(tmp)
+        run_id, config = make_run_config(
+            question="How do verification pipelines keep citations honest?",
+            context="",
+            methodology_slug="rapid-answer",
+            overrides={},
+            registry=registry,
+        )
+        plan = (
+            '{"sub_questions": ["verification passes", "quote anchoring"], '
+            '"source_universe": ["web"], "outline": ["intro", "findings"]}'
+        )
+        writer = (
+            "# Research Report\n\n"
+            "The verification pass re-anchored every quoted claim [p-1] "
+            "before the report shipped [p-2].\n"
+        )
+        model = FakeModelAdapter(
+            canned=[
+                ChatResult(content=plan, model="fake-planner"),
+                ChatResult(content=writer, model="fake-writer"),
+            ]
+        )
+        search = FakeSearchAdapter(CORPUS)
+        storage = LocalStorageAdapter(base_dir=base)
+        bus = EventBus()
+        actor = RunActor(
+            run_config=config, model=model, search=search, storage=storage, registry=registry,
+            event_bus=bus,
+        )
+        task = asyncio.create_task(actor.run())
+        while actor.state.state.value == "PLANNING":
+            await asyncio.sleep(0.01)
+        await actor.approve()
+        await task
+
+        state = (storage.run_dir(run_id) / "state.json").read_text()
+        report = (storage.run_dir(run_id) / "report.md").read_text()
+        ledger = (storage.run_dir(run_id) / "ledger.json").read_text()
+        checklist = (storage.run_dir(run_id) / "checklist.md").read_text()
+        print(f"run {run_id} → {actor.state.state.value}")
+        print(f"artifacts under {storage.run_dir(run_id)}:")
+        for name, content in [
+            ("state.json", state),
+            ("report.md", report),
+            ("ledger.json", ledger),
+            ("checklist.md", checklist),
+        ]:
+            print(f"--- {name} ({len(content)} chars) ---")
+        print(f"events: {len(bus.replay())}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
@@ -0,0 +1,41 @@
+from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class RunEvent(BaseModel):
+    type: str
+    run_id: str
+    at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    data: dict[str, str] = {}
+
+
+class EventBus:
+    def __init__(self) -> None:
+        self._subscribers: list[Callable[[RunEvent], Awaitable[None]]] = []
+        self._log: list[RunEvent] = []
+
+    def subscribe(self, handler: Callable[[RunEvent], Awaitable[None]]) -> None:
+        self._subscribers.append(handler)
+
+    async def publish(self, event: RunEvent) -> None:
+        self._log.append(event)
+        for handler in self._subscribers:
+            await handler(event)
+
+    def replay(self) -> list[RunEvent]:
+        return list(self._log)
+
+
+class TraceWriter:
+    def __init__(self, emit: Callable[[str], None]) -> None:
+        self._emit = emit
+
+    async def record(self, event: RunEvent) -> None:
+        self._emit(event.model_dump_json())
+
+
+def event_type_of(event: RunEvent) -> Literal["run_started", "other"]:
+    return "run_started" if event.type == "run_started" else "other"
@@ -0,0 +1,36 @@
+from datetime import UTC, datetime
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from researchx.engine.states import RunStateName, can_transition
+
+
+class Plan(BaseModel):
+    sub_questions: list[str] = Field(default_factory=list)
+    source_universe: list[str] = Field(default_factory=list)
+    outline: list[str] = Field(default_factory=list)
+
+
+class RunState(BaseModel):
+    run_id: str
+    question: str
+    context: str = ""
+    methodology_slug: str
+    state: RunStateName = RunStateName.PLANNING
+    plan: Plan | None = None
+    depth_cursor: int = 0
+    depth_target: int = 1
+    counters: dict[str, int] = Field(default_factory=dict)
+    probe: dict[str, Any] | None = None
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+
+    def transition(self, target: RunStateName) -> None:
+        if not can_transition(self.state, target):
+            raise ValueError(f"illegal transition: {self.state} -> {target}")
+        self.state = target
+        self.updated_at = datetime.now(UTC)
+
+    def bump(self, key: str) -> None:
+        self.counters[key] = self.counters.get(key, 0) + 1
@@ -0,0 +1,65 @@
+from enum import StrEnum
+
+
+class RunStateName(StrEnum):
+    PLANNING = "PLANNING"
+    AWAITING_APPROVAL = "AWAITING_APPROVAL"
+    SEARCHING = "SEARCHING"
+    EXTRACTING = "EXTRACTING"
+    REFLECTING = "REFLECTING"
+    SYNTHESIZING = "SYNTHESIZING"
+    VERIFYING = "VERIFYING"
+    DONE = "DONE"
+    CANCELLED = "CANCELLED"
+    FAILED = "FAILED"
+
+
+_LEGAL_TRANSITIONS: dict[RunStateName, set[RunStateName]] = {
+    RunStateName.PLANNING: {RunStateName.AWAITING_APPROVAL, RunStateName.FAILED},
+    RunStateName.AWAITING_APPROVAL: {
+        RunStateName.SEARCHING,
+        RunStateName.CANCELLED,
+        RunStateName.FAILED,
+    },
+    RunStateName.SEARCHING: {
+        RunStateName.EXTRACTING,
+        RunStateName.SYNTHESIZING,
+        RunStateName.CANCELLED,
+        RunStateName.FAILED,
+    },
+    RunStateName.EXTRACTING: {
+        RunStateName.REFLECTING,
+        RunStateName.SYNTHESIZING,
+        RunStateName.CANCELLED,
+        RunStateName.FAILED,
+    },
+    RunStateName.REFLECTING: {
+        RunStateName.SEARCHING,
+        RunStateName.SYNTHESIZING,
+        RunStateName.CANCELLED,
+        RunStateName.FAILED,
+    },
+    RunStateName.SYNTHESIZING: {
+        RunStateName.VERIFYING,
+        RunStateName.CANCELLED,
+        RunStateName.FAILED,
+    },
+    RunStateName.VERIFYING: {
+        RunStateName.DONE,
+        RunStateName.CANCELLED,
+        RunStateName.FAILED,
+    },
+    RunStateName.DONE: set(),
+    RunStateName.CANCELLED: set(),
+    RunStateName.FAILED: set(),
+}
+
+TERMINAL_STATES = {
+    RunStateName.DONE,
+    RunStateName.CANCELLED,
+    RunStateName.FAILED,
+}
+
+
+def can_transition(current: RunStateName, target: RunStateName) -> bool:
+    return target in _LEGAL_TRANSITIONS[current]
@@ -0,0 +1,70 @@
+from pydantic import BaseModel, Field
+
+
+class Passage(BaseModel):
+    passage_id: str
+    url: str
+    quote: str
+    text: str
+
+
+class VisitedPage(BaseModel):
+    url: str
+    title: str
+    passage_ids: list[str] = Field(default_factory=list)
+
+
+def extract_passages(text: str, *, min_chars: int = 120) -> list[str]:
+    paragraphs = [p.strip() for p in text.split("\n") if p.strip()]
+    return [p for p in paragraphs if len(p) >= min_chars]
+
+
+class EvidenceStore:
+    def __init__(self) -> None:
+        self._passages: dict[str, Passage] = {}
+        self._pages: dict[str, VisitedPage] = {}
+        self._seen_quotes: set[str] = set()
+        self._counter = 0
+
+    def add_page(self, url: str, title: str, text: str) -> VisitedPage:
+        page = self._pages.setdefault(url, VisitedPage(url=url, title=title))
+        for paragraph in extract_passages(text):
+            self.add_passage(url, paragraph)
+        return page
+
+    def add_passage(self, url: str, text: str) -> Passage:
+        self._counter += 1
+        passage_id = f"p-{self._counter}"
+        passage = Passage(passage_id=passage_id, url=url, quote=text[:200], text=text)
+        self._passages[passage_id] = passage
+        page = self._pages.get(url)
+        if page is None:
+            page = VisitedPage(url=url, title="")
+            self._pages[url] = page
+        page.passage_ids.append(passage_id)
+        self._seen_quotes.add(text[:200])
+        return passage
+
+    def has_seen(self, quote: str) -> bool:
+        return quote[:200] in self._seen_quotes
+
+    def get(self, passage_id: str) -> Passage | None:
+        return self._passages.get(passage_id)
+
+    def passages_for_url(self, url: str) -> list[Passage]:
+        page = self._pages.get(url)
+        if page is None:
+            return []
+        return [self._passages[pid] for pid in page.passage_ids if pid in self._passages]
+
+    def writer_context(self) -> list[Passage]:
+        return [self._passages[pid] for pid in sorted(self._passages, key=self._passage_order)]
+
+    def _passage_order(self, passage_id: str) -> int:
+        return int(passage_id.split("-")[1])
+
+    def pages(self) -> list[VisitedPage]:
+        return list(self._pages.values())
+
+    def __len__(self) -> int:
+        return len(self._passages)
@@ -11,11 +11,12 @@ _RAPID_PLANNER_PROMPT = (

 _RAPID_WRITER_PROMPT = (
    "You are a rapid memo writer. Answer the question directly in a short memo.\n"
-    "Cite a source per claim. Every cited quote must come verbatim from the provided "
-    "evidence passages.\n"
+    "Cite a source per claim using the passage markers [p-N]. Every cited quote must "
+    "come verbatim from the evidence passages listed below and from nowhere else.\n"
    "Question: {question}\n"
    "Context: {context}\n"
-    "Plan: {plan}"
+    "Plan: {plan}\n"
+    "Evidence passages:\n{evidence}"
 )

 _DEEP_PLANNER_PROMPT = (
@@ -30,11 +31,12 @@ _DEEP_WRITER_PROMPT = (
    "You are an executive report writer. Synthesize the evidence into a structured "
    "executive report following the plan: executive summary, findings, risks, and "
    "recommendations.\n"
-    "Cite a source per claim. Every cited quote must come verbatim from the provided "
-    "evidence passages.\n"
+    "Cite a source per claim using the passage markers [p-N]. Every cited quote must "
+    "come verbatim from the evidence passages listed below and from nowhere else.\n"
    "Question: {question}\n"
    "Context: {context}\n"
-    "Plan: {plan}"
+    "Plan: {plan}\n"
+    "Evidence passages:\n{evidence}"
 )

 _BUILTIN: dict[str, Methodology] = {
@@ -21,8 +21,12 @@ class Methodology(BaseModel, frozen=True):
    def render_planner_prompt(self, question: str, context: str) -> str:
        return self.planner_prompt.format(question=question, context=context)

-    def render_writer_prompt(self, question: str, context: str, plan: str) -> str:
-        return self.writer_prompt.format(question=question, context=context, plan=plan)
+    def render_writer_prompt(
+        self, question: str, context: str, plan: str, evidence: str
+    ) -> str:
+        return self.writer_prompt.format(
+            question=question, context=context, plan=plan, evidence=evidence
+        )


 __all__ = ["Methodology", "OutputShape"]
@@ -0,0 +1,60 @@
+import re
+import unicodedata
+from difflib import SequenceMatcher
+
+from pydantic import BaseModel
+
+from researchx.evidence.store import EvidenceStore, Passage
+
+ANCHOR_THRESHOLD = 0.85
+WEAK_MATCH_MIN = 0.80
+
+
+def normalize(text: str) -> str:
+    nfc = unicodedata.normalize("NFC", text)
+    collapsed = re.sub(r"\s+", " ", nfc)
+    return collapsed.casefold().strip()
+
+
+class AnchorVerdict(BaseModel):
+    quote: str
+    passage_id: str | None
+    ratio: float
+    matched: bool
+    weak: bool
+
+
+def anchor_quote(quote: str, passages: list[Passage]) -> AnchorVerdict:
+    normalized_quote = normalize(quote)
+    best_ratio = 0.0
+    best_id: str | None = None
+    for passage in passages:
+        normalized_passage = normalize(passage.text)
+        if normalized_quote and normalized_quote in normalized_passage:
+            return AnchorVerdict(
+                quote=quote, passage_id=passage.passage_id, ratio=1.0, matched=True, weak=False
+            )
+        ratio = SequenceMatcher(None, normalized_quote, normalized_passage).ratio()
+        if ratio > best_ratio:
+            best_ratio = ratio
+            best_id = passage.passage_id
+    matched = best_ratio >= ANCHOR_THRESHOLD
+    weak = WEAK_MATCH_MIN <= best_ratio < ANCHOR_THRESHOLD
+    return AnchorVerdict(
+        quote=quote,
+        passage_id=best_id if (matched or weak) else None,
+        ratio=round(best_ratio, 4),
+        matched=matched,
+        weak=weak,
+    )
+
+
+def extract_quotes_from_report(report_text: str, evidence: EvidenceStore) -> list[str]:
+    marker_re = re.compile(r"\[(p-\d+)\]")
+    quotes: list[str] = []
+    for match in marker_re.finditer(report_text):
+        passage_id = match.group(1)
+        passage = evidence.get(passage_id)
+        if passage:
+            quotes.append(passage.quote)
+    return quotes
@@ -0,0 +1,181 @@
+import re
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from researchx.evidence.store import EvidenceStore, Passage
+from researchx.trust.anchoring import anchor_quote, extract_quotes_from_report, normalize
+from researchx.trust.liveness import LivenessStatus
+
+_AUTHORITY_TIERS: dict[str, str] = {
+    "example.org": "reference",
+    "gov": "official",
+    "edu": "academic",
+    "org": "reference",
+    "com": "general",
+}
+
+
+def authority_tier(url: str) -> str:
+    host = re.sub(r"^https?://", "", url).split("/")[0].lower().split(":")[0]
+    labels = host.split(".")
+    for suffix, tier in _AUTHORITY_TIERS.items():
+        if labels[-1] == suffix or (len(labels) > 1 and labels[-2] == suffix):
+            return tier
+    return "general"
+
+
+class CitationEntry(BaseModel):
+    url: str
+    visited_at: datetime
+    passages: list[str]
+    claims_supported: int
+    health: str
+    authority_tier: str
+
+
+class VerificationChecklist(BaseModel):
+    run_id: str
+    citations_checked: int
+    live: int
+    stale: int
+    dead: int
+    anchored_claims: int
+    unanchored_claims: int
+    weak_matches: int
+    confidence_labels: dict[str, str]
+    liveness_ratio: float
+
+
+class CitationLedger(BaseModel):
+    run_id: str
+    generated_at: datetime
+    entries: list[CitationEntry]
+    checklist: VerificationChecklist
+
+
+_CLAIM_RE = re.compile(r"\[p-(\d+)\]")
+
+
+def _claims_per_passage(report_text: str) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for match in _CLAIM_RE.finditer(report_text):
+        pid = f"p-{match.group(1)}"
+        counts[pid] = counts.get(pid, 0) + 1
+    return counts
+
+
+def build_ledger(
+    *,
+    run_id: str,
+    report_text: str,
+    evidence: EvidenceStore,
+    liveness: dict[str, LivenessStatus],
+    visited_at: datetime,
+) -> CitationLedger:
+    claims = _claims_per_passage(report_text)
+    all_passages = evidence.writer_context()
+    report_quotes = extract_quotes_from_report(report_text, evidence)
+    entries: list[CitationEntry] = []
+    anchored = 0
+    unanchored = 0
+    weak = 0
+    confidence: dict[str, str] = {}
+
+    for quote in report_quotes:
+        verdict = anchor_quote(quote, all_passages)
+        if verdict.matched:
+            anchored += 1
+        elif verdict.weak:
+            weak += 1
+        else:
+            unanchored += 1
+
+    for page in evidence.pages():
+        passages = evidence.passages_for_url(page.url)
+        passage_ids = [p.passage_id for p in passages]
+        page_claims = sum(claims.get(pid, 0) for pid in passage_ids)
+        health = str(liveness.get(page.url, LivenessStatus.DEAD))
+        entries.append(
+            CitationEntry(
+                url=page.url,
+                visited_at=visited_at,
+                passages=passage_ids,
+                claims_supported=page_claims,
+                health=health,
+                authority_tier=authority_tier(page.url),
+            )
+        )
+        for passage in passages:
+            confidence[passage.passage_id] = (
+                "corroborated" if _corroborated(passage, evidence) else "single-source"
+            )
+
+    live = sum(1 for e in entries if e.health == "live")
+    stale = sum(1 for e in entries if e.health == "stale")
+    dead = sum(1 for e in entries if e.health == "dead")
+    total = len(entries)
+    ratio = round(live / total, 4) if total > 0 else 1.0
+    checklist = VerificationChecklist(
+        run_id=run_id,
+        citations_checked=total,
+        live=live,
+        stale=stale,
+        dead=dead,
+        anchored_claims=anchored,
+        unanchored_claims=unanchored,
+        weak_matches=weak,
+        confidence_labels=confidence,
+        liveness_ratio=ratio,
+    )
+    return CitationLedger(
+        run_id=run_id, generated_at=visited_at, entries=entries, checklist=checklist
+    )
+
+
+def _corroborated(passage: Passage, evidence: EvidenceStore) -> bool:
+    normalized = normalize(passage.text)
+    for other in evidence.writer_context():
+        if other.passage_id == passage.passage_id:
+            continue
+        if normalized in normalize(other.text):
+            return True
+    return False
+
+
+LIVENESS_EXPORT_GATE = 0.99
+
+
+def liveness_export_gate(ledger: CitationLedger) -> tuple[bool, list[str]]:
+    failures = [e.url for e in ledger.entries if e.health != "live"]
+    return ledger.checklist.liveness_ratio >= LIVENESS_EXPORT_GATE, failures
+
+
+def export_checklist(ledger: CitationLedger) -> str:
+    gate_passed, failures = liveness_export_gate(ledger)
+    lines = [
+        "# Verification Checklist",
+        "",
+        f"Run: {ledger.run_id}",
+        f"Citations checked: {ledger.checklist.citations_checked}",
+        f"Live: {ledger.checklist.live} / Stale: {ledger.checklist.stale} / Dead: {ledger.checklist.dead}",
+        (
+            f"Liveness ratio: {ledger.checklist.liveness_ratio} (export gate >= {LIVENESS_EXPORT_GATE}: "
+            f"{'PASS' if gate_passed else 'FAIL'})"
+        ),
+        f"Anchored claims: {ledger.checklist.anchored_claims}",
+        f"Unanchored claims: {ledger.checklist.unanchored_claims}",
+        f"Weak matches (flagged, not accepted): {ledger.checklist.weak_matches}",
+        "",
+        "## Per-URL ledger",
+        "",
+    ]
+    for entry in ledger.entries:
+        lines.append(
+            f"- {entry.url} — health: {entry.health}, tier: {entry.authority_tier}, "
+            f"passages: {len(entry.passages)}, claims: {entry.claims_supported}"
+        )
+    if failures:
+        lines += ["", "## Liveness failures", ""]
+        lines += [f"- {url}" for url in failures]
+    return "\n".join(lines) + "\n"
@@ -0,0 +1,65 @@
+from enum import StrEnum
+
+import httpx
+
+from researchx.adapters.url_guard import url_allowed
+
+_LIVENESS_TIMEOUT = 10.0
+_MAX_RESPONSE_BYTES = 2 * 1024 * 1024
+_MAX_REDIRECTS = 5
+_USER_AGENT = "researchx/0.2 (liveness check)"
+
+
+class LivenessStatus(StrEnum):
+    LIVE = "live"
+    STALE = "stale"
+    DEAD = "dead"
+
+
+async def check_liveness(url: str, client: httpx.AsyncClient | None = None) -> LivenessStatus:
+    own_client = client is None
+    active: httpx.AsyncClient = client or httpx.AsyncClient(
+        timeout=_LIVENESS_TIMEOUT,
+        follow_redirects=False,
+    )
+    current_url = url
+    try:
+        for _ in range(_MAX_REDIRECTS + 1):
+            if not url_allowed(current_url):
+                return LivenessStatus.DEAD
+            response = await active.get(
+                current_url,
+                headers={"User-Agent": _USER_AGENT},
+            )
+            if 300 <= response.status_code < 400:
+                location = response.headers.get("location")
+                if not location:
+                    return LivenessStatus.STALE
+                from urllib.parse import urljoin
+
+                current_url = urljoin(current_url, location)
+                continue
+            content = b""
+            async for chunk in response.aiter_bytes():
+                content += chunk
+                if len(content) > _MAX_RESPONSE_BYTES:
+                    break
+            status = response.status_code
+            if 200 <= status < 300:
+                return LivenessStatus.LIVE
+            return LivenessStatus.DEAD
+        return LivenessStatus.STALE
+    except (httpx.HTTPError, OSError):
+        return LivenessStatus.DEAD
+    finally:
+        if own_client:
+            await active.aclose()
+
+
+async def liveness_report(
+    urls: list[str],
+) -> dict[str, LivenessStatus]:
+    report: dict[str, LivenessStatus] = {}
+    for url in dict.fromkeys(urls):
+        report[url] = await check_liveness(url)
+    return report
@@ -0,0 +1,183 @@
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+
+from researchx.adapters.fakes import CORPUS, FakeModelAdapter, FakeSearchAdapter
+from researchx.adapters.local_storage import LocalStorageAdapter
+from researchx.adapters.types import ChatResult
+from researchx.engine.actor import RunActor, make_run_config
+from researchx.engine.events import EventBus
+from researchx.engine.states import RunStateName, can_transition
+from researchx.methodologies import registry
+
+_PLAN = json.dumps(
+    {
+        "sub_questions": ["methodology verification", "quote anchoring"],
+        "source_universe": ["web"],
+        "outline": ["intro", "findings"],
+    }
+)
+
+_WRITER_REPORT = (
+    "# Report\n\n"
+    "The verification pass re-anchored every quoted claim [p-1]. "
+    "Methodology-first selection converts ad-hoc prompting [p-1] "
+    "into reproducible decisions [p-2].\n"
+)
+
+
+def _canned() -> list[ChatResult]:
+    return [
+        ChatResult(content=_PLAN, model="fake-planner"),
+        ChatResult(content=_WRITER_REPORT, model="fake-writer"),
+    ]
+
+
+def _actor(tmp_path: Path) -> tuple[RunActor, EventBus, FakeSearchAdapter]:
+    _run_id, config = make_run_config(
+        question="How do verification pipelines keep citations honest?",
+        context="",
+        methodology_slug="rapid-answer",
+        overrides={},
+        registry=registry,
+        run_id="abcd2345",
+    )
+    model = FakeModelAdapter(canned=_canned())
+    search = FakeSearchAdapter(CORPUS)
+    storage = LocalStorageAdapter(base_dir=tmp_path)
+    bus = EventBus()
+    actor = RunActor(
+        run_config=config,
+        model=model,
+        search=search,
+        storage=storage,
+        registry=registry,
+        event_bus=bus,
+    )
+    return actor, bus, search
+
+
+async def test_full_walk_reaches_done(tmp_path: Path) -> None:
+    actor, bus, _ = _actor(tmp_path)
+    task = asyncio.create_task(actor.run())
+    while actor.state.state != RunStateName.AWAITING_APPROVAL:
+        await asyncio.sleep(0.01)
+        if task.done():
+            break
+    await actor.approve()
+    state = await task
+    assert state.state == RunStateName.DONE
+    assert state.plan is not None and len(state.plan.sub_questions) == 2
+
+    saved = await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact("abcd2345", "state.json")
+    assert json.loads(saved)["state"] == "DONE"
+
+    report = await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact("abcd2345", "report.md")
+    assert "[p-1]" in report
+
+    ledger_raw = await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact(
+        "abcd2345", "ledger.json"
+    )
+    ledger = json.loads(ledger_raw)
+    assert ledger["checklist"]["citations_checked"] >= 1
+    assert ledger["checklist"]["liveness_ratio"] >= 0
+
+    checklist = await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact(
+        "abcd2345", "checklist.md"
+    )
+    assert "Verification Checklist" in checklist
+
+    types = [e.type for e in bus.replay()]
+    assert "run_started" in types
+    assert "query_fired" in types
+    assert "source_visited" in types
+    assert "passage_extracted" in types
+    assert "run_done" in types
+
+
+async def test_halt_at_awaiting_approval_until_approved(tmp_path: Path) -> None:
+    actor, _, _ = _actor(tmp_path)
+    task = asyncio.create_task(actor.run())
+    for _ in range(100):
+        await asyncio.sleep(0.01)
+        if actor.state.state == RunStateName.AWAITING_APPROVAL:
+            break
+    assert actor.state.state == RunStateName.AWAITING_APPROVAL
+    await asyncio.sleep(0.05)
+    assert not task.done()
+    await actor.approve()
+    state = await task
+    assert state.state == RunStateName.DONE
+
+
+async def test_cancel_persists_cancelled_state(tmp_path: Path) -> None:
+    actor, _, _ = _actor(tmp_path)
+    task = asyncio.create_task(actor.run())
+    while actor.state.state != RunStateName.AWAITING_APPROVAL:
+        await asyncio.sleep(0.01)
+    actor.cancel()
+    await actor.approve()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    saved = await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact("abcd2345", "state.json")
+    assert json.loads(saved)["state"] == "CANCELLED"
+
+
+async def test_pause_resume_mid_run(tmp_path: Path) -> None:
+    actor, _, _ = _actor(tmp_path)
+    task = asyncio.create_task(actor.run())
+    while actor.state.state != RunStateName.AWAITING_APPROVAL:
+        await asyncio.sleep(0.01)
+    await actor.approve()
+    actor.pause()
+    await asyncio.sleep(0.05)
+    assert not task.done()
+    actor.resume()
+    state = await task
+    assert state.state == RunStateName.DONE
+
+
+def test_illegal_transitions_rejected() -> None:
+    assert not can_transition(RunStateName.PLANNING, RunStateName.DONE)
+    assert not can_transition(RunStateName.DONE, RunStateName.PLANNING)
+    assert can_transition(RunStateName.PLANNING, RunStateName.AWAITING_APPROVAL)
+    assert can_transition(RunStateName.REFLECTING, RunStateName.SEARCHING)
+
+
+async def test_planner_non_json_fails_honestly(tmp_path: Path) -> None:
+    _run_id, config = make_run_config(
+        question="q",
+        context="",
+        methodology_slug="rapid-answer",
+        overrides={},
+        registry=registry,
+        run_id="abcd2345",
+    )
+    model = FakeModelAdapter(canned=[ChatResult(content="not json", model="fake-planner")])
+    actor = RunActor(
+        run_config=config,
+        model=model,
+        search=FakeSearchAdapter(CORPUS),
+        storage=LocalStorageAdapter(base_dir=tmp_path),
+        registry=registry,
+        event_bus=EventBus(),
+    )
+    task = asyncio.create_task(actor.run())
+    with pytest.raises(Exception, match="planner returned non-JSON plan"):
+        await task
+    saved = await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact("abcd2345", "state.json")
+    assert json.loads(saved)["state"] == "FAILED"
+
+
+async def test_writer_context_is_visited_only(tmp_path: Path) -> None:
+    actor, _, _ = _actor(tmp_path)
+    task = asyncio.create_task(actor.run())
+    while actor.state.state != RunStateName.AWAITING_APPROVAL:
+        await asyncio.sleep(0.01)
+    await actor.approve()
+    await task
+    assert actor._store.writer_context(), "evidence collected"
+    outside_ids = {p.passage_id for p in actor._store.writer_context()}
+    assert all(pid.startswith("p-") for pid in outside_ids)
@@ -0,0 +1,56 @@
+from researchx.evidence.store import EvidenceStore, extract_passages
+
+
+def test_extract_passages_min_length() -> None:
+    text = "short\n" + ("x" * 150) + "\nanother short"
+    passages = extract_passages(text)
+    assert passages == ["x" * 150]
+
+
+def test_store_stable_ids() -> None:
+    store = EvidenceStore()
+    first = store.add_passage("https://a", "first passage text " * 10)
+    second = store.add_passage("https://a", "second passage text " * 10)
+    assert first.passage_id == "p-1"
+    assert second.passage_id == "p-2"
+
+
+def test_store_add_page_creates_passages() -> None:
+    store = EvidenceStore()
+    body = "\n".join(f"paragraph {i} " + "y" * 150 for i in range(3))
+    store.add_page("https://example.org/a", "Title A", body)
+    assert len(store) == 3
+    page = store.pages()[0]
+    assert page.url == "https://example.org/a"
+    assert page.title == "Title A"
+    assert len(page.passage_ids) == 3
+
+
+def test_writer_context_contains_only_stored() -> None:
+    store = EvidenceStore()
+    store.add_passage("https://a", "stored passage " * 20)
+    context = store.writer_context()
+    assert len(context) == 1
+    assert context[0].url == "https://a"
+
+    outside = EvidenceStore()
+    outside_context = outside.writer_context()
+    assert outside_context == []
+
+
+def test_dedupe_by_quote() -> None:
+    store = EvidenceStore()
+    text = "identical passage text " * 10
+    store.add_passage("https://a", text)
+    assert store.has_seen(text[:200])
+    assert not store.has_seen("something never stored " * 20)
+
+
+def test_passages_for_url() -> None:
+    store = EvidenceStore()
+    store.add_passage("https://a", "from a " * 30)
+    store.add_passage("https://b", "from b " * 30)
+    from_a = store.passages_for_url("https://a")
+    assert len(from_a) == 1
+    assert from_a[0].url == "https://a"
+    assert store.passages_for_url("https://missing") == []
@@ -72,9 +72,19 @@ def test_prompt_templates_render_placeholders() -> None:
    planner = deep.render_planner_prompt(question="What changed?", context="none")
    assert "What changed?" in planner
    assert "none" in planner
-    writer = deep.render_writer_prompt(question="What changed?", context="none", plan="p1")
+    writer = deep.render_writer_prompt(
+        question="What changed?", context="none", plan="p1", evidence="[p-1] (https://a) quote text"
+    )
    assert "{plan}" not in writer
+    assert "{evidence}" not in writer
    assert "p1" in writer
+    assert "[p-1] (https://a) quote text" in writer
+
+
+def test_writer_prompt_requires_evidence_slot() -> None:
+    for slug in ("rapid-answer", "deep-report"):
+        methodology = registry.get(slug)
+        assert "{evidence}" in methodology.writer_prompt


 def test_mutation_of_returned_copy_does_not_affect_registry() -> None:
@@ -0,0 +1,214 @@
+import asyncio
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+
+import httpx
+import pytest
+
+from researchx.adapters.fakes import CORPUS, FakeModelAdapter, FakeSearchAdapter
+from researchx.adapters.local_storage import LocalStorageAdapter
+from researchx.adapters.secrets import EnvFileSecretsProvider
+from researchx.adapters.tavily import TavilySearchAdapter
+from researchx.adapters.types import AdapterError, ChatResult, ModelMessage
+from researchx.engine.actor import RunActor, make_run_config
+from researchx.engine.events import EventBus
+from researchx.engine.run_state import Plan, RunState
+from researchx.engine.states import RunStateName
+from researchx.evidence.store import EvidenceStore
+from researchx.methodologies import registry
+from researchx.trust.anchoring import anchor_quote, extract_quotes_from_report
+
+
+class RecordingModel:
+    def __init__(self, canned: list[ChatResult]) -> None:
+        self._canned = list(canned)
+        self.prompts: list[str] = []
+
+    async def chat(
+        self, messages: list[ModelMessage], *, model: str, temperature: float = 0.2,
+        json_mode: bool = False,
+    ) -> ChatResult:
+        self.prompts.append(messages[0].content)
+        return self._canned.pop(0)
+
+    async def list_models(self) -> list[str]:
+        return ["fake-model"]
+
+    def roles(self):
+        from researchx.adapters.model import ModelRoles
+
+        return ModelRoles(planner="fake-planner", writer="fake-writer")
+
+
+_PLAN = json.dumps({"sub_questions": ["verification"], "source_universe": ["web"], "outline": []})
+
+
+async def test_evidence_reaches_writer_prompt(tmp_path: Path) -> None:
+    _rid, config = make_run_config(
+        question="q", context="", methodology_slug="rapid-answer",
+        overrides={}, registry=registry, run_id="abcd2345",
+    )
+    model = RecordingModel(
+        canned=[
+            ChatResult(content=_PLAN, model="fake-planner"),
+            ChatResult(content="# Report\n\nClaim [p-1].\n", model="fake-writer"),
+        ]
+    )
+    actor = RunActor(
+        run_config=config,
+        model=model,  # type: ignore[arg-type]
+        search=FakeSearchAdapter(CORPUS),
+        storage=LocalStorageAdapter(base_dir=tmp_path),
+        registry=registry,
+        event_bus=EventBus(),
+    )
+    task = asyncio.create_task(actor.run())
+    while actor.state.state == RunStateName.PLANNING:
+        await asyncio.sleep(0.01)
+    await actor.approve()
+    await task
+
+    writer_prompt = model.prompts[-1]
+    assert "[p-1]" in writer_prompt
+    assert "(https://example.org/" in writer_prompt
+    assert "Evidence passages:" in writer_prompt
+
+
+async def test_engine_refuses_private_urls_at_extract(tmp_path: Path, monkeypatch) -> None:
+    env_file = tmp_path / ".env.secrets"
+    env_file.write_text("TAVILY_API_KEY=t-key\n", encoding="utf-8")
+    secrets = EnvFileSecretsProvider(paths=[env_file])
+    tavily = TavilySearchAdapter(secrets)
+
+    called_urls: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        called_urls.append(str(request.url))
+        return httpx.Response(200, text="<html><body>never</body></html>")
+
+    transport = httpx.MockTransport(handler)
+    original = httpx.AsyncClient
+    monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: original(transport=transport, **kw))
+
+    with pytest.raises(AdapterError, match="fetch refused"):
+        await tavily.fetch("http://169.254.169.254/latest/meta-data")
+    assert called_urls == []
+
+
+async def test_resume_from_searching_mid_run(tmp_path: Path) -> None:
+    _rid, config = make_run_config(
+        question="q", context="", methodology_slug="rapid-answer",
+        overrides={}, registry=registry, run_id="abcd2345",
+    )
+    storage = LocalStorageAdapter(base_dir=tmp_path)
+    mid_state = RunState(
+        run_id="abcd2345",
+        question="q",
+        methodology_slug="rapid-answer",
+        state=RunStateName.SEARCHING,
+        plan=Plan(sub_questions=["verification"], source_universe=["web"], outline=[]),
+        depth_cursor=0,
+        depth_target=1,
+    )
+    await storage.write_run_artifact("abcd2345", "state.json", mid_state.model_dump_json())
+
+    model = RecordingModel(
+        canned=[
+            ChatResult(content="# Report\n\nClaim [p-1].\n", model="fake-writer"),
+        ]
+    )
+    actor = RunActor(
+        run_config=config,
+        model=model,  # type: ignore[arg-type]
+        search=FakeSearchAdapter(CORPUS),
+        storage=storage,
+        registry=registry,
+        event_bus=EventBus(),
+    )
+    actor.state = RunState.model_validate_json(
+        await storage.read_run_artifact("abcd2345", "state.json")
+    )
+    state = await actor.run()
+    assert state.state == RunStateName.DONE
+
+
+async def test_cancel_after_approval_persists_cancelled(tmp_path: Path) -> None:
+    _rid, config = make_run_config(
+        question="q", context="", methodology_slug="rapid-answer",
+        overrides={}, registry=registry, run_id="abcd2345",
+    )
+    model = FakeModelAdapter(
+        canned=[
+            ChatResult(content=_PLAN, model="fake-planner"),
+            ChatResult(content="# Report\n\nClaim [p-1].\n", model="fake-writer"),
+        ]
+    )
+    actor = RunActor(
+        run_config=config,
+        model=model,
+        search=FakeSearchAdapter(CORPUS),
+        storage=LocalStorageAdapter(base_dir=tmp_path),
+        registry=registry,
+        event_bus=EventBus(),
+    )
+    task = asyncio.create_task(actor.run())
+    while actor.state.state == RunStateName.PLANNING:
+        await asyncio.sleep(0.01)
+    await actor.approve()
+    actor.pause()
+    actor.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    saved = json.loads(
+        await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact("abcd2345", "state.json")
+    )
+    assert saved["state"] == "CANCELLED"
+
+
+def test_quote_mismatch_flagged_in_ledger_path() -> None:
+    store = EvidenceStore()
+    real_passage = store.add_passage("https://a", "The verified sentence about anchoring " * 5)
+    hallucinated = "This quote was never in any stored passage at all whatsoever."
+
+    verdict = anchor_quote(hallucinated, [real_passage])
+    assert verdict.matched is False
+    assert verdict.weak is False
+
+    good = anchor_quote(real_passage.quote, [real_passage])
+    assert good.matched is True
+
+    report = f"Claim [p-1].\n\n> {hallucinated} [p-1]\n"
+    quotes = extract_quotes_from_report(report, store)
+    assert len(quotes) == 2
+    bad_verdict = anchor_quote(hallucinated, store.writer_context())
+    assert bad_verdict.matched is False
+
+
+def test_ledger_export_gate() -> None:
+    from researchx.trust.ledger import build_ledger, liveness_export_gate
+    from researchx.trust.liveness import LivenessStatus
+
+    store = EvidenceStore()
+    store.add_passage("https://a", "passage one " * 20)
+    ledger = build_ledger(
+        run_id="abcd2345",
+        report_text="[p-1]",
+        evidence=store,
+        liveness={"https://a": LivenessStatus.LIVE},
+        visited_at=datetime.now(UTC),
+    )
+    passed, failures = liveness_export_gate(ledger)
+    assert passed is True
+    assert failures == []
+
+    ledger_dead = build_ledger(
+        run_id="abcd2345",
+        report_text="[p-1]",
+        evidence=store,
+        liveness={"https://a": LivenessStatus.DEAD},
+        visited_at=datetime.now(UTC),
+    )
+    passed, failures = liveness_export_gate(ledger_dead)
+    assert passed is False
+    assert failures == ["https://a"]
@@ -0,0 +1,129 @@
+import asyncio
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+
+from researchx.adapters.capability_probe import probe_model_adapter
+from researchx.adapters.fakes import CORPUS, FakeModelAdapter, FakeSearchAdapter
+from researchx.adapters.local_storage import LocalStorageAdapter
+from researchx.adapters.ollama import OllamaModelAdapter
+from researchx.adapters.secrets import EnvFileSecretsProvider
+from researchx.adapters.types import ChatResult
+from researchx.engine.actor import RunActor, make_run_config
+from researchx.engine.events import EventBus
+from researchx.engine.states import RunStateName
+from researchx.methodologies import registry
+
+
+async def test_probe_attaches_to_run_state(tmp_path: Path, monkeypatch) -> None:
+    env_file = tmp_path / ".env.secrets"
+    env_file.write_text("OLLAMA_API_KEY=cloud-key\n", encoding="utf-8")
+    secrets = EnvFileSecretsProvider(paths=[env_file])
+    adapter = OllamaModelAdapter(secrets, planner_model="p", writer_model="w")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if str(request.url).endswith("/models"):
+            return httpx.Response(200, json={"data": [{"id": "w"}]})
+        return httpx.Response(
+            200, json={"choices": [{"message": {"content": '{"ok": true}'}}]}
+        )
+
+    transport = httpx.MockTransport(handler)
+    original = httpx.AsyncClient
+    monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: original(transport=transport, **kw))
+
+    probe = await probe_model_adapter(adapter)
+    assert probe.structured_outputs_supported is True
+
+    _rid, config = make_run_config(
+        question="q",
+        context="",
+        methodology_slug="rapid-answer",
+        overrides={},
+        registry=registry,
+        run_id="abcd2345",
+    )
+    plan = json.dumps({"sub_questions": ["verification"], "source_universe": ["web"], "outline": []})
+    writer = "# Report\n\nClaim [p-1].\n"
+    fake_model = FakeModelAdapter(
+        canned=[
+            ChatResult(content=plan, model="fake-planner"),
+            ChatResult(content=writer, model="fake-writer"),
+        ]
+    )
+    actor = RunActor(
+        run_config=config,
+        model=fake_model,
+        search=FakeSearchAdapter(CORPUS),
+        storage=LocalStorageAdapter(base_dir=tmp_path),
+        registry=registry,
+        event_bus=EventBus(),
+    )
+    await actor.attach_probe(probe.model_dump(mode="json"))
+    assert actor.state.probe is not None
+    assert actor.state.probe["structured_outputs_supported"] is True
+
+    task = asyncio.create_task(actor.run())
+    while actor.state.state == RunStateName.PLANNING:
+        await asyncio.sleep(0.01)
+    await actor.approve()
+    state = await task
+    assert state.state == RunStateName.DONE
+    saved = json.loads(
+        await LocalStorageAdapter(base_dir=tmp_path).read_run_artifact("abcd2345", "state.json")
+    )
+    assert saved["probe"]["endpoint"] == "https://ollama.com/v1"
+
+
+async def test_probe_failure_never_crashes_the_run(tmp_path: Path, monkeypatch) -> None:
+    env_file = tmp_path / ".env.secrets"
+    env_file.write_text("OLLAMA_API_KEY=cloud-key\n", encoding="utf-8")
+    secrets = EnvFileSecretsProvider(paths=[env_file])
+    adapter = OllamaModelAdapter(secrets, planner_model="p", writer_model="w")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, text="down")
+
+    transport = httpx.MockTransport(handler)
+    original = httpx.AsyncClient
+    monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: original(transport=transport, **kw))
+
+    from researchx.adapters.types import AdapterError
+
+    with pytest.raises(AdapterError):
+        await probe_model_adapter(adapter)
+
+    _rid, config = make_run_config(
+        question="q",
+        context="",
+        methodology_slug="rapid-answer",
+        overrides={},
+        registry=registry,
+        run_id="abcd2345",
+    )
+    plan = json.dumps({"sub_questions": ["verification"], "source_universe": ["web"], "outline": []})
+    writer = "# Report\n\nClaim [p-1].\n"
+    fake_model = FakeModelAdapter(
+        canned=[
+            ChatResult(content=plan, model="fake-planner"),
+            ChatResult(content=writer, model="fake-writer"),
+        ]
+    )
+    bus = EventBus()
+    actor = RunActor(
+        run_config=config,
+        model=fake_model,
+        search=FakeSearchAdapter(CORPUS),
+        storage=LocalStorageAdapter(base_dir=tmp_path),
+        registry=registry,
+        event_bus=bus,
+    )
+    task = asyncio.create_task(actor.run())
+    while actor.state.state == RunStateName.PLANNING:
+        await asyncio.sleep(0.01)
+    await actor.approve()
+    state = await task
+    assert state.state == RunStateName.DONE
+    assert state.probe is None
@@ -0,0 +1,105 @@
+import json
+from pathlib import Path
+
+from researchx.adapters.fakes import CORPUS, FakeModelAdapter, FakeSearchAdapter
+from researchx.adapters.local_storage import LocalStorageAdapter
+from researchx.adapters.types import ChatResult
+from researchx.engine.actor import RunActor, make_run_config
+from researchx.engine.events import EventBus
+from researchx.engine.run_state import Plan, RunState
+from researchx.engine.states import RunStateName
+from researchx.methodologies import registry
+
+_PLAN = json.dumps(
+    {
+        "sub_questions": ["verification passes"],
+        "source_universe": ["web"],
+        "outline": ["intro"],
+    }
+)
+_WRITER = "# Report\n\nClaim [p-1] and claim [p-2].\n"
+
+
+def test_run_id_generator_is_the_storage_one() -> None:
+    from researchx.engine.actor import make_run_config as mrc
+
+    run_id, _ = mrc(
+        question="q", context="", methodology_slug="rapid-answer", overrides={}, registry=registry
+    )
+    assert len(run_id) == 8
+    alphabet = set("abcdefghjkmnpqrstuvwxyz23456789")
+    assert all(c in alphabet for c in run_id)
+
+
+async def test_state_file_round_trip_enables_restart(tmp_path: Path) -> None:
+    run_id, config = make_run_config(
+        question="How do engines resume?",
+        context="",
+        methodology_slug="rapid-answer",
+        overrides={},
+        registry=registry,
+        run_id="abcd2345",
+    )
+    storage = LocalStorageAdapter(base_dir=tmp_path)
+    state = RunState(
+        run_id=run_id,
+        question=config.question,
+        methodology_slug="rapid-answer",
+        state=RunStateName.SEARCHING,
+        depth_cursor=1,
+        depth_target=3,
+    )
+    await storage.write_run_artifact(run_id, "state.json", state.model_dump_json())
+
+    restored = RunState.model_validate_json(
+        await storage.read_run_artifact(run_id, "state.json")
+    )
+    assert restored.depth_cursor == 1
+    assert restored.depth_target == 3
+    assert restored.state == RunStateName.SEARCHING
+
+
+async def test_actor_resumes_from_restored_state(tmp_path: Path) -> None:
+    run_id, config = make_run_config(
+        question="How do engines resume?",
+        context="",
+        methodology_slug="rapid-answer",
+        overrides={},
+        registry=registry,
+        run_id="abcd2345",
+    )
+    storage = LocalStorageAdapter(base_dir=tmp_path)
+    mid_state = RunState(
+        run_id=run_id,
+        question=config.question,
+        methodology_slug="rapid-answer",
+        state=RunStateName.AWAITING_APPROVAL,
+        depth_cursor=0,
+        depth_target=1,
+        plan=Plan(sub_questions=["verification passes"], source_universe=["web"], outline=["intro"]),
+    )
+    await storage.write_run_artifact(run_id, "state.json", mid_state.model_dump_json())
+
+    model = FakeModelAdapter(
+        canned=[
+            ChatResult(content=_PLAN, model="fake-planner"),
+            ChatResult(content=_WRITER, model="fake-writer"),
+        ]
+    )
+    actor = RunActor(
+        run_config=config,
+        model=model,
+        search=FakeSearchAdapter(CORPUS),
+        storage=storage,
+        registry=registry,
+        event_bus=EventBus(),
+    )
+    restored = RunState.model_validate_json(
+        await storage.read_run_artifact(run_id, "state.json")
+    )
+    actor.state = restored
+    await actor.approve()
+    state = await actor.run()
+    assert state.state == RunStateName.DONE
+    assert state.plan is not None
+    assert state.plan.sub_questions == ["verification passes"]
@@ -0,0 +1,155 @@
+from datetime import UTC, datetime
+
+import httpx
+
+from researchx.adapters.fakes import CORPUS, SHARED_SENTENCE
+from researchx.evidence.store import EvidenceStore
+from researchx.trust.anchoring import anchor_quote, normalize
+from researchx.trust.ledger import build_ledger, export_checklist
+from researchx.trust.liveness import LivenessStatus, check_liveness, liveness_report, url_allowed
+
+
+def test_normalize_pipeline() -> None:
+    assert normalize("  Foo   Bar  ") == "foo bar"
+    assert normalize("Café") == normalize("café")
+
+
+def test_anchor_exact_substring_short_circuits() -> None:
+    store = EvidenceStore()
+    passage = store.add_passage("https://a", SHARED_SENTENCE)
+    verdict = anchor_quote(SHARED_SENTENCE, [passage])
+    assert verdict.matched is True
+    assert verdict.ratio == 1.0
+    assert verdict.weak is False
+
+
+def test_anchor_weak_match_flagged() -> None:
+    store = EvidenceStore()
+    passage = store.add_passage("https://a", SHARED_SENTENCE)
+    mutated = SHARED_SENTENCE.replace(
+        "The verification pass re-anchored every quoted claim",
+        "A liveness check validated the cited evidence",
+    )
+    verdict = anchor_quote(mutated, [passage])
+    assert verdict.ratio < 0.85
+
+
+def test_anchor_high_fuzzy_match() -> None:
+    store = EvidenceStore()
+    passage = store.add_passage("https://a", "The pipeline verifies every quoted claim against its stored passage before shipping.")
+    near = "The pipeline verify every quoted claim against its stored passage before shipping."
+    verdict = anchor_quote(near, [passage])
+    assert verdict.ratio >= 0.85
+    assert verdict.matched is True
+
+
+def test_anchor_no_match_returns_unmatched() -> None:
+    store = EvidenceStore()
+    passage = store.add_passage("https://a", SHARED_SENTENCE)
+    verdict = anchor_quote("completely unrelated text about something else entirely", [passage])
+    assert verdict.matched is False
+    assert verdict.weak is False
+    assert verdict.passage_id is None
+
+
+def test_url_guard_blocks_private_and_schemes() -> None:
+    assert url_allowed("http://169.254.169.254/latest/meta-data") is False
+    assert url_allowed("http://localhost:8000/api") is False
+    assert url_allowed("http://127.0.0.1/x") is False
+    assert url_allowed("http://10.0.0.1/x") is False
+    assert url_allowed("file:///etc/passwd") is False
+    assert url_allowed("ftp://example.org/x") is False
+    assert url_allowed("not a url") is False
+
+
+async def test_liveness_classification_live(monkeypatch) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, text="alive page")
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, follow_redirects=True) as client:
+        status = await check_liveness("https://example.org/live", client)
+    assert status == LivenessStatus.LIVE
+
+
+async def test_liveness_dead_on_error(monkeypatch) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, text="boom")
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport) as client:
+        status = await check_liveness("https://example.org/dead", client)
+    assert status == LivenessStatus.DEAD
+
+
+async def test_liveness_report_dedupes_and_covers() -> None:
+    report = await liveness_report(["https://a", "https://a", "https://b"])
+    assert report["https://a"] == LivenessStatus.DEAD
+    assert len(report) == 2
+
+
+async def test_liveness_blocked_url_reports_dead() -> None:
+    status = await check_liveness("http://169.254.169.254/latest/meta-data")
+    assert status == LivenessStatus.DEAD
+
+
+def _store_with_report() -> tuple[EvidenceStore, str]:
+    store = EvidenceStore()
+    for page in CORPUS[:3]:
+        store.add_page(page.url, page.title, page.text)
+    report = f"# Report\n\nClaim one [{store.pages()[0].passage_ids[0]}] and claim two [p-1].\n"
+    return store, report
+
+
+async def test_ledger_build_and_checklist_export(monkeypatch) -> None:
+    store, report = _store_with_report()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, text="ok")
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, follow_redirects=True) as client:
+        liveness = {}
+        for page in store.pages():
+            liveness[page.url] = await check_liveness(page.url, client)
+
+    ledger = build_ledger(
+        run_id="abcd2345",
+        report_text=report,
+        evidence=store,
+        liveness=liveness,
+        visited_at=datetime.now(UTC),
+    )
+    assert ledger.run_id == "abcd2345"
+    assert len(ledger.entries) == len(store.pages())
+    assert ledger.checklist.citations_checked == len(store.pages())
+    assert ledger.checklist.live == len(store.pages())
+    assert ledger.checklist.dead == 0
+    assert ledger.checklist.anchored_claims >= 1
+    assert all(
+        entry.authority_tier in ("reference", "official", "academic", "general")
+        for entry in ledger.entries
+    )
+
+    checklist = export_checklist(ledger)
+    assert "Verification Checklist" in checklist
+    assert "Live:" in checklist
+    assert "Weak matches" in checklist
+
+
+def test_ledger_confidence_labels_single_vs_corroborated() -> None:
+    store = EvidenceStore()
+    store.add_passage("https://a", SHARED_SENTENCE)
+    store.add_passage("https://b", SHARED_SENTENCE)
+    report = "[p-1] [p-2]"
+    ledger = build_ledger(
+        run_id="abcd2345",
+        report_text=report,
+        evidence=store,
+        liveness={"https://a": LivenessStatus.LIVE, "https://b": LivenessStatus.LIVE},
+        visited_at=datetime.now(UTC),
+    )
+    labels = ledger.checklist.confidence_labels
+    assert len(labels) == 2
+    corroborated = [v for v in labels.values() if v == "corroborated"]
+    assert len(corroborated) >= 1