From fbac3380b339f9e23a45e3761d90e2a02177893c Mon Sep 17 00:00:00 2001 From: Spyros Date: Fri, 24 Jul 2026 20:31:21 +0100 Subject: [PATCH] =?UTF-8?q?Add=20PoC=20OIE=20orchestrator=20for=20Modules?= =?UTF-8?q?=20A=E2=86=92B=E2=86=92C.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scaffold a demo runner that calls Module B/C entry points when present and records TODO stages for Module A's missing harvester entry and unmerged hooks, without modifying the GSoC module packages. Co-authored-by: Cursor --- Makefile | 5 + application/tests/oie_orchestrator_test.py | 94 ++++++ .../utils/oie_orchestrator/__init__.py | 21 ++ .../utils/oie_orchestrator/pipeline.py | 298 ++++++++++++++++++ scripts/run_oie_demo_pipeline.py | 97 ++++++ 5 files changed, 515 insertions(+) create mode 100644 application/tests/oie_orchestrator_test.py create mode 100644 application/utils/oie_orchestrator/__init__.py create mode 100644 application/utils/oie_orchestrator/pipeline.py create mode 100644 scripts/run_oie_demo_pipeline.py diff --git a/Makefile b/Makefile index 9534605f3..7b430039c 100644 --- a/Makefile +++ b/Makefile @@ -157,6 +157,11 @@ migrate-upgrade: export FLASK_APP="$(CURDIR)/cre.py" flask db upgrade +# PoC: Module A→B→C orchestrator (skips A by default; see scripts/run_oie_demo_pipeline.py) +oie-demo-pipeline: + [ -d "./venv" ] && . ./venv/bin/activate &&\ + python scripts/run_oie_demo_pipeline.py --skip-b --skip-c $(OIE_DEMO_ARGS) + alembic-guardrail: [ -d "./venv" ] && . ./venv/bin/activate &&\ python scripts/check_alembic_revision_guardrail.py diff --git a/application/tests/oie_orchestrator_test.py b/application/tests/oie_orchestrator_test.py new file mode 100644 index 000000000..4e3789964 --- /dev/null +++ b/application/tests/oie_orchestrator_test.py @@ -0,0 +1,94 @@ +"""Tests for the OIE PoC orchestrator (no Module A/B/C patches).""" + +from __future__ import annotations + +import json +import unittest +from types import SimpleNamespace + +from application.utils.oie_orchestrator.pipeline import run_oie_demo_pipeline + + +class OieOrchestratorTest(unittest.TestCase): + def test_default_skips_a_and_records_b_todo_or_ok(self) -> None: + """On main without #989, B is todo; with inject, B is ok.""" + result = run_oie_demo_pipeline( + cache_file="sqlite:///:memory:", + pipeline_run_id="run-test-1", + skip_a=True, + skip_b=False, + skip_c=True, + dry_run=True, + ) + self.assertEqual(result.run_id, "run-test-1") + by_name = {s.name: s for s in result.stages} + self.assertEqual(by_name["module_a_harvester"].status, "skipped") + self.assertIn(by_name["module_b_noise_filter"].status, ("todo", "ok", "error")) + self.assertEqual(by_name["module_c_librarian"].status, "skipped") + + def test_run_a_without_entry_is_todo(self) -> None: + result = run_oie_demo_pipeline( + cache_file="sqlite:///:memory:", + pipeline_run_id="run-a", + skip_a=False, + skip_b=True, + skip_c=True, + ) + stage = result.stages[0] + self.assertEqual(stage.name, "module_a_harvester") + self.assertEqual(stage.status, "todo") + self.assertIn("TODO:", stage.detail) + self.assertIn("run_harvester", stage.detail) + + def test_injected_b_and_c_entry_points(self) -> None: + calls: list[str] = [] + + def fake_b(session: object, run_id: str, dry_run: bool = False) -> object: + calls.append(f"b:{run_id}:{dry_run}") + return SimpleNamespace( + to_json=lambda: json.dumps( + {"run_id": run_id, "read": 2, "inserted": 1, "dry_run": dry_run} + ) + ) + + def fake_c( + cache_file: str, dry_run: bool = True, source_jsonl: str | None = None + ) -> None: + calls.append(f"c:{dry_run}:{source_jsonl}") + + result = run_oie_demo_pipeline( + cache_file="sqlite:///:memory:", + pipeline_run_id="run-inject", + skip_a=True, + skip_b=False, + skip_c=False, + dry_run=True, + librarian_source_jsonl="fixture.jsonl", + run_noise_filter_fn=fake_b, + run_librarian_fn=fake_c, + ) + by_name = {s.name: s for s in result.stages} + self.assertEqual(by_name["module_b_noise_filter"].status, "ok") + self.assertEqual(by_name["module_b_noise_filter"].summary["inserted"], 1) + self.assertEqual(by_name["module_c_librarian"].status, "ok") + self.assertEqual(calls, ["b:run-inject:True", "c:True:fixture.jsonl"]) + self.assertTrue(result.to_dict()["ok"]) + + def test_b_error_surfaces_without_raising(self) -> None: + def boom(*_a: object, **_k: object) -> object: + raise RuntimeError("llm down") + + result = run_oie_demo_pipeline( + cache_file="sqlite:///:memory:", + pipeline_run_id="run-err", + skip_a=True, + skip_c=True, + run_noise_filter_fn=boom, + ) + self.assertEqual(result.stages[1].status, "error") + self.assertIn("llm down", result.stages[1].detail) + self.assertFalse(result.to_dict()["ok"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/oie_orchestrator/__init__.py b/application/utils/oie_orchestrator/__init__.py new file mode 100644 index 000000000..7c57a0c9f --- /dev/null +++ b/application/utils/oie_orchestrator/__init__.py @@ -0,0 +1,21 @@ +"""PoC orchestrator for the OpenCRE Information Extraction (OIE) pipeline. + +Wires Module A (Harvester) → Module B (Noise Filter) → Module C (Librarian) +for a single ``pipeline_run_id`` in a demo/local environment. + +Module entry points are invoked only when present on the branch; missing +hooks are recorded as ``todo`` / ``skipped`` stages rather than patched +inside Modules A/B/C. See ``pipeline.py`` for the per-module TODO list. +""" + +from application.utils.oie_orchestrator.pipeline import ( + OrchestratorResult, + StageResult, + run_oie_demo_pipeline, +) + +__all__ = [ + "OrchestratorResult", + "StageResult", + "run_oie_demo_pipeline", +] diff --git a/application/utils/oie_orchestrator/pipeline.py b/application/utils/oie_orchestrator/pipeline.py new file mode 100644 index 000000000..fb6a0ed49 --- /dev/null +++ b/application/utils/oie_orchestrator/pipeline.py @@ -0,0 +1,298 @@ +"""OIE PoC orchestrator — A → B → C for one ``pipeline_run_id``. + +Designed for a local/demo environment. Does not modify Module A/B/C packages; +it only *calls* their public entry points when importable, and leaves explicit +``TODO:`` markers where a callable is missing or not yet on ``main``. + +Entry-point status (as of 2026-07-24 PR survey): + +* Module A (tip #987 ``week_5-clean``): **no** top-level entry. Components only + (config, git client, change detector, file filter, diff retrieve/parse/ + normalize). No CLI flag, no ``run_harvester``, no ``harvest_input`` writer, + no ChangeRecord/JSONL emitter. +* Module B (tip #989 ``module_b_w5``): **yes** — + ``noise_filter.pipeline.run_noise_filter(session, pipeline_run_id, …)`` and + ``cre.py --run_noise_filter --run_id …``. Not on ``main`` until #989 merges. +* Module C (``main`` + tip #991): **yes (partial)** — + ``cre_main.run_librarian(cache, dry_run, source_jsonl)`` on ``main`` (C.0→C.2 + dry-run over a JSONL fixture). ``LibrarianPipeline.run(at=…)`` on #991 + (C.0→C.4 envelopes, still dry-run / no graph write). DB-backed + ``knowledge_queue`` source is W8. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from dataclasses import asdict, dataclass, field +from typing import Any, Callable, Dict, List, Optional + +logger = logging.getLogger(__name__) + +DEFAULT_LIBRARIAN_SOURCE = ( + "application/tests/librarian/fixtures/sample_knowledge_queue.jsonl" +) + + +@dataclass +class StageResult: + """Outcome of one orchestrator stage (module A, B, or C).""" + + name: str + status: str # ok | skipped | todo | error + detail: str + summary: Optional[Dict[str, Any]] = None + + +@dataclass +class OrchestratorResult: + """Full A→B→C run summary (JSON-serializable for PoC logging).""" + + run_id: str + dry_run: bool + stages: List[StageResult] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "run_id": self.run_id, + "dry_run": self.dry_run, + "stages": [asdict(s) for s in self.stages], + "ok": all(s.status in ("ok", "skipped", "todo") for s in self.stages), + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), indent=2) + + +def _stage_module_a(run_id: str, *, skip: bool) -> StageResult: + """Module A (Harvester) — no callable entry point yet. + + TODO: Once Module A lands a top-level entry, call it here. Needed shape + (either is fine for the orchestrator): + + # library + run_harvester(pipeline_run_id=run_id, ...) -> RunSummary + # must write ChangeRecord payloads into harvest_input (status=pending) + + # or CLI + python cre.py --run_harvester --run_id [--cache_file ...] + + Until then, Module B has nothing to read from ``harvest_input``. For a PoC + that only demos B/C, pass ``skip_a=True`` and seed ``harvest_input`` (or a + librarian JSONL fixture) out of band. + """ + if skip: + return StageResult( + name="module_a_harvester", + status="skipped", + detail="skip_a=True; harvester not invoked", + ) + return StageResult( + name="module_a_harvester", + status="todo", + detail=( + "TODO: Module A has no orchestrator entry point yet " + "(stacked PRs #920→#987). Need run_harvester / --run_harvester that " + f"writes harvest_input rows for pipeline_run_id={run_id!r}. " + "Do not patch Module A from this orchestrator — land the entry in " + "the harvester package, then wire the call here." + ), + ) + + +def _stage_module_b( + run_id: str, + cache_file: str, + *, + skip: bool, + dry_run: bool, + run_noise_filter_fn: Optional[Callable[..., Any]] = None, +) -> StageResult: + """Module B (Noise Filter) — call ``run_noise_filter`` when importable. + + TODO: Land / merge Module B PR #989 so ``application.utils.noise_filter.pipeline`` + and the ``harvest_input`` / ``knowledge_queue`` migration exist on ``main``. + TODO: Until Module A writes ``harvest_input``, seed pending rows for + ``pipeline_run_id`` manually (or keep skip_b and demo C from a JSONL fixture). + """ + if skip: + return StageResult( + name="module_b_noise_filter", + status="skipped", + detail="skip_b=True; noise filter not invoked", + ) + + injected = run_noise_filter_fn is not None + fn = run_noise_filter_fn + if fn is None: + try: + from application.utils.noise_filter.pipeline import run_noise_filter + + fn = run_noise_filter + except ImportError: + return StageResult( + name="module_b_noise_filter", + status="todo", + detail=( + "TODO: Module B orchestrator entry not on this branch. " + "Merge PR #989 (run_noise_filter + harvest_input/knowledge_queue), " + "then this stage will call " + "noise_filter.pipeline.run_noise_filter(session, " + f"pipeline_run_id={run_id!r}, dry_run={dry_run}). " + "CLI equivalent after #989: " + f"python cre.py --run_noise_filter --run_id {run_id} " + "[--noise_filter_dry_run]." + ), + ) + + try: + session: Any = None + if not injected: + from application import sqla # type: ignore[attr-defined] + from application.cmd.cre_main import db_connect + + db_connect(cache_file) + session = sqla.session + summary = fn(session, run_id, dry_run=dry_run) + summary_dict: Dict[str, Any] + if hasattr(summary, "to_json"): + summary_dict = json.loads(summary.to_json()) + elif hasattr(summary, "__dict__"): + summary_dict = dict(summary.__dict__) + else: + summary_dict = {"raw": str(summary)} + return StageResult( + name="module_b_noise_filter", + status="ok", + detail=f"run_noise_filter completed for run_id={run_id!r}", + summary=summary_dict, + ) + except Exception as exc: # noqa: BLE001 — PoC boundary; report and continue + logger.exception("Module B stage failed") + return StageResult( + name="module_b_noise_filter", + status="error", + detail=f"run_noise_filter failed: {exc}", + ) + + +def _stage_module_c( + run_id: str, + cache_file: str, + *, + skip: bool, + dry_run: bool, + source_jsonl: Optional[str], + run_librarian_fn: Optional[Callable[..., Any]] = None, +) -> StageResult: + """Module C (Librarian) — prefer CLI dry-run entry on ``main``. + + Available now on ``main``: + cre_main.run_librarian(cache_file, dry_run=True, source_jsonl=…) + # CLI: python cre.py --librarian_dry_run [--librarian_source PATH] + + TODO: After Module C #991 merges, prefer + LibrarianPipeline(source, retriever, reranker, scaler, + threshold=…, pipeline_run_id=run_id).run(at=…) + and emit LinkProposal | ReviewItem envelopes. Keep dry-run until W8 + graph / queue write-back exists. + TODO: Replace FixtureKnowledgeSource / JSONL with a DB-backed + KnowledgeSource that polls ``knowledge_queue`` where + ``pipeline_run_id == run_id`` and ``consumed_at IS NULL`` (Module C W8). + TODO: cre_main.run_librarian on tip #991 still uses the W3 C.0→C.2 path; + wire it to LibrarianPipeline when that glue is ready — do not fork + Module C inside this orchestrator. + """ + if skip: + return StageResult( + name="module_c_librarian", + status="skipped", + detail="skip_c=True; librarian not invoked", + ) + + jsonl = source_jsonl or DEFAULT_LIBRARIAN_SOURCE + fn = run_librarian_fn + if fn is None: + try: + from application.utils.librarian.pipeline import LibrarianPipeline + + # TODO: Build a real LibrarianPipeline (retriever/reranker/scaler + + # DB KnowledgeSource) once #991 is on main and a component factory + # exists. Until then fall through to run_librarian below. + _ = LibrarianPipeline # import proves the symbol exists + logger.info( + "LibrarianPipeline is importable (Module C #991+); " + "orchestrator still uses run_librarian until a production " + "component factory is available (TODO)." + ) + except ImportError: + pass + + from application.cmd.cre_main import run_librarian + + fn = run_librarian + + try: + # run_librarian ignores pipeline_run_id today (fixture stream); recorded + # for correlation in the orchestrator summary. + fn(cache_file, dry_run=dry_run, source_jsonl=jsonl) + return StageResult( + name="module_c_librarian", + status="ok", + detail=( + f"run_librarian completed (dry_run={dry_run}, source={jsonl}, " + f"pipeline_run_id={run_id!r} for correlation only)" + ), + summary={"source_jsonl": jsonl, "pipeline_run_id": run_id}, + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Module C stage failed") + return StageResult( + name="module_c_librarian", + status="error", + detail=f"run_librarian failed: {exc}", + ) + + +def run_oie_demo_pipeline( + *, + cache_file: str, + pipeline_run_id: Optional[str] = None, + skip_a: bool = True, + skip_b: bool = False, + skip_c: bool = False, + librarian_source_jsonl: Optional[str] = None, + dry_run: bool = True, + run_noise_filter_fn: Optional[Callable[..., Any]] = None, + run_librarian_fn: Optional[Callable[..., Any]] = None, +) -> OrchestratorResult: + """Run the PoC A→B→C sequence for one ``pipeline_run_id``. + + Defaults skip Module A (no entry point) and run B/C in dry-run-friendly + mode. Inject ``run_*_fn`` in tests to avoid hitting the real DB/LLM. + """ + run_id = (pipeline_run_id or "").strip() or str(uuid.uuid4()) + result = OrchestratorResult(run_id=run_id, dry_run=dry_run) + + result.stages.append(_stage_module_a(run_id, skip=skip_a)) + result.stages.append( + _stage_module_b( + run_id, + cache_file, + skip=skip_b, + dry_run=dry_run, + run_noise_filter_fn=run_noise_filter_fn, + ) + ) + result.stages.append( + _stage_module_c( + run_id, + cache_file, + skip=skip_c, + dry_run=dry_run, + source_jsonl=librarian_source_jsonl, + run_librarian_fn=run_librarian_fn, + ) + ) + return result diff --git a/scripts/run_oie_demo_pipeline.py b/scripts/run_oie_demo_pipeline.py new file mode 100644 index 000000000..35eaf8d61 --- /dev/null +++ b/scripts/run_oie_demo_pipeline.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""CLI for the OIE PoC orchestrator (Module A → B → C). + +Examples (local / demo): + + # Librarian-only dry-run (works on main today with fixture JSONL) + python scripts/run_oie_demo_pipeline.py \\ + --cache_file "$DATABASE_URL" \\ + --skip-a --skip-b \\ + --librarian-source application/tests/librarian/fixtures/sample_knowledge_queue.jsonl + + # Full sequence once Module B #989 (and ideally Module A entry) are available + python scripts/run_oie_demo_pipeline.py \\ + --cache_file "$DATABASE_URL" \\ + --run-id demo-2026-07-24 \\ + --skip-a + +See application/utils/oie_orchestrator/pipeline.py for per-module TODO markers. +Does not modify Modules A/B/C — only calls their entry points when present. +""" + +from __future__ import annotations + +import argparse +import logging +import os +import sys + +# Allow running directly as `python scripts/run_oie_demo_pipeline.py`. +_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _REPO_ROOT not in sys.path: + sys.path.insert(0, _REPO_ROOT) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="PoC orchestrator: Module A (harvester) → B (noise filter) → C (librarian)." + ) + parser.add_argument( + "--cache_file", + default=os.environ.get("DATABASE_URL", "sqlite:////tmp/cre_oie_demo.db"), + help="SQLAlchemy DB URL (same as cre.py --cache_file)", + ) + parser.add_argument( + "--run-id", + default="", + help="pipeline_run_id shared across stages (default: random UUID)", + ) + parser.add_argument( + "--skip-a", + action="store_true", + default=True, + help="skip Module A (default: true — no harvester entry point yet)", + ) + parser.add_argument( + "--run-a", + action="store_true", + help="attempt Module A (records TODO until harvester entry lands)", + ) + parser.add_argument("--skip-b", action="store_true", help="skip Module B") + parser.add_argument("--skip-c", action="store_true", help="skip Module C") + parser.add_argument( + "--librarian-source", + default=None, + help="knowledge_queue JSONL for Module C (fixture until DB source / W8)", + ) + parser.add_argument( + "--write", + action="store_true", + help="disable dry_run where a module supports writes (B queue / C links). " + "C still cannot write links pre-W8.", + ) + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(levelname)s %(name)s: %(message)s", + ) + + from application.utils.oie_orchestrator import run_oie_demo_pipeline + + result = run_oie_demo_pipeline( + cache_file=args.cache_file, + pipeline_run_id=args.run_id or None, + skip_a=not args.run_a, + skip_b=args.skip_b, + skip_c=args.skip_c, + librarian_source_jsonl=args.librarian_source, + dry_run=not args.write, + ) + print(result.to_json()) + return 0 if result.to_dict()["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main())