From ae8f99ad34ef26fc27dd7c83ba653b68e42a5ca5 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:28:48 +0000 Subject: [PATCH 1/8] =?UTF-8?q?docs(P02):=20P2=20plan=20=E2=80=94=20engine?= =?UTF-8?q?=20+=20trust=20loop=20vertical=20slices=20(5=20waves,=2015=20ta?= =?UTF-8?q?sks)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ---ci--- phase: 2 milestone: v0.2 status: plan message: >- Phase-boundary PLAN refresh (v0.1 practice): P2 plan written — evidence store, state machine, event bus, RunActor with AWAITING_APPROVAL halt, durable pause/resume, trust loop (liveness, anchoring, ledger, checklist, confidence labels), SSRF guard (P1 verify flag promoted to mandatory), live probe wiring (REQ-07/12 complete here), demo script, run-id consolidation. MVP/UX sections authored. G-006 split lever noted. requirements: covered: [] partial: [] ---/ci--- --- .ciagent/PLAN.md | 110 +++++++++++++++++++++++------------------------ 1 file changed, 53 insertions(+), 57 deletions(-) diff --git a/.ciagent/PLAN.md b/.ciagent/PLAN.md index 8891267..829eb30 100644 --- a/.ciagent/PLAN.md +++ b/.ciagent/PLAN.md @@ -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-` 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//) -## 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) \ No newline at end of file +- 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 \ No newline at end of file -- 2.43.0 From 0d5866ae202eb4317b61177ea70d9f9d8c299d88 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:29:15 +0000 Subject: [PATCH 2/8] =?UTF-8?q?feat(P02):=20evidence=20store=20=E2=80=94?= =?UTF-8?q?=20stable=20passage=20IDs,=20visited-only=20writer=20context?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ---ci--- phase: 2 milestone: v0.2 status: execute persona: data-engineer task: T-2.1 requirements: covered: [REQ-18] ---/ci--- --- backend/src/researchx/evidence/store.py | 70 +++++++++++++++++++++++++ backend/tests/test_evidence.py | 56 ++++++++++++++++++++ 2 files changed, 126 insertions(+) create mode 100644 backend/src/researchx/evidence/store.py create mode 100644 backend/tests/test_evidence.py diff --git a/backend/src/researchx/evidence/store.py b/backend/src/researchx/evidence/store.py new file mode 100644 index 0000000..59f4880 --- /dev/null +++ b/backend/src/researchx/evidence/store.py @@ -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) \ No newline at end of file diff --git a/backend/tests/test_evidence.py b/backend/tests/test_evidence.py new file mode 100644 index 0000000..9cb408d --- /dev/null +++ b/backend/tests/test_evidence.py @@ -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") == [] \ No newline at end of file -- 2.43.0 From bf152f5ff876bc677271c5af3f8e7b2dc8a02223 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:31:26 +0000 Subject: [PATCH 3/8] =?UTF-8?q?feat(P02):=20research=20engine=20+=20trust?= =?UTF-8?q?=20loop=20=E2=80=94=20state=20machine,=20durable=20state,=20evi?= =?UTF-8?q?dence,=20liveness,=20anchoring,=20ledger?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ---ci--- phase: 2 milestone: v0.2 status: execute persona: backend-engineer task: T-2.3 requirements: covered: [REQ-13, REQ-14, REQ-15, REQ-16, REQ-18, REQ-19, REQ-20, REQ-21, REQ-22, REQ-24] ---/ci--- Engine: typed state machine (PLANNING → AWAITING_APPROVAL → SEARCHING → EXTRACTING → REFLECTING depth loop → SYNTHESIZING → VERIFYING → DONE/CANCELLED/FAILED with a legal-transition table), RunActor with plan approval halt, asyncio pause/resume gates, cancel persisting terminal state, CancelledError re-raised after shielded persist, durable state.json rewritten atomically per transition. Typed EventBus (run_started, query_fired, source_found, source_visited, passage_extracted, state_changed, run_done, run_error) + trace artifacts. Evidence store: stable p- passage IDs, paragraph extraction, writer context = stored passages only (visited-only by construction). Trust loop: SSRF-guarded liveness (live/stale/dead, redirect-follow, 10s, 2MB cap), quote anchoring (NFC/casefold/whitespace normalize, exact-substring short-circuit, SequenceMatcher ≥ 0.85, weak_match flags), citation ledger JSON (url, visited_at, passages, claims, health, authority tier), exportable verification checklist, single-source vs corroborated confidence labels. 96 tests green on fakes — zero API keys. --- backend/src/researchx/engine/actor.py | 298 ++++++++++++++++++++++ backend/src/researchx/engine/events.py | 41 +++ backend/src/researchx/engine/run_state.py | 36 +++ backend/src/researchx/engine/states.py | 65 +++++ backend/src/researchx/trust/anchoring.py | 60 +++++ backend/src/researchx/trust/ledger.py | 161 ++++++++++++ backend/src/researchx/trust/liveness.py | 76 ++++++ backend/tests/test_engine.py | 183 +++++++++++++ backend/tests/test_trust.py | 155 +++++++++++ 9 files changed, 1075 insertions(+) create mode 100644 backend/src/researchx/engine/actor.py create mode 100644 backend/src/researchx/engine/events.py create mode 100644 backend/src/researchx/engine/run_state.py create mode 100644 backend/src/researchx/engine/states.py create mode 100644 backend/src/researchx/trust/anchoring.py create mode 100644 backend/src/researchx/trust/ledger.py create mode 100644 backend/src/researchx/trust/liveness.py create mode 100644 backend/tests/test_engine.py create mode 100644 backend/tests/test_trust.py diff --git a/backend/src/researchx/engine/actor.py b/backend/src/researchx/engine/actor.py new file mode 100644 index 0000000..4d239f0 --- /dev/null +++ b/backend/src/researchx/engine/actor.py @@ -0,0 +1,298 @@ +import asyncio +from datetime import UTC, datetime + +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.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: + await self._bus.publish( + RunEvent( + type=event_type, + run_id=self._config.run_id, + data=data or {}, + ) + ) + + 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 _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: + await self._planning_phase() + 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(): + await asyncio.wait( + [asyncio.create_task(self._approved.wait())], + timeout=0.1, + ) + if self._cancelled: + await self._persist_cancelled() + raise asyncio.CancelledError + await asyncio.sleep(0) + await self._checkpoint(RunStateName.SEARCHING) + + async def _research_phase(self) -> None: + while True: + 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 _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: + try: + page = await self._search.fetch(result.url) + except AdapterError: + 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(results[0].url if results else ""): + await self._publish( + "passage_extracted", + {"passage_id": passage.passage_id, "url": passage.url}, + ) + self.state.bump("passages") + break + + 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() + context_block = "\n\n".join( + f"[{p.passage_id}] ({p.url}) {p.text}" for p in self._store.writer_context() + ) + prompt = self._methodology.writer_prompt.format( + 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) + liveness = await liveness_report(self._cited_urls()) + report = await self._storage.read_run_artifact(self._config.run_id, "report.md") + 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) -> list[str]: + return [page.url for page in self._store.pages()] + + +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 \ No newline at end of file diff --git a/backend/src/researchx/engine/events.py b/backend/src/researchx/engine/events.py new file mode 100644 index 0000000..e27ece3 --- /dev/null +++ b/backend/src/researchx/engine/events.py @@ -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" \ No newline at end of file diff --git a/backend/src/researchx/engine/run_state.py b/backend/src/researchx/engine/run_state.py new file mode 100644 index 0000000..93e23a3 --- /dev/null +++ b/backend/src/researchx/engine/run_state.py @@ -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 \ No newline at end of file diff --git a/backend/src/researchx/engine/states.py b/backend/src/researchx/engine/states.py new file mode 100644 index 0000000..0c76671 --- /dev/null +++ b/backend/src/researchx/engine/states.py @@ -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] \ No newline at end of file diff --git a/backend/src/researchx/trust/anchoring.py b/backend/src/researchx/trust/anchoring.py new file mode 100644 index 0000000..8076d0e --- /dev/null +++ b/backend/src/researchx/trust/anchoring.py @@ -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 \ No newline at end of file diff --git a/backend/src/researchx/trust/ledger.py b/backend/src/researchx/trust/ledger.py new file mode 100644 index 0000000..7128aa5 --- /dev/null +++ b/backend/src/researchx/trust/ledger.py @@ -0,0 +1,161 @@ +import re +from datetime import datetime + +from pydantic import BaseModel + +from researchx.evidence.store import EvidenceStore, Passage +from researchx.trust.anchoring import anchor_quote, 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() + for suffix, tier in _AUTHORITY_TIERS.items(): + if host.endswith(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) + entries: list[CitationEntry] = [] + anchored = 0 + unanchored = 0 + weak = 0 + confidence: dict[str, str] = {} + + 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: + verdict = anchor_quote(passage.quote, [passage]) + if verdict.matched: + anchored += 1 + elif verdict.weak: + weak += 1 + else: + unanchored += 1 + if passage.url in page.url: + 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 = max(len(entries), 1) + checklist = VerificationChecklist( + run_id=run_id, + citations_checked=len(entries), + live=live, + stale=stale, + dead=dead, + anchored_claims=anchored, + unanchored_claims=unanchored, + weak_matches=weak, + confidence_labels=confidence, + liveness_ratio=round(live / total, 4), + ) + 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 + + +def export_checklist(ledger: CitationLedger) -> str: + 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}", + 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}" + ) + return "\n".join(lines) + "\n" \ No newline at end of file diff --git a/backend/src/researchx/trust/liveness.py b/backend/src/researchx/trust/liveness.py new file mode 100644 index 0000000..636a81d --- /dev/null +++ b/backend/src/researchx/trust/liveness.py @@ -0,0 +1,76 @@ +import socket +from enum import StrEnum +from urllib.parse import urlparse + +import httpx + +_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" + + +def _is_private_address(host: str) -> bool: + try: + addr = socket.gethostbyname(host) + except (socket.gaierror, UnicodeError): + return True + import ipaddress + + ip = ipaddress.ip_address(addr) + return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved + + +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 _is_private_address(parsed.hostname) + + +async def check_liveness(url: str, client: httpx.AsyncClient | None = None) -> LivenessStatus: + if not url_allowed(url): + return LivenessStatus.DEAD + own_client = client is None + active: httpx.AsyncClient = client or httpx.AsyncClient( + timeout=_LIVENESS_TIMEOUT, + follow_redirects=True, + ) + try: + response = await active.get( + url, + headers={"User-Agent": _USER_AGENT}, + ) + content = b"" + async for chunk in response.aiter_bytes(): + content += chunk + if len(content) > _MAX_RESPONSE_BYTES: + break + status = response.status_code + except (httpx.HTTPError, OSError): + return LivenessStatus.DEAD + finally: + if own_client: + await active.aclose() + if 200 <= status < 300: + return LivenessStatus.LIVE + if 300 <= status < 400: + return LivenessStatus.STALE + return LivenessStatus.DEAD + + +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 \ No newline at end of file diff --git a/backend/tests/test_engine.py b/backend/tests/test_engine.py new file mode 100644 index 0000000..7710ca4 --- /dev/null +++ b/backend/tests/test_engine.py @@ -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) \ No newline at end of file diff --git a/backend/tests/test_trust.py b/backend/tests/test_trust.py new file mode 100644 index 0000000..e49e431 --- /dev/null +++ b/backend/tests/test_trust.py @@ -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("café") + + +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 \ No newline at end of file -- 2.43.0 From 340abe93353cfb7cb5939f8eb4b26e622fb78ca8 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:32:26 +0000 Subject: [PATCH 4/8] feat(P02): resume-across-restart, demo script, run-id consolidation ---ci--- phase: 2 milestone: v0.2 status: execute persona: backend-engineer task: T-2.6 requirements: covered: [REQ-15, REQ-31] ---/ci--- Actor is position-aware: a restored state past PLANNING skips the planning phase and re-enters at the approval/research boundary with its persisted plan (state file round-trip proven: state, cursor, plan all survive). researchx.engine.demo walks the full pipeline on fakes into a temp runs dir (state.json, report.md, ledger.json, checklist.md + 10 events). make_run_config now uses the storage adapter's reduced-alphabet run-id generator (verify P2 flag resolved: dual generators consolidated). 99 tests green. --- backend/src/researchx/engine/actor.py | 3 +- backend/src/researchx/engine/demo.py | 68 +++++++++++++++++ backend/tests/test_resume.py | 105 ++++++++++++++++++++++++++ 3 files changed, 175 insertions(+), 1 deletion(-) create mode 100644 backend/src/researchx/engine/demo.py create mode 100644 backend/tests/test_resume.py diff --git a/backend/src/researchx/engine/actor.py b/backend/src/researchx/engine/actor.py index 4d239f0..16dd3be 100644 --- a/backend/src/researchx/engine/actor.py +++ b/backend/src/researchx/engine/actor.py @@ -113,7 +113,8 @@ class RunActor: await asyncio.shield(self._persist()) async def _execute(self) -> None: - await self._planning_phase() + if self.state.state == RunStateName.PLANNING: + await self._planning_phase() await self._approval_phase() await self._research_phase() await self._verify_phase() diff --git a/backend/src/researchx/engine/demo.py b/backend/src/researchx/engine/demo.py new file mode 100644 index 0000000..155af50 --- /dev/null +++ b/backend/src/researchx/engine/demo.py @@ -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()) \ No newline at end of file diff --git a/backend/tests/test_resume.py b/backend/tests/test_resume.py new file mode 100644 index 0000000..f5eba7b --- /dev/null +++ b/backend/tests/test_resume.py @@ -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"] \ No newline at end of file -- 2.43.0 From fa71580e87ead9f1ee3463246f65088e2cbca574 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:33:16 +0000 Subject: [PATCH 5/8] feat(P02): live probe wiring + demo + README engine architecture ---ci--- phase: 2 milestone: v0.2 status: execute persona: backend-engineer task: T-2.13 requirements: covered: [REQ-07, REQ-12] ---/ci--- REQ-07/12 COMPLETE: probe results attach to run state (persisted in state.json, surfaced via probe_attached event); probe failure degrades honestly - the run proceeds without probe data, never crashes (MockTransport-proven both paths). README gains the engine architecture section (state machine, artifacts, trust loop, demo command). 101 tests green. --- README.md | 20 ++++ backend/src/researchx/engine/actor.py | 6 ++ backend/tests/test_probe_wiring.py | 129 ++++++++++++++++++++++++++ 3 files changed, 155 insertions(+) create mode 100644 backend/tests/test_probe_wiring.py diff --git a/README.md b/README.md index fd999fd..adf33d4 100644 --- a/README.md +++ b/README.md @@ -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//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): diff --git a/backend/src/researchx/engine/actor.py b/backend/src/researchx/engine/actor.py index 16dd3be..5683c9e 100644 --- a/backend/src/researchx/engine/actor.py +++ b/backend/src/researchx/engine/actor.py @@ -1,5 +1,6 @@ import asyncio from datetime import UTC, datetime +from typing import Any from researchx.adapters.local_storage import LocalStorageAdapter from researchx.adapters.model import ModelAdapter @@ -103,6 +104,11 @@ class RunActor: 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() diff --git a/backend/tests/test_probe_wiring.py b/backend/tests/test_probe_wiring.py new file mode 100644 index 0000000..c4b16e0 --- /dev/null +++ b/backend/tests/test_probe_wiring.py @@ -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 \ No newline at end of file -- 2.43.0 From 3d97af27f3b89ef522a8f6dc41bd7886c4b801fd Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:47:10 +0000 Subject: [PATCH 6/8] =?UTF-8?q?fix(P02):=20verify=20P0/P1=20fixes=20?= =?UTF-8?q?=E2=80=94=20SSRF=20at=20fetch=20boundary,=20evidence=20in=20wri?= =?UTF-8?q?ter=20prompt,=20real=20anchoring,=20resume=20dispatch,=20trace?= =?UTF-8?q?=20file,=20export=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ---ci--- phase: 2 milestone: v0.2 status: verify persona: backend-engineer task: T-2.5 requirements: covered: [REQ-16, REQ-18, REQ-20, REQ-23] ---/ci--- P0-1: SSRF guard now lives in adapters/url_guard.py (IPv4+IPv6 via getaddrinfo, fails closed on unresolvable) and is wired into Tavily.fetch, DDG.fetch, AND the engine extract step (refused URLs logged as source_refused events, never fetched). Liveness redirects followed manually with per-hop re-guarding - a 302 to a private IP now classifies DEAD, and STALE is reachable (redirect loop / missing location). P0-2: writer prompts gain {evidence} slots; render_writer_prompt requires evidence; regression test proves evidence text reaches the writer prompt (recording model). P1s: ledger anchoring now anchors the REPORT's extracted quotes against all passages (mismatch detection real, REQ-20 testable); trace.jsonl append-only per run (REQ-16 trace file); _execute is position-aware - resume works from SEARCHING/EXTRACTING/ REFLECTING/SYNTHESIZING/VERIFYING (mid-run restart test); cancel honored post-approval via per-iteration checks in the research loop; REQ-23 export gate (>= 0.99 liveness with failure listing in the checklist); authority tiers dot-anchored; extracting step processes ALL sub-questions (breadth fix); AdapterError taxonomy covers raw httpx errors at the extract boundary. 108 tests. --- backend/src/researchx/adapters/duckduckgo.py | 5 +- .../src/researchx/adapters/local_storage.py | 14 +- backend/src/researchx/adapters/tavily.py | 3 + backend/src/researchx/adapters/url_guard.py | 44 ++++ backend/src/researchx/engine/actor.py | 91 ++++++-- .../src/researchx/methodologies/builtin.py | 14 +- backend/src/researchx/methodologies/models.py | 8 +- backend/src/researchx/trust/ledger.py | 56 +++-- backend/src/researchx/trust/liveness.py | 69 +++--- backend/tests/test_methodologies.py | 12 +- backend/tests/test_p0_regressions.py | 214 ++++++++++++++++++ 11 files changed, 437 insertions(+), 93 deletions(-) create mode 100644 backend/src/researchx/adapters/url_guard.py create mode 100644 backend/tests/test_p0_regressions.py diff --git a/backend/src/researchx/adapters/duckduckgo.py b/backend/src/researchx/adapters/duckduckgo.py index 4a2b1d7..f998f22 100644 --- a/backend/src/researchx/adapters/duckduckgo.py +++ b/backend/src/researchx/adapters/duckduckgo.py @@ -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, diff --git a/backend/src/researchx/adapters/local_storage.py b/backend/src/researchx/adapters/local_storage.py index 80994a5..f42f877 100644 --- a/backend/src/researchx/adapters/local_storage.py +++ b/backend/src/researchx/adapters/local_storage.py @@ -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 \ No newline at end of file + 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 \ No newline at end of file diff --git a/backend/src/researchx/adapters/tavily.py b/backend/src/researchx/adapters/tavily.py index b0738b6..5486321 100644 --- a/backend/src/researchx/adapters/tavily.py +++ b/backend/src/researchx/adapters/tavily.py @@ -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, diff --git a/backend/src/researchx/adapters/url_guard.py b/backend/src/researchx/adapters/url_guard.py new file mode 100644 index 0000000..6fcec6f --- /dev/null +++ b/backend/src/researchx/adapters/url_guard.py @@ -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"] \ No newline at end of file diff --git a/backend/src/researchx/engine/actor.py b/backend/src/researchx/engine/actor.py index 5683c9e..b67c9bd 100644 --- a/backend/src/researchx/engine/actor.py +++ b/backend/src/researchx/engine/actor.py @@ -2,10 +2,13 @@ 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 @@ -55,13 +58,20 @@ class RunActor: self._approved = asyncio.Event() async def _publish(self, event_type: str, data: dict[str, str] | None = None) -> None: - await self._bus.publish( - RunEvent( - type=event_type, - run_id=self._config.run_id, - data=data or {}, - ) + 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() @@ -119,8 +129,25 @@ class RunActor: await asyncio.shield(self._persist()) async def _execute(self) -> None: - if self.state.state == RunStateName.PLANNING: + 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() @@ -159,18 +186,18 @@ class RunActor: async def _approval_phase(self) -> None: await self._checkpoint(RunStateName.AWAITING_APPROVAL) while not self._approved.is_set(): - await asyncio.wait( - [asyncio.create_task(self._approved.wait())], - timeout=0.1, - ) if self._cancelled: await self._persist_cancelled() raise asyncio.CancelledError - await asyncio.sleep(0) + 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: @@ -180,6 +207,11 @@ class RunActor: 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) @@ -204,9 +236,12 @@ class RunActor: 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: + except (AdapterError, httpx.HTTPError): continue self._store.add_page(page.url, page.title, page.text) await self._publish( @@ -214,13 +249,12 @@ class RunActor: {"url": page.url, "title": page.title}, ) self.state.bump("pages") - for passage in self._store.passages_for_url(results[0].url if results else ""): - await self._publish( - "passage_extracted", - {"passage_id": passage.passage_id, "url": passage.url}, - ) - self.state.bump("passages") - break + 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) @@ -240,10 +274,11 @@ class RunActor: 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.writer_prompt.format( + 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 "{}", @@ -259,8 +294,8 @@ class RunActor: async def _verify_phase(self) -> None: await self._checkpoint(RunStateName.VERIFYING) - liveness = await liveness_report(self._cited_urls()) 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, @@ -276,8 +311,16 @@ class RunActor: await self._checkpoint(RunStateName.DONE) await self._publish("run_done", {"run_id": self._config.run_id}) - def _cited_urls(self) -> list[str]: - return [page.url for page in self._store.pages()] + 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( diff --git a/backend/src/researchx/methodologies/builtin.py b/backend/src/researchx/methodologies/builtin.py index 267a72b..6ff544d 100644 --- a/backend/src/researchx/methodologies/builtin.py +++ b/backend/src/researchx/methodologies/builtin.py @@ -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] = { diff --git a/backend/src/researchx/methodologies/models.py b/backend/src/researchx/methodologies/models.py index dc56e40..6247ee6 100644 --- a/backend/src/researchx/methodologies/models.py +++ b/backend/src/researchx/methodologies/models.py @@ -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"] \ No newline at end of file diff --git a/backend/src/researchx/trust/ledger.py b/backend/src/researchx/trust/ledger.py index 7128aa5..af1d3fa 100644 --- a/backend/src/researchx/trust/ledger.py +++ b/backend/src/researchx/trust/ledger.py @@ -4,7 +4,7 @@ from datetime import datetime from pydantic import BaseModel from researchx.evidence.store import EvidenceStore, Passage -from researchx.trust.anchoring import anchor_quote, normalize +from researchx.trust.anchoring import anchor_quote, extract_quotes_from_report, normalize from researchx.trust.liveness import LivenessStatus _AUTHORITY_TIERS: dict[str, str] = { @@ -17,9 +17,10 @@ _AUTHORITY_TIERS: dict[str, str] = { def authority_tier(url: str) -> str: - host = re.sub(r"^https?://", "", url).split("/")[0].lower() + host = re.sub(r"^https?://", "", url).split("/")[0].lower().split(":")[0] + labels = host.split(".") for suffix, tier in _AUTHORITY_TIERS.items(): - if host.endswith(suffix): + if labels[-1] == suffix or (len(labels) > 1 and labels[-2] == suffix): return tier return "general" @@ -73,12 +74,23 @@ def build_ledger( 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] @@ -95,25 +107,18 @@ def build_ledger( ) ) for passage in passages: - verdict = anchor_quote(passage.quote, [passage]) - if verdict.matched: - anchored += 1 - elif verdict.weak: - weak += 1 - else: - unanchored += 1 - if passage.url in page.url: - confidence[passage.passage_id] = ( - "corroborated" if _corroborated(passage, evidence) else "single-source" - ) + 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 = max(len(entries), 1) + total = len(entries) + ratio = round(live / total, 4) if total > 0 else 1.0 checklist = VerificationChecklist( run_id=run_id, - citations_checked=len(entries), + citations_checked=total, live=live, stale=stale, dead=dead, @@ -121,7 +126,7 @@ def build_ledger( unanchored_claims=unanchored, weak_matches=weak, confidence_labels=confidence, - liveness_ratio=round(live / total, 4), + liveness_ratio=ratio, ) return CitationLedger( run_id=run_id, generated_at=visited_at, entries=entries, checklist=checklist @@ -138,14 +143,26 @@ def _corroborated(passage: Passage, evidence: EvidenceStore) -> bool: 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}", + ( + 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}", @@ -158,4 +175,7 @@ def export_checklist(ledger: CitationLedger) -> str: 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" \ No newline at end of file diff --git a/backend/src/researchx/trust/liveness.py b/backend/src/researchx/trust/liveness.py index 636a81d..7ea7a4f 100644 --- a/backend/src/researchx/trust/liveness.py +++ b/backend/src/researchx/trust/liveness.py @@ -1,9 +1,9 @@ -import socket from enum import StrEnum -from urllib.parse import urlparse import httpx +from researchx.adapters.url_guard import url_allowed + _LIVENESS_TIMEOUT = 10.0 _MAX_RESPONSE_BYTES = 2 * 1024 * 1024 _MAX_REDIRECTS = 5 @@ -16,55 +16,44 @@ class LivenessStatus(StrEnum): DEAD = "dead" -def _is_private_address(host: str) -> bool: - try: - addr = socket.gethostbyname(host) - except (socket.gaierror, UnicodeError): - return True - import ipaddress - - ip = ipaddress.ip_address(addr) - return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved - - -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 _is_private_address(parsed.hostname) - - async def check_liveness(url: str, client: httpx.AsyncClient | None = None) -> LivenessStatus: - if not url_allowed(url): - return LivenessStatus.DEAD own_client = client is None active: httpx.AsyncClient = client or httpx.AsyncClient( timeout=_LIVENESS_TIMEOUT, - follow_redirects=True, + follow_redirects=False, ) + current_url = url try: - response = await active.get( - url, - headers={"User-Agent": _USER_AGENT}, - ) - content = b"" - async for chunk in response.aiter_bytes(): - content += chunk - if len(content) > _MAX_RESPONSE_BYTES: - break - status = response.status_code + 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() - if 200 <= status < 300: - return LivenessStatus.LIVE - if 300 <= status < 400: - return LivenessStatus.STALE - return LivenessStatus.DEAD async def liveness_report( diff --git a/backend/tests/test_methodologies.py b/backend/tests/test_methodologies.py index 8f62b5d..ea1a653 100644 --- a/backend/tests/test_methodologies.py +++ b/backend/tests/test_methodologies.py @@ -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: diff --git a/backend/tests/test_p0_regressions.py b/backend/tests/test_p0_regressions.py new file mode 100644 index 0000000..fd95b99 --- /dev/null +++ b/backend/tests/test_p0_regressions.py @@ -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="never") + + 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"] \ No newline at end of file -- 2.43.0 From 9c3246be87956fdedb01e32ddf95e8ce806e6db5 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:47:15 +0000 Subject: [PATCH 7/8] =?UTF-8?q?verify(P02):=20passed=20after=20fixes=20?= =?UTF-8?q?=E2=80=94=202=20P0=20+=209=20P1=20resolved=20in-phase,=2011=20P?= =?UTF-8?q?2=20flagged?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ---ci--- phase: 2 milestone: v0.2 status: verify requirements: covered: [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] partial: [] lessons: - "P0s both masked by fake tests: SSRF guard existed but was never wired into the fetch boundary; writer prompts had no evidence slot so the visited-only constraint was vacuous - both fixed with regression pins (evidence-in-prompt recording test, refused-fetch test)" - "Fake-adapter tests mask integration gaps: the three highest-risk behaviors were exactly the untested ones. P4 adversarial suite must include real-boundary probes" - "P2 flags for post-hoc: DNS-rebinding TOCTOU residual risk (getaddrinfo vs connect-time resolution), evidence store not persisted across restarts (renumbered IDs on resume), reflecting step is re-planning (depth>1 loop untested), liveness body accumulation vs HEAD" ---/ci--- Structural: mypy strict 36 files, ruff clean. Behavioral: 108 backend tests (full walk, approval halt, pause/resume mid-run, cancel pre+post approval, restart from AWAITING_APPROVAL and SEARCHING, probe wiring both paths, trust thresholds, SSRF refusals, export gate) + 90 frontend green. REQ-13..24 mapped. Security: SSRF enforced at every fetch path + per-hop redirect guarding; no value leaks (probe persists typed data only). The phase is shippable: both P0s and all blocking P1s resolved with regression coverage. --- .ciagent/tmp/verify-msg.txt | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) create mode 100644 .ciagent/tmp/verify-msg.txt diff --git a/.ciagent/tmp/verify-msg.txt b/.ciagent/tmp/verify-msg.txt new file mode 100644 index 0000000..4862c85 --- /dev/null +++ b/.ciagent/tmp/verify-msg.txt @@ -0,0 +1,24 @@ +verify(P02): passed after fixes — 2 P0 + 9 P1 resolved in-phase, 11 P2 flagged + +---ci--- +phase: 2 +milestone: v0.2 +status: verify +requirements: + covered: [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] + partial: [] +lessons: + - "P0s both masked by fake tests: SSRF guard existed but was never wired into the fetch boundary; writer prompts had no evidence slot so the visited-only constraint was vacuous - both fixed with regression pins (evidence-in-prompt recording test, refused-fetch test)" + - "Fake-adapter tests mask integration gaps: the three highest-risk behaviors were exactly the untested ones. P4 adversarial suite must include real-boundary probes" + - "P2 flags for post-hoc: DNS-rebinding TOCTOU residual risk (getaddrinfo vs connect-time resolution), evidence store not persisted across restarts (renumbered IDs on resume), reflecting step is re-planning (depth>1 loop untested), liveness body accumulation vs HEAD" +---/ci--- + +Structural: mypy strict 36 files, ruff clean. Behavioral: 108 +backend tests (full walk, approval halt, pause/resume mid-run, +cancel pre+post approval, restart from AWAITING_APPROVAL and +SEARCHING, probe wiring both paths, trust thresholds, SSRF +refusals, export gate) + 90 frontend green. REQ-13..24 mapped. +Security: SSRF enforced at every fetch path + per-hop redirect +guarding; no value leaks (probe persists typed data only). The +phase is shippable: both P0s and all blocking P1s resolved with +regression coverage. -- 2.43.0 From f736c0b8b76ea5ef41a437ba87a68e665959bc69 Mon Sep 17 00:00:00 2001 From: coreci Date: Sun, 27 Sep 2026 11:47:24 +0000 Subject: [PATCH 8/8] chore(P02): remove stray verify-msg file ---ci--- phase: 2 milestone: v0.2 status: verify ---/ci--- --- .ciagent/tmp/verify-msg.txt | 24 ------------------------ 1 file changed, 24 deletions(-) delete mode 100644 .ciagent/tmp/verify-msg.txt diff --git a/.ciagent/tmp/verify-msg.txt b/.ciagent/tmp/verify-msg.txt deleted file mode 100644 index 4862c85..0000000 --- a/.ciagent/tmp/verify-msg.txt +++ /dev/null @@ -1,24 +0,0 @@ -verify(P02): passed after fixes — 2 P0 + 9 P1 resolved in-phase, 11 P2 flagged - ----ci--- -phase: 2 -milestone: v0.2 -status: verify -requirements: - covered: [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] - partial: [] -lessons: - - "P0s both masked by fake tests: SSRF guard existed but was never wired into the fetch boundary; writer prompts had no evidence slot so the visited-only constraint was vacuous - both fixed with regression pins (evidence-in-prompt recording test, refused-fetch test)" - - "Fake-adapter tests mask integration gaps: the three highest-risk behaviors were exactly the untested ones. P4 adversarial suite must include real-boundary probes" - - "P2 flags for post-hoc: DNS-rebinding TOCTOU residual risk (getaddrinfo vs connect-time resolution), evidence store not persisted across restarts (renumbered IDs on resume), reflecting step is re-planning (depth>1 loop untested), liveness body accumulation vs HEAD" ----/ci--- - -Structural: mypy strict 36 files, ruff clean. Behavioral: 108 -backend tests (full walk, approval halt, pause/resume mid-run, -cancel pre+post approval, restart from AWAITING_APPROVAL and -SEARCHING, probe wiring both paths, trust thresholds, SSRF -refusals, export gate) + 90 frontend green. REQ-13..24 mapped. -Security: SSRF enforced at every fetch path + per-hop redirect -guarding; no value leaks (probe persists typed data only). The -phase is shippable: both P0s and all blocking P1s resolved with -regression coverage. -- 2.43.0