feat(pipeline): stateful, resumable study store + runner (+ bibliography study)

docs/stateful-study-pipeline.md

@@ -0,0 +1,104 @@
+# Stateful Study Pipeline (design)
+
+## Motivation
+
+Long analyses — for example validating a paper's reported findings against an
+open dataset — tend to run **inline in a single agent turn** as one-shot
+scripts. Two failure modes follow from that:
+
+- **Memory exhaustion** — loading a multi-gigabyte dataset into the agent's
+  process can exceed the container's memory budget and get the process killed
+  mid-analysis.
+- **Loss on restart** — if the container is recreated (deploy, rebuild, crash),
+  an in-flight analysis and its progress are gone, with no way to resume.
+
+Both have the same remedy: **durable, checkpointed analysis state** so an
+analysis resumes from the last completed step after a memory kill *or* a
+container restart, instead of vanishing.
+
+## Hard requirements
+
+1. **Survives container rebuild** — study state lives on the mounted data volume
+   (`HERMES_HOME`/Matilde engagements dir), not in process memory.
+2. **Survives OOM / mid-step failure** — a failed step leaves the study resumable;
+   completed steps are never re-run.
+3. **Stdlib-only core** — mirrors the rest of `matilde_plugin/engine` (pure,
+   injected-I/O, SQLite from stdlib). Heavy scientific deps (mne, etc.) live only
+   in optional step implementations, never in the framework.
+4. **Agent-drivable** — the agent kicks off / advances / inspects a study through
+   tools, so it never has to hold a long analysis in a single turn.
+5. **TDD** — injected I/O, deterministic, matches the existing ~99-test suite.
+
+## Components
+
+### `matilde_plugin/engine/store.py` — `StudyStore` (SQLite)
+
+Persistent store; injected db path (default under the engagements dir). WAL,
+atomic writes. Tables:
+
+- `studies(id, slug UNIQUE, title, status, plan_json, created_at, updated_at, meta_json)`
+  - status: `created | running | done | failed | blocked`
+- `steps(id, study_id, idx, name, status, result_json, error, started_at, finished_at)`
+  - status: `pending | running | done | failed | skipped`
+  - UNIQUE(study_id, name)
+- `artifacts(id, study_id, step_name, path, kind, sha256, bytes, meta_json, created_at)`
+  - references on-disk files (downloads, intermediate `.npy`/`.fif`, plots) — **never** blobs in the DB
+- `findings(id, study_id, step_name, claim, verdict, score, evidence_json, created_at)`
+  - the scientific output: claim + verdict (`supported|refuted|inconclusive`) + evidence
+
+API (all idempotent / upsert where sensible): `create_study`, `get_study`,
+`list_studies`, `add_steps(plan)`, `set_step_status`, `record_step_result`,
+`add_artifact`, `add_finding`, `get_findings`, `study_summary`.
+
+### `matilde_plugin/engine/pipeline.py` — resumable runner
+
+- `Step = (name: str, fn: Callable[[StepContext], StepResult])`
+- `StepContext` carries the store, study_id, results of prior steps, and injected
+  I/O handles (so steps are testable without network/fs side effects).
+- `StepResult` = `{data: dict, artifacts: [...], findings: [...]}`.
+- `run(store, study_id, steps, *, resume=True)`:
+  - for each step in order: if its stored status is `done`, **skip** (resume);
+    else mark `running`, call `fn(ctx)`, persist result + artifacts + findings,
+    mark `done`; on exception mark `failed`, record error, **stop** (study
+    `blocked`/`failed`, resumable later). Returns a per-step summary.
+- `resume(store, study_id, steps)` = `run(..., resume=True)` — continues from the
+  first non-`done` step. Re-running a fully-done study is a no-op.
+- Idempotency contract: step fns must be safe to re-run from scratch (the runner
+  only guarantees done-steps are skipped, not partial-step rollback).
+- Plan evolution: the runner only iterates the steps passed to it. If a step is
+  dropped from a later plan, its stored row is left as-is (orphaned, never
+  re-run) rather than reconciled. Fine for fixed-plan studies; a study whose
+  plan changes between runs should reconcile stored-vs-passed steps explicitly.
+
+### Agent tools (`matilde_plugin/tools.py`)
+
+Match the existing tool-dict + handler conventions in this file:
+
+- `matilde_study_create(slug, title, plan)` → `{study_id}` (plan = ordered step names + params)
+- `matilde_study_run(study_id)` → advances pending steps; returns status + per-step summary (so an OOM/restart mid-run is resumed by simply calling this again)
+- `matilde_study_status(study_id)` → full status + findings + artifacts
+- `matilde_study_list()` → recent studies
+
+### First concrete pipeline (this PR — lightweight, proves the framework, no heavy deps)
+
+**Bibliography-validation study**: steps `parse_refs → verify_each → summarize`,
+reusing the existing citations engine. `verify_each` records one finding per
+reference. Demonstrated resumability: run with the verifier failing on ref #3,
+assert the study is `blocked` with refs 1–2 `done`; fix the injected verifier;
+`resume`; assert it completes from ref #3 without re-verifying 1–2.
+
+### Follow-up (separate PR — heavy data analysis)
+
+**Data-validation study** (e.g. checking a reported evoked-response peak in an
+open MEG dataset): steps `fetch_dataset → preprocess → epoch → evoked →
+validate_finding`, each **memory-bounded** (chunked/streamed, intermediates
+checkpointed as artifacts) so a memory kill resumes from the last completed step.
+Requires deciding the agent container memory budget and where heavy scientific
+deps (e.g. mne) are installed. Out of scope for the framework PR.
+
+## Testing
+
+stdlib + injected I/O. Cover: store CRUD + status transitions + persistence
+across a fresh `StudyStore` instance (simulates restart); runner skip-done /
+stop-on-failure / resume; the bibliography study end-to-end incl. the
+fail-then-resume scenario; the agent tools' arg coercion + outputs.

matilde_plugin/__init__.py

@@ -55,6 +55,10 @@
 OPENNEURO_SEARCH_SCHEMA = _tools_mod.OPENNEURO_SEARCH_SCHEMA
 OPENNEURO_FILES_SCHEMA = _tools_mod.OPENNEURO_FILES_SCHEMA
 FETCH_FULLTEXT_SCHEMA = _tools_mod.FETCH_FULLTEXT_SCHEMA
+STUDY_CREATE_SCHEMA = _tools_mod.STUDY_CREATE_SCHEMA
+STUDY_RUN_SCHEMA = _tools_mod.STUDY_RUN_SCHEMA
+STUDY_STATUS_SCHEMA = _tools_mod.STUDY_STATUS_SCHEMA
+STUDY_LIST_SCHEMA = _tools_mod.STUDY_LIST_SCHEMA
 _check_available = _tools_mod._check_available
 _handle_verify_citation = _tools_mod._handle_verify_citation
 _handle_verify_bibliography = _tools_mod._handle_verify_bibliography
@@ -63,6 +67,10 @@
 _handle_openneuro_search = _tools_mod._handle_openneuro_search
 _handle_openneuro_list_files = _tools_mod._handle_openneuro_list_files
 _handle_fetch_fulltext = _tools_mod._handle_fetch_fulltext
+_handle_study_create = _tools_mod._handle_study_create
+_handle_study_run = _tools_mod._handle_study_run
+_handle_study_status = _tools_mod._handle_study_status
+_handle_study_list = _tools_mod._handle_study_list
 
 # ---------------------------------------------------------------------------
 # Tool registry — (tool_name, schema, handler, emoji)
@@ -75,6 +83,10 @@
     ("matilde_openneuro_search", OPENNEURO_SEARCH_SCHEMA, _handle_openneuro_search, "🔎"),
     ("matilde_openneuro_list_files", OPENNEURO_FILES_SCHEMA, _handle_openneuro_list_files, "🗂"),
     ("matilde_fetch_fulltext", FETCH_FULLTEXT_SCHEMA, _handle_fetch_fulltext, "📄"),
+    ("matilde_study_create", STUDY_CREATE_SCHEMA, _handle_study_create, "🧪"),
+    ("matilde_study_run", STUDY_RUN_SCHEMA, _handle_study_run, "▶"),
+    ("matilde_study_status", STUDY_STATUS_SCHEMA, _handle_study_status, "📊"),
+    ("matilde_study_list", STUDY_LIST_SCHEMA, _handle_study_list, "🗒"),
 )
 
 

matilde_plugin/engine/bibliography_study.py

@@ -0,0 +1,130 @@
+"""Bibliography-validation study — the first concrete pipeline.
+
+Three steps, reusing the existing parsing + citations engine:
+
+  1. ``parse_refs``  — parse a BibTeX blob into ``Reference`` objects.
+  2. ``verify_each`` — verify each reference, recording one *finding* per ref.
+  3. ``summarize``   — roll the findings up into a per-verdict tally.
+
+This is deliberately lightweight (no heavy scientific deps) and exists to *prove
+the framework*: it is the worked example for resumability. ``verify_each`` is
+written to be safe to re-run from scratch — on resume it skips references that
+already have a persisted finding — so when the verifier fails partway through, a
+later ``resume`` continues from the unfinished reference without re-verifying the
+ones that already passed. (The runner only skips done *steps*; mid-step
+idempotency is the step's own responsibility — see ``pipeline`` docstring.)
+
+The verifier is **injected**: ``verifier(ref) -> {claim, verdict, score,
+evidence}``. Production passes :func:`default_verifier` (which wraps the real
+citations engine); tests pass a fake that can force a failure on a chosen ref.
+"""
+from __future__ import annotations
+
+from typing import Callable, List, Optional
+
+from .citations import Reference, verify_reference
+from .parsing import parse_bibtex
+from .pipeline import Step, StepContext, StepResult
+
+VerifierFn = Callable[[Reference], dict]
+
+# Map the citation engine's verdicts onto the finding vocabulary
+# (supported | refuted | inconclusive) the study records.
+_VERDICT_MAP = {
+    "verified": "supported",
+    "warnings": "inconclusive",
+    "unverifiable": "inconclusive",
+    "not_found": "refuted",
+    "retracted": "refuted",
+}
+
+
+def _ref_key(ref: Reference) -> str:
+    """A stable identifier for a reference (for resume de-duplication + claims)."""
+    return ref.doi or ref.title or ref.raw[:80] or "(unidentified)"
+
+
+def default_verifier(ref: Reference) -> dict:
+    """Verify *ref* with the real citations engine (network). Injected in prod."""
+    result = verify_reference(ref)
+    verdict = _VERDICT_MAP.get(result.verdict, "inconclusive")
+    return {
+        "claim": f"reference is valid: {_ref_key(ref)}",
+        "verdict": verdict,
+        "score": result.score,
+        "evidence": {
+            "citation_verdict": result.verdict,
+            "axes": {
+                "existence": result.existence.status,
+                "metadata_match": result.metadata_match.status,
+                "retraction": result.retraction.status,
+                "url_liveness": result.url_liveness.status,
+            },
+        },
+    }
+
+
+def _parse_refs_step(bibtex: str) -> Step:
+    def fn(ctx: StepContext) -> StepResult:
+        refs = parse_bibtex(bibtex)
+        return StepResult(data={
+            "count": len(refs),
+            "refs": [r.raw or _ref_key(r) for r in refs],
+        })
+    return Step(name="parse_refs", fn=fn)
+
+
+def _verify_each_step(bibtex: str, verifier: VerifierFn) -> Step:
+    def fn(ctx: StepContext) -> StepResult:
+        refs = parse_bibtex(bibtex)
+        # Resume-safe: skip refs that already have a persisted finding.
+        already = {f["evidence"].get("ref_key")
+                   for f in ctx.store.get_findings(ctx.study_id)
+                   if f["step_name"] == "verify_each"}
+        for ref in refs:
+            key = _ref_key(ref)
+            if key in already:
+                continue
+            v = verifier(ref)  # may raise -> step fails, study resumable
+            evidence = dict(v.get("evidence") or {})
+            evidence["ref_key"] = key
+            # Persist immediately so a later failure doesn't lose this ref.
+            ctx.store.add_finding(
+                ctx.study_id, "verify_each",
+                claim=v.get("claim", f"reference: {key}"),
+                verdict=v.get("verdict", "inconclusive"),
+                score=v.get("score"),
+                evidence=evidence)
+        return StepResult(data={"verified": len(refs)})
+    return Step(name="verify_each", fn=fn)
+
+
+def _summarize_step() -> Step:
+    def fn(ctx: StepContext) -> StepResult:
+        findings = [f for f in ctx.store.get_findings(ctx.study_id)
+                    if f["step_name"] == "verify_each"]
+        counts: dict = {}
+        for f in findings:
+            counts[f["verdict"]] = counts.get(f["verdict"], 0) + 1
+        return StepResult(data={
+            "total": len(findings),
+            "supported": counts.get("supported", 0),
+            "refuted": counts.get("refuted", 0),
+            "inconclusive": counts.get("inconclusive", 0),
+            "by_verdict": counts,
+        })
+    return Step(name="summarize", fn=fn)
+
+
+def build_steps(bibtex: str, verifier: Optional[VerifierFn] = None) -> List[Step]:
+    """Build the ``parse_refs -> verify_each -> summarize`` steps for *bibtex*.
+
+    Pass *verifier* to inject the per-reference check (defaults to the real
+    citations engine). The returned steps are fed to ``pipeline.run`` / ``resume``.
+    """
+    verifier = verifier or default_verifier
+    return [
+        _parse_refs_step(bibtex),
+        _verify_each_step(bibtex, verifier),
+        _summarize_step(),
+    ]

matilde_plugin/engine/pipeline.py

@@ -0,0 +1,119 @@
+"""Resumable study runner — checkpoints each step into a ``StudyStore``.
+
+A study is an ordered list of :class:`Step`\\s. The runner walks them in order:
+a step whose stored status is already ``done`` is **skipped** (resume); otherwise
+it is marked ``running``, its ``fn`` is called, its result + artifacts + findings
+are persisted, and it is marked ``done``. On exception the step is marked
+``failed`` (with the error), the study is marked ``blocked``, and the run **stops**
+— leaving everything before it ``done`` and the study resumable later. This is the
+durability contract: an OOM or container rebuild mid-run is recovered by simply
+calling ``run``/``resume`` again on the same store.
+
+Pure and injected: a :class:`Step`'s ``fn`` receives a :class:`StepContext`
+carrying the store, the study id, and the results of prior steps, so steps are
+unit-testable with in-memory fakes (no network/fs needed).
+
+Idempotency contract: the runner only guarantees *done steps are skipped*, not
+partial-step rollback. A step that does internal sub-work (e.g. verifying N refs)
+must itself be safe to re-run from scratch — typically by checking the store for
+what it already persisted and skipping that (see ``bibliography_study``).
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+from .store import StudyStore
+
+
+@dataclass
+class StepResult:
+    """What a step produces. All optional — a step may only record findings."""
+
+    data: dict = field(default_factory=dict)
+    artifacts: List[dict] = field(default_factory=list)
+    findings: List[dict] = field(default_factory=list)
+
+
+@dataclass
+class StepContext:
+    """Handle passed to a step's ``fn``: the store, the study, and prior results."""
+
+    store: StudyStore
+    study_id: int
+    step_name: str
+    results: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class Step:
+    name: str
+    fn: Callable[[StepContext], StepResult]
+
+
+def run(store: StudyStore, study_id: int, steps: List[Step], *,
+        resume: bool = True) -> dict:
+    """Execute *steps* in order against *study_id*, persisting after each.
+
+    Returns a summary dict: ``{study_id, status, steps: [...], failed_step}``.
+    Done steps are skipped when ``resume`` is True (the default).
+    """
+    store.add_steps(study_id, [s.name for s in steps])
+
+    # Carry forward results of steps already done (so a resumed step can read them).
+    prior: Dict[str, Any] = {}
+    for s in store.get_steps(study_id):
+        if s["status"] == "done" and s["result"] is not None:
+            prior[s["name"]] = s["result"]
+
+    store.set_study_status(study_id, "running")
+    per_step: List[dict] = []
+    failed_step: Optional[str] = None
+
+    for step in steps:
+        stored = store.get_step(study_id, step.name)
+        if resume and stored is not None and stored["status"] == "done":
+            per_step.append({"name": step.name, "status": "done", "skipped": True})
+            continue
+
+        store.set_step_status(study_id, step.name, "running")
+        ctx = StepContext(store=store, study_id=study_id,
+                          step_name=step.name, results=dict(prior))
+        try:
+            result = step.fn(ctx)
+        except Exception as exc:  # stop, leave resumable
+            store.set_step_status(study_id, step.name, "failed",
+                                  error=f"{type(exc).__name__}: {exc}")
+            store.set_study_status(study_id, "blocked")
+            per_step.append({"name": step.name, "status": "failed",
+                             "error": f"{type(exc).__name__}: {exc}"})
+            failed_step = step.name
+            return {"study_id": study_id, "status": "blocked",
+                    "steps": per_step, "failed_step": failed_step}
+
+        result = result or StepResult()
+        store.record_step_result(study_id, step.name, result.data)
+        for art in result.artifacts:
+            store.add_artifact(
+                study_id, step.name,
+                path=art.get("path", ""), kind=art.get("kind", ""),
+                sha256=art.get("sha256", ""), bytes=art.get("bytes"),
+                meta=art.get("meta"))
+        for fnd in result.findings:
+            store.add_finding(
+                study_id, step.name,
+                claim=fnd.get("claim", ""), verdict=fnd.get("verdict", ""),
+                score=fnd.get("score"), evidence=fnd.get("evidence"))
+        store.set_step_status(study_id, step.name, "done")
+        prior[step.name] = result.data
+        per_step.append({"name": step.name, "status": "done"})
+
+    store.set_study_status(study_id, "done")
+    return {"study_id": study_id, "status": "done",
+            "steps": per_step, "failed_step": None}
+
+
+def resume(store: StudyStore, study_id: int, steps: List[Step]) -> dict:
+    """Continue a study from its first non-``done`` step. A fully-done study is a
+    no-op. Equivalent to ``run(..., resume=True)``."""
+    return run(store, study_id, steps, resume=True)