diff --git a/.github/workflows/librarian_regression.yml b/.github/workflows/librarian_regression.yml new file mode 100644 index 000000000..509f5f3d5 --- /dev/null +++ b/.github/workflows/librarian_regression.yml @@ -0,0 +1,67 @@ +name: Librarian Regression Gate +# Module C's accuracy gate. The unit suite proves the code runs; this proves the +# pipeline still *decides* correctly over the golden set. +# +# Deliberately hermetic: no DB, no API key, no model download. The semantic +# reports (C.1 recall, C.2 top-1, the C.3 ECE gate, C.4 decision accuracy) need +# live CRE vectors and are measured locally — seeding a candidate pool from +# golden text offline is the leakage the hub firewall exists to strip, so a CI +# job that "measured" them would be measuring nothing. +# +# What this gate does catch is the failure that actually bit us: a Module B +# schema change silently rejecting every row at the C.0 boundary, which used to +# leave the explicit gate with nothing to count and still exit 0. + +on: + pull_request: + paths: + - 'application/utils/librarian/**' + - 'application/tests/librarian/**' + - 'scripts/evaluate_librarian.py' + - '.github/workflows/librarian_regression.yml' + push: + branches: [main] + paths: + - 'application/utils/librarian/**' + - 'application/tests/librarian/**' + - 'scripts/evaluate_librarian.py' + +permissions: + contents: read + +jobs: + regression: + name: Librarian Regression Gate + runs-on: ubuntu-latest + steps: + - name: Check out code + uses: actions/checkout@v4 + with: + # This job only reads the tree; leaving the token in .git/config would + # expose it to anything the test run executes. + persist-credentials: false + + - uses: actions/setup-python@v5 + with: + python-version: '3.12.3' + + - name: Install python dependencies + # Installed straight into the runner's interpreter rather than through + # `make install-python`: that target builds a venv (needing an apt + # `virtualenv` the other workflows install first) and then runs + # `playwright install`, neither of which a hermetic librarian run uses. + # The steps below call `python` directly, so a venv the job never + # activates would leave them running against a bare interpreter. + run: | + pip install --upgrade pip setuptools + pip install -r requirements-dev.txt + + - name: Librarian unit suite + run: python -m unittest discover -s application/tests/librarian -p '*_test.py' -t . + + - name: Golden-set decision gate + # Non-zero on: a failed explicit-slice gate (C.0.5 must be 100%), or a + # C.0 boundary that rejected the whole dataset. + run: | + python scripts/evaluate_librarian.py \ + --dataset application/tests/librarian/fixtures/golden_dataset.json diff --git a/.gitignore b/.gitignore index 26b6ceb61..d442600cf 100644 --- a/.gitignore +++ b/.gitignore @@ -69,6 +69,8 @@ standards_cache.sqlite !AGENTS.md !docs/faq.md !docs/Mid_eval_blog_gsoc2026/module_B_mideval_blog.md +!application/utils/librarian/README.md +!docs/gsoc_2026_module_c/*.md ### Dev DBDumps *.sql diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index e9872e752..5f9e8575d 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -1032,11 +1032,23 @@ def run(args: argparse.Namespace) -> None: # pragma: no cover if args.upstream_sync: download_graph_from_upstream(args.cache_file) if args.run_librarian or args.librarian_dry_run: - run_librarian( - cache_file=args.cache_file, - dry_run=args.librarian_dry_run or not args.run_librarian, - source_jsonl=args.librarian_source, - ) + # --run_id selects the live path: drain Module B's knowledge_queue for + # that run. Without it, the fixture walk-through stays the default so + # the pre-W8 command keeps behaving the way it always has. + run_id = (getattr(args, "run_id", "") or "").strip() + if run_id: + run_librarian_live( + cache_file=args.cache_file, + pipeline_run_id=run_id, + dry_run=args.librarian_dry_run, + envelopes_out=args.librarian_envelopes_out, + ) + else: + run_librarian( + cache_file=args.cache_file, + dry_run=args.librarian_dry_run or not args.run_librarian, + source_jsonl=args.librarian_source, + ) def ai_client_init(database: db.Node_collection): @@ -1238,6 +1250,67 @@ def run_librarian( ) +def run_librarian_live( + cache_file: str, + pipeline_run_id: str, + dry_run: bool = False, + envelopes_out: Optional[str] = None, +) -> None: + """Module C entrypoint against Module B's live queue (W8). + + The counterpart to ``--run_noise_filter``: where ``run_librarian`` walks a + JSONL fixture and logs shortlists, this drains the ``knowledge_queue`` rows + Module B wrote for ``pipeline_run_id`` through C.0->C.4 and stamps + ``consumed_at`` on the ones it finished. Prints the ``RunSummary`` as JSON + on stdout so the OIE orchestrator can read it, exactly as Module B does. + + Still writes no links: the graph/review writers are W8b. What a real run + does write is the envelopes (as JSONL, to ``envelopes_out``) and one column + on B's queue. Those two go together — retiring a row whose envelope was + discarded would lose the chunk — so a non-dry run requires an output path. + ``dry_run`` writes neither. + + Ops note: unchanged from ``run_librarian`` — opt-in CLI only, not on the + Procfile, not wired into the web app or the worker. It calls the paid + embedding API, so a deployment never triggers it on its own. + """ + from datetime import datetime, timezone + + from application.utils.librarian.config_loader import load_config + from application.utils.librarian.envelope_sink import ( + JsonlEnvelopeSink, + NullEnvelopeSink, + ) + from application.utils.librarian.factory import build_components + from application.utils.librarian.queue_runner import run_librarian_queue + + sink = JsonlEnvelopeSink(envelopes_out) if envelopes_out else NullEnvelopeSink() + if not dry_run and envelopes_out is None: + raise SystemExit( + "--run_librarian --run_id needs --librarian_envelopes_out : a " + "real run marks queue rows consumed, so the envelopes it built have " + "to land somewhere first. Add --librarian_dry_run to run without " + "writing anything." + ) + + cfg = load_config() + database = db_connect(path=cache_file) + components = build_components(database, config=cfg) + + # The CLI boundary is the one place a clock read belongs; everything below + # takes `at` as an argument so a run stays reproducible. + summary = run_librarian_queue( + database.session, + pipeline_run_id, + components, + cfg, + at=datetime.now(timezone.utc), + sink=sink, + dry_run=dry_run, + ) + print(summary.to_json()) + + def regenerate_embeddings(db_url: str) -> None: """Wipe all embedding rows, then rebuild (CRE + every node type) like ``--generate_embeddings``.""" from application.prompt_client import prompt_client as prompt_client diff --git a/application/tests/librarian/config_loader_test.py b/application/tests/librarian/config_loader_test.py index 27000a76c..63818888f 100644 --- a/application/tests/librarian/config_loader_test.py +++ b/application/tests/librarian/config_loader_test.py @@ -16,6 +16,9 @@ def test_defaults_when_env_unset(self): self.assertEqual(cfg.top_k_retrieval, 20) self.assertEqual(cfg.top_k_rerank, 5) self.assertEqual(cfg.link_threshold, 0.8) + # 1.0 is the identity transform: an honestly *uncalibrated* softmax, + # rather than a temperature nobody fitted. + self.assertEqual(cfg.temperature, 1.0) self.assertEqual(cfg.batch_size, 32) self.assertEqual(cfg.ece_target, 0.10) self.assertEqual(cfg.conformal_alpha, 0.10) @@ -34,6 +37,7 @@ class TestConfigLoaderOverrides(unittest.TestCase): "CRE_LIBRARIAN_TOP_K_RETRIEVAL": "50", "CRE_LIBRARIAN_TOP_K_RERANK": "10", "CRE_LIBRARIAN_LINK_THRESHOLD": "0.7", + "CRE_LIBRARIAN_TEMPERATURE": "1.208", "CRE_LIBRARIAN_BATCH_SIZE": "64", "CRE_LIBRARIAN_ECE_TARGET": "0.05", "CRE_LIBRARIAN_CONFORMAL_ALPHA": "0.20", @@ -47,6 +51,7 @@ def test_env_overrides_apply(self): self.assertEqual(cfg.top_k_retrieval, 50) self.assertEqual(cfg.top_k_rerank, 10) self.assertAlmostEqual(cfg.link_threshold, 0.7) + self.assertAlmostEqual(cfg.temperature, 1.208) self.assertEqual(cfg.batch_size, 64) self.assertAlmostEqual(cfg.ece_target, 0.05) self.assertAlmostEqual(cfg.conformal_alpha, 0.20) @@ -65,6 +70,17 @@ def test_link_threshold_above_one_raises(self): with self.assertRaises(ValueError): load_config() + def test_non_positive_temperature_raises(self): + """T divides the logits, so zero or negative is undefined, not merely a + bad setting — the same guard TemperatureScaler applies.""" + for value in ("0", "-1.5", "nan"): + with self.subTest(value=value): + with mock.patch.dict( + os.environ, {"CRE_LIBRARIAN_TEMPERATURE": value}, clear=True + ): + with self.assertRaises(ValueError): + load_config() + def test_negative_top_k_retrieval_raises(self): with mock.patch.dict( os.environ, {"CRE_LIBRARIAN_TOP_K_RETRIEVAL": "-1"}, clear=True diff --git a/application/tests/librarian/envelope_sink_test.py b/application/tests/librarian/envelope_sink_test.py new file mode 100644 index 000000000..8e8ab282d --- /dev/null +++ b/application/tests/librarian/envelope_sink_test.py @@ -0,0 +1,189 @@ +"""Tests for the envelope sinks. + +``persists`` is the load-bearing property here: the queue runner reads it to +decide whether retiring a source row would lose work, so a sink that lies about +it would silently destroy chunks. +""" + +import json +import os +import shutil +import tempfile +import unittest +from datetime import datetime, timezone + +from application.utils.librarian.envelope_sink import ( + JsonlEnvelopeSink, + NullEnvelopeSink, +) +from application.utils.librarian.schemas import ( + SCHEMA_VERSION, + CreCandidate, + KnowledgeSnapshot, + LinkProposal, + Locator, + ProposedLink, + RetrievalAudit, + SourceRef, + UpdateDetection, +) + +AT = datetime(2026, 8, 5, 12, 0, 0, tzinfo=timezone.utc) + + +def _proposal(chunk_id: str = "chk:1") -> LinkProposal: + return LinkProposal( + schema_version=SCHEMA_VERSION, + chunk_id=chunk_id, + artifact_id="art:1", + pipeline_run_id="run-1", + classified_at=AT, + knowledge=KnowledgeSnapshot( + text="Verify passwords are at least 12 characters.", + source=SourceRef( + type="github", + repo="OWASP/ASVS", + commit_sha="abc1234567890", + committed_at=AT, + ), + locator=Locator(kind="repo_path", id="a.md", path="a.md"), + ), + retrieval=RetrievalAudit( + retriever="stub/1.0.0", + candidates=[CreCandidate(cre_id="616-305", score_vector=0.9)], + reranked=[CreCandidate(cre_id="616-305", score_rerank=4.0)], + threshold=0.8, + ), + links=[ + ProposedLink( + cre_id="616-305", link_type="Automatically linked to", confidence=0.95 + ) + ], + update_detection=UpdateDetection(is_update=False), + ) + + +class NullEnvelopeSinkTest(unittest.TestCase): + def test_declares_that_it_does_not_persist(self) -> None: + self.assertFalse(NullEnvelopeSink().persists) + + def test_counts_without_keeping(self) -> None: + sink = NullEnvelopeSink() + self.assertEqual(sink.write([_proposal(), _proposal("chk:2")]), 2) + self.assertEqual(sink.written, 2) + + +class JsonlEnvelopeSinkTest(unittest.TestCase): + def setUp(self) -> None: + self.dir = tempfile.mkdtemp() + # Every method writes envelopes containing chunk text, so the directory + # is removed rather than left in the system temp dir once per test. + self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True) + self.path = os.path.join(self.dir, "envelopes.jsonl") + + def test_declares_that_it_persists(self) -> None: + self.assertTrue(JsonlEnvelopeSink(self.path).persists) + + def test_writes_one_rfc_envelope_per_line(self) -> None: + JsonlEnvelopeSink(self.path).write([_proposal("chk:1"), _proposal("chk:2")]) + + with open(self.path, encoding="utf-8") as fh: + lines = [json.loads(line) for line in fh if line.strip()] + + self.assertEqual([r["chunk_id"] for r in lines], ["chk:1", "chk:2"]) + # The file holds the RFC shape Module D will read, not a Python repr. + self.assertEqual(lines[0]["status"], "linked") + self.assertEqual(lines[0]["schema_version"], SCHEMA_VERSION) + + def test_appends_across_runs(self) -> None: + """Several pipeline runs share one output file; each envelope carries + its own run id, so truncating would throw away earlier runs.""" + sink = JsonlEnvelopeSink(self.path) + sink.write([_proposal("chk:1")]) + sink.write([_proposal("chk:2")]) + + with open(self.path, encoding="utf-8") as fh: + self.assertEqual(len([ln for ln in fh if ln.strip()]), 2) + + def test_empty_batch_writes_nothing(self) -> None: + self.assertEqual(JsonlEnvelopeSink(self.path).write([]), 0) + self.assertFalse(os.path.exists(self.path)) + + def test_creates_the_parent_directory(self) -> None: + nested = os.path.join(self.dir, "a", "b", "envelopes.jsonl") + JsonlEnvelopeSink(nested).write([_proposal()]) + self.assertTrue(os.path.exists(nested)) + + def test_written_envelope_validates_against_the_vendored_rfc_schema(self) -> None: + """What lands on disk is what Module D validates, so validate it here. + + The RFC types its optional fields as plain ``"string"`` and leaves them + out of ``required``, so an absent value has to be an absent *key*. + Pydantic's default dump writes ``"repo": null`` instead, which the + schema rejects — every envelope failed, and the file still looked + perfectly well-formed. Asserting "one JSON object per line" cannot catch + that; only validating against the schema can. + """ + from jsonschema import Draft202012Validator + from referencing import Registry, Resource + + schema_dir = os.path.abspath( + os.path.join( + os.path.dirname(__file__), + "..", + "..", + "utils", + "librarian", + "_rfc_schemas", + ) + ) + schemas = [ + json.load(open(os.path.join(schema_dir, name), encoding="utf-8")) + for name in sorted(os.listdir(schema_dir)) + if name.endswith(".json") + ] + # The schemas cross-reference each other by ``$id``, and link-proposal + # also uses local ``#/$defs/...`` pointers, so resolution has to happen + # in ``$id`` space rather than against a directory path. + registry = Registry().with_resources( + [(s["$id"], Resource.from_contents(s)) for s in schemas if "$id" in s] + ) + schema = next(s for s in schemas if s["$id"].endswith("link-proposal.json")) + + JsonlEnvelopeSink(self.path).write([_proposal()]) + with open(self.path, encoding="utf-8") as fh: + envelope = json.loads(fh.readline()) + + errors = list( + Draft202012Validator(schema, registry=registry).iter_errors(envelope) + ) + + self.assertEqual( + errors, + [], + "written envelope must satisfy the RFC schema Module D reads:\n" + + "\n".join(f" {list(e.path)}: {e.message}" for e in errors[:5]), + ) + + def test_absent_optional_fields_are_omitted_not_nulled(self) -> None: + # A github-sourced envelope carries no feed_url/post_guid. The key must + # be gone, not present-and-null, or the RFC validator rejects it. + JsonlEnvelopeSink(self.path).write([_proposal()]) + with open(self.path, encoding="utf-8") as fh: + envelope = json.loads(fh.readline()) + + def _nulls(node, path=""): + if isinstance(node, dict): + for k, v in node.items(): + yield from _nulls(v, f"{path}.{k}") + elif isinstance(node, list): + for i, v in enumerate(node): + yield from _nulls(v, f"{path}[{i}]") + elif node is None: + yield path + + self.assertEqual(list(_nulls(envelope)), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/evaluate_harness_test.py b/application/tests/librarian/evaluate_harness_test.py index 9c2c59356..ed3b62bef 100644 --- a/application/tests/librarian/evaluate_harness_test.py +++ b/application/tests/librarian/evaluate_harness_test.py @@ -14,10 +14,12 @@ without touching the retriever or reranker again. """ +import contextlib import importlib.util +import io import os import unittest -from typing import List, Optional +from typing import Iterator, List, Optional from application.utils.librarian.schemas import CreCandidate, RetrievalAudit @@ -73,6 +75,14 @@ def _golden_row( ) +@contextlib.contextmanager +def _captured_stdout() -> Iterator[io.StringIO]: + """The reports return only a status; their numbers are printed.""" + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + yield buf + + class CountingPipeline: """Stub retriever+reranker that records how many passes it was asked for.""" @@ -233,26 +243,97 @@ def test_grades_expected_decision_rows_off_shared_audits(self) -> None: audits = harness.live_audits(rows, pipe, pipe) before = (pipe.retrieve_calls, pipe.rerank_calls) - status = harness.report_decision_accuracy( - rows, audits, TemperatureScaler(1.0), 0.80 - ) + with _captured_stdout() as out: + status = harness.report_decision_accuracy( + rows, audits, TemperatureScaler(1.0), 0.80 + ) self.assertEqual(status, 0, "the C.4 report is informational, never a gate") self.assertEqual((pipe.retrieve_calls, pipe.rerank_calls), before) + # Both rows are graded and both land where the golden set expects: the + # dominant top-1 auto-links, the near-tie falls under tau and routes to + # review with BELOW_THRESHOLD. Assert the numbers, not just the status, + # or this passes on a report that counted nothing. + report = out.getvalue() + self.assertIn("decision (C.4, 2 rows @ tau=0.80): overall 2/2", report) + self.assertIn("auto-link recall (expected-linked rows): 1/1", report) + self.assertIn("review recall (expected-review rows): 1/1", report) + self.assertIn("reason_code match 1/1", report) def test_no_graded_rows_is_not_an_error(self) -> None: from application.utils.librarian.calibration.temperature import ( TemperatureScaler, ) + # No audits at all, so nothing is gradeable even though the row carries an + # expected decision — this is the branch a --slice selection hits. rows = [_golden_row("p1", "positive", "alpha", ["616-305"])] - pipe = CountingPipeline({"alpha": [("616-305", 4.0)]}) - audits = harness.live_audits(rows, pipe, pipe) - status = harness.report_decision_accuracy( - rows, audits, TemperatureScaler(1.0), 0.80 - ) + with _captured_stdout() as out: + status = harness.report_decision_accuracy( + rows, {}, TemperatureScaler(1.0), 0.80 + ) + self.assertEqual(status, 0) + self.assertIn("no rows with an expected decision", out.getvalue()) + + +class BoundaryGateTest(unittest.TestCase): + """The C.0 boundary must not be able to fail silently. + + Golden rows are adapted into a synthetic ``knowledge_queue`` row before C.0 + validates them. When Module B's table shape moved in #989, that adapter kept + minting the old flat row: every row was rejected, so the explicit gate had + nothing to count, skipped itself, and the run still exited 0. Both halves are + pinned here — the adapter produces a row the real validator accepts, and a + drifted adapter fails the run instead of reporting success. + """ + + _DATASET = os.path.join( + os.path.dirname(__file__), "fixtures", "golden_dataset.json" + ) + + def test_synthetic_row_satisfies_the_live_queue_schema(self) -> None: + from application.utils.librarian.section_validator import ( + section_from_queue_row, + ) + + rows = harness.load_dataset(self._DATASET) + + # Every row, not a sample: main() validates the whole selection, so a + # row past any cap could fail C.0 while this test still passed. + for row in rows: + # Raises SectionValidationError if the shape drifted from B's table. + section = section_from_queue_row(harness.queue_row_from_golden(row)) + self.assertTrue(section.text) + + def test_minted_ids_are_deterministic(self) -> None: + # The live reports key their shared audits by these ids; if a second run + # minted different ones, the audits would stop lining up with the rows. + row = harness.load_dataset(self._DATASET)[0] + + self.assertEqual( + harness.queue_row_from_golden(row), harness.queue_row_from_golden(row) + ) + + def test_total_boundary_rejection_fails_the_run(self) -> None: + original = harness.queue_row_from_golden + # Regress the adapter to the pre-W8 flat shape B no longer writes. + harness.queue_row_from_golden = lambda row: { + "id": row.id, + "text": row.input.text, + "confidence": 0.99, + "llm_label": "KNOWLEDGE", + "created_at": harness._SYNTHETIC_CREATED_AT, + } + try: + with _captured_stdout() as out: + status = harness.main(["--dataset", self._DATASET]) + finally: + harness.queue_row_from_golden = original + + self.assertEqual(status, 1, "a boundary that rejects everything must fail") + self.assertIn("FAILED (gates did not run)", out.getvalue()) if __name__ == "__main__": diff --git a/application/tests/librarian/factory_test.py b/application/tests/librarian/factory_test.py new file mode 100644 index 000000000..976df5e7d --- /dev/null +++ b/application/tests/librarian/factory_test.py @@ -0,0 +1,113 @@ +"""Tests for the component factory (the orchestrator's entry into Module C). + +``build_components`` reaches for a live database and the cross-encoder, so the +test here fakes the database and patches the one heavy loader. That is still +worth doing rather than skipping: everything else in the function — the +``cre_defs`` import, the hub read, the pool build, the wiring of all three +stages — is code that only ever runs in production otherwise, and a typo in any +of it is a crash in the orchestrator's entry point. +""" + +import os +import unittest +from unittest import mock + +from application.utils.librarian.config_loader import load_config +from application.utils.librarian.factory import build_components, build_scaler + + +class BuildScalerTest(unittest.TestCase): + def test_uses_the_configured_temperature(self) -> None: + with mock.patch.dict( + os.environ, {"CRE_LIBRARIAN_TEMPERATURE": "1.208"}, clear=True + ): + scaler = build_scaler(load_config()) + self.assertAlmostEqual(scaler.temperature, 1.208) + + def test_default_temperature_warns_that_it_is_uncalibrated(self) -> None: + """T=1.0 is a plain softmax. Running the C.4 threshold against an + unfitted confidence is exactly the mistake W5 existed to prevent, so it + has to be loud rather than silent.""" + with mock.patch.dict(os.environ, {}, clear=True): + with self.assertLogs( + "application.utils.librarian.factory", level="WARNING" + ) as logs: + scaler = build_scaler(load_config()) + self.assertAlmostEqual(scaler.temperature, 1.0) + self.assertIn("uncalibrated", "\n".join(logs.output).lower()) + + def test_fitted_temperature_is_quiet(self) -> None: + with mock.patch.dict( + os.environ, {"CRE_LIBRARIAN_TEMPERATURE": "1.033"}, clear=True + ): + with mock.patch( + "application.utils.librarian.factory.logger" + ) as fake_logger: + build_scaler(load_config()) + fake_logger.warning.assert_not_called() + + +class _FakeDatabase: + """The two hub reads ``build_components`` makes, and nothing else.""" + + def __init__(self) -> None: + self.embeddings = {"616-305": [0.1, 0.2, 0.3], "111-111": [0.3, 0.2, 0.1]} + self.texts = {"616-305": "password storage", "111-111": "session handling"} + + def get_embeddings_by_doc_type(self, doc_type): + return self.embeddings + + def get_embedding_contents_by_doc_type(self, doc_type): + return self.texts + + +class BuildComponentsTest(unittest.TestCase): + def _build(self): + # Only the cross-encoder load is patched — it pulls in torch. The rest + # of the factory runs for real. + with mock.patch( + "application.utils.librarian.cross_encoder." "build_cross_encoder_score_fn", + return_value=lambda pairs: [0.0 for _ in pairs], + ): + with mock.patch.dict( + os.environ, {"CRE_LIBRARIAN_TEMPERATURE": "1.2"}, clear=True + ): + return build_components( + _FakeDatabase(), + config=load_config(), + embed_fn=lambda text: [0.1, 0.2, 0.3], + ) + + def test_builds_all_three_stages(self) -> None: + components = self._build() + self.assertTrue(hasattr(components.retriever, "retrieve")) + self.assertTrue(hasattr(components.reranker, "rerank")) + # The configured temperature reaches C.3 rather than the 1.0 default. + self.assertAlmostEqual(components.scaler.temperature, 1.2) + self.assertTrue(0.0 < components.scaler.confidence([2.0, 0.0]) < 1.0) + + def test_exposes_the_hub_ids_as_the_link_registry(self) -> None: + """These are the only ids C may link to, and what the explicit-reference + fast path validates a cited id against.""" + self.assertEqual(self._build().known_cre_ids, frozenset({"616-305", "111-111"})) + + def test_embed_fn_is_injectable_so_no_paid_call_is_made(self) -> None: + calls = [] + + def embed(text): + calls.append(text) + return [0.1, 0.2, 0.3] + + with mock.patch( + "application.utils.librarian.cross_encoder." "build_cross_encoder_score_fn", + return_value=lambda pairs: [0.0 for _ in pairs], + ): + components = build_components( + _FakeDatabase(), config=load_config(), embed_fn=embed + ) + components.retriever.retrieve("verify passwords") + self.assertEqual(calls, ["verify passwords"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/fixtures/sample_knowledge_queue.jsonl b/application/tests/librarian/fixtures/sample_knowledge_queue.jsonl index de82c456a..cb54b82fc 100644 --- a/application/tests/librarian/fixtures/sample_knowledge_queue.jsonl +++ b/application/tests/librarian/fixtures/sample_knowledge_queue.jsonl @@ -1,3 +1,5 @@ -{"id":"4a8c1b2e-1d2f-4e3a-9b4c-5d6e7f8a9b0c","source_repo":"OWASP/ASVS","source_path":"4.0/en/0x11-V2-Authentication.md","source_commit_sha":"abc123def456789012345678901234567890abcd","text":"Verify that user-set passwords are at least 12 characters in length.","confidence":0.93,"llm_label":"KNOWLEDGE","llm_reasoning":"clear security requirement on password length","created_at":"2026-05-25T02:25:00Z","consumed_at":null} -{"id":"5b9d2c3f-2e3a-4f4b-ac5d-6e7f8a9b0c1d","source_repo":"OWASP/ASVS","source_path":"4.0/en/0x11-V2-Authentication.md","source_commit_sha":"abc123def456789012345678901234567890abcd","text":"Do NOT use MD5 for password hashing; it is cryptographically broken for this purpose.","confidence":0.88,"llm_label":"KNOWLEDGE","llm_reasoning":"explicit security guidance against a deprecated primitive","created_at":"2026-05-25T02:25:01Z","consumed_at":null} -{"id":"6cae3d40-3f4b-4a5c-bd6e-7f8a9b0c1d2e","source_repo":"OWASP/wstg","source_path":"document/4-Web_Application_Security_Testing/05-Authentication_Testing.md","source_commit_sha":"def78901234567890123456789012345678901ab","text":"Testing for weak lockout mechanisms: confirm the application locks accounts after a defined number of failed login attempts.","confidence":0.81,"llm_label":"KNOWLEDGE","llm_reasoning":"testing methodology for anti-automation","created_at":"2026-05-25T02:25:02Z","consumed_at":null} +{"id":"4a8c1b2e-1d2f-4e3a-9b4c-5d6e7f8a9b0c","content_hash":"9f2a1c7d3e5b8a04c6d1e2f3a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7","chunk_id":"chk:art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md:0","artifact_id":"art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md","pipeline_run_id":"run-2026-08-05-001","schema_version":"0.2.0","source_type":"github","source_repo":"OWASP/ASVS","source_commit_sha":"abc123def456789012345678901234567890abcd","source_committed_at":"2026-05-24T18:02:11Z","feed_url":null,"post_guid":null,"locator_kind":"repo_path","locator_path":"4.0/en/0x11-V2-Authentication.md","span_index":0,"span_total":3,"span_heading_path":"[\"V2 Authentication\",\"V2.1 Password Security\"]","text":"Verify that user-set passwords are at least 12 characters in length.","llm_label":"KNOWLEDGE","confidence":0.93,"llm_reasoning":"clear security requirement on password length","created_at":"2026-05-25T02:25:00Z","consumed_at":null} +{"id":"5b9d2c3f-2e3a-4f4b-ac5d-6e7f8a9b0c1d","content_hash":"1b2c3d4e5f60718293a4b5c6d7e8f90a9f2a1c7d3e5b8a04c6d1e2f3a4b5c6d7","chunk_id":"chk:art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md:1","artifact_id":"art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md","pipeline_run_id":"run-2026-08-05-001","schema_version":"0.2.0","source_type":"github","source_repo":"OWASP/ASVS","source_commit_sha":"abc123def456789012345678901234567890abcd","source_committed_at":"2026-05-24T18:02:11Z","feed_url":null,"post_guid":null,"locator_kind":"repo_path","locator_path":"4.0/en/0x11-V2-Authentication.md","span_index":1,"span_total":3,"span_heading_path":"[\"V2 Authentication\",\"V2.4 Credential Storage\"]","text":"Do NOT use MD5 for password hashing; it is cryptographically broken for this purpose.","llm_label":"KNOWLEDGE","confidence":0.88,"llm_reasoning":"explicit security guidance against a deprecated primitive","created_at":"2026-05-25T02:25:01Z","consumed_at":null} +{"id":"6cae3d40-3f4b-4a5c-bd6e-7f8a9b0c1d2e","content_hash":"c6d7e8f90a1b2c3d4e5f60718293a4b59f2a1c7d3e5b8a04c6d1e2f3a4b5c6d7","chunk_id":"chk:art:OWASP/wstg:document/4-Web_Application_Security_Testing/05-Authentication_Testing.md:0","artifact_id":"art:OWASP/wstg:document/4-Web_Application_Security_Testing/05-Authentication_Testing.md","pipeline_run_id":"run-2026-08-05-001","schema_version":"0.2.0","source_type":"github","source_repo":"OWASP/wstg","source_commit_sha":"def78901234567890123456789012345678901ab","source_committed_at":"2026-05-24T20:14:52Z","feed_url":null,"post_guid":null,"locator_kind":"repo_path","locator_path":"document/4-Web_Application_Security_Testing/05-Authentication_Testing.md","span_index":0,"span_total":1,"span_heading_path":"[\"4.5 Authentication Testing\",\"4.5.3 Testing for Weak Lockout Mechanism\"]","text":"Testing for weak lockout mechanisms: confirm the application locks accounts after a defined number of failed login attempts.","llm_label":"KNOWLEDGE","confidence":0.81,"llm_reasoning":"testing methodology for anti-automation","created_at":"2026-05-25T02:25:02Z","consumed_at":null} +{"id":"7dbf4e51-4a5c-4b6d-ce7f-8a9b0c1d2e3f","content_hash":"5f60718293a4b5c6d7e8f90a1b2c3d4e9f2a1c7d3e5b8a04c6d1e2f3a4b5c6d7","chunk_id":"chk:art:owasp-blog:https://owasp.org/blog/2026/05/20/session-fixation.html:0","artifact_id":"art:owasp-blog:https://owasp.org/blog/2026/05/20/session-fixation.html","pipeline_run_id":"run-2026-08-05-001","schema_version":"0.2.0","source_type":"rss","source_repo":null,"source_commit_sha":null,"source_committed_at":null,"feed_url":"https://owasp.org/blog/feed.xml","post_guid":"https://owasp.org/blog/2026/05/20/session-fixation","locator_kind":"feed_item","locator_path":"https://owasp.org/blog/2026/05/20/session-fixation.html","span_index":0,"span_total":1,"span_heading_path":"[\"Preventing session fixation\"]","text":"Always regenerate the session identifier after a successful authentication to prevent session fixation attacks.","llm_label":"KNOWLEDGE","confidence":0.86,"llm_reasoning":"actionable session-management guidance from an OWASP feed post","created_at":"2026-05-25T02:25:03Z","consumed_at":null} +{"id":"8ec05f62-5b6d-4c7e-df80-9b0c1d2e3f40","content_hash":"718293a4b5c6d7e8f90a1b2c3d4e5f609f2a1c7d3e5b8a04c6d1e2f3a4b5c6d7","chunk_id":"chk:art:OWASP/ASVS:4.0/en/0x12-V3-Session-management.md:0","artifact_id":"art:OWASP/ASVS:4.0/en/0x12-V3-Session-management.md","pipeline_run_id":"run-2026-08-05-001","schema_version":"0.2.0","source_type":"github","source_repo":"OWASP/ASVS","source_commit_sha":"abc123def456789012345678901234567890abcd","source_committed_at":"2026-05-24T18:02:11Z","feed_url":null,"post_guid":null,"locator_kind":"repo_path","locator_path":"4.0/en/0x12-V3-Session-management.md","span_index":0,"span_total":1,"span_heading_path":null,"text":"This section was rewritten; see the changelog for details.","llm_label":"UNCERTAIN","confidence":0.41,"llm_reasoning":"no concrete security requirement stated; routed to human review","created_at":"2026-05-25T02:25:04Z","consumed_at":null} diff --git a/application/tests/librarian/knowledge_source_test.py b/application/tests/librarian/knowledge_source_test.py new file mode 100644 index 000000000..b621d7263 --- /dev/null +++ b/application/tests/librarian/knowledge_source_test.py @@ -0,0 +1,159 @@ +"""Tests for the two knowledge sources (C's read side). + +``DbKnowledgeSource`` runs against an in-memory SQLite DB +(``create_app(mode="test")`` + ``create_all``), matching the project's +``db_test.py`` pattern — no migration needed. + +The behaviour that matters here is what the source *refuses* to read: consumed +rows, other runs' rows, and — the one with a cross-module consequence — +``UNCERTAIN`` rows, which belong to Module D. +""" + +import json +import os +import tempfile +import unittest +from datetime import datetime, timezone + +from application import create_app, sqla +from application.database.db import KnowledgeQueueItem as KnowledgeQueueRow +from application.utils.librarian.knowledge_source import ( + DbKnowledgeSource, + FixtureKnowledgeSource, +) + +_FIXTURE = os.path.join( + os.path.dirname(__file__), "fixtures", "sample_knowledge_queue.jsonl" +) + + +def _row(row_id: str, **overrides) -> KnowledgeQueueRow: + values = dict( + id=row_id, + content_hash=f"hash-{row_id}", + chunk_id=f"chk:art:OWASP/ASVS:a.md:{row_id}", + artifact_id="art:OWASP/ASVS:a.md", + pipeline_run_id="run-1", + schema_version="0.2.0", + source_type="github", + source_repo="OWASP/ASVS", + source_commit_sha="abc1234567890", + source_committed_at="2026-05-24T18:02:11Z", + locator_kind="repo_path", + locator_path="a.md", + span_index=0, + span_total=1, + text="Verify that passwords are at least 12 characters.", + llm_label="KNOWLEDGE", + confidence=0.9, + created_at=datetime(2026, 5, 25, 2, 25, 0, tzinfo=timezone.utc), + ) + values.update(overrides) + return KnowledgeQueueRow(**values) + + +class FixtureKnowledgeSourceTest(unittest.TestCase): + def test_reads_the_bundled_v0_2_fixture(self) -> None: + rows = list(FixtureKnowledgeSource(_FIXTURE).items()) + self.assertEqual(len(rows), 5) + # The fixture carries both source shapes B writes. + self.assertEqual({r.source_type.value for r in rows}, {"github", "rss"}) + + def test_malformed_line_is_skipped_not_fatal(self) -> None: + with tempfile.NamedTemporaryFile( + "w", suffix=".jsonl", delete=False, encoding="utf-8" + ) as fh: + fh.write(json.dumps({"id": "nope"}) + "\n") + with open(_FIXTURE, encoding="utf-8") as src: + fh.write(src.readline()) + tmp = fh.name + try: + with self.assertLogs( + "application.utils.librarian.knowledge_source", level="WARNING" + ): + rows = list(FixtureKnowledgeSource(tmp).items()) + self.assertEqual(len(rows), 1) + finally: + os.unlink(tmp) + + +class DbKnowledgeSourceTest(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.ctx = self.app.app_context() + self.ctx.push() + sqla.create_all() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.ctx.pop() + + def test_reads_unconsumed_knowledge_rows(self) -> None: + sqla.session.add_all([_row("a"), _row("b")]) + sqla.session.commit() + + items = list(DbKnowledgeSource(sqla.session).items()) + + self.assertEqual([i.id for i in items], ["a", "b"]) + # The SQLAlchemy row validates straight into C's mirror. + self.assertEqual(items[0].chunk_id, "chk:art:OWASP/ASVS:a.md:a") + + def test_consumed_rows_are_not_re_read(self) -> None: + sqla.session.add_all( + [ + _row("a"), + _row("b", consumed_at=datetime(2026, 6, 1, tzinfo=timezone.utc)), + ] + ) + sqla.session.commit() + + items = list(DbKnowledgeSource(sqla.session).items()) + + self.assertEqual([i.id for i in items], ["a"]) + + def test_uncertain_rows_are_left_for_module_d(self) -> None: + """B writes UNCERTAIN for Module D's human review. If C read those rows + it would also mark them consumed, silently emptying D's queue.""" + sqla.session.add_all([_row("a"), _row("b", llm_label="UNCERTAIN")]) + sqla.session.commit() + + items = list(DbKnowledgeSource(sqla.session).items()) + + self.assertEqual([i.id for i in items], ["a"]) + + def test_scopes_to_one_pipeline_run(self) -> None: + sqla.session.add_all([_row("a"), _row("b", pipeline_run_id="run-2")]) + sqla.session.commit() + + items = list(DbKnowledgeSource(sqla.session, pipeline_run_id="run-2").items()) + + self.assertEqual([i.id for i in items], ["b"]) + + def test_limit_is_applied_in_a_stable_order(self) -> None: + """created_at alone is not unique — B inserts a batch in one + transaction — so the id break is what makes a limited run repeatable.""" + sqla.session.add_all([_row("c"), _row("a"), _row("b")]) + sqla.session.commit() + + first = [i.id for i in DbKnowledgeSource(sqla.session, limit=2).items()] + again = [i.id for i in DbKnowledgeSource(sqla.session, limit=2).items()] + + self.assertEqual(first, ["a", "b"]) + self.assertEqual(first, again) + + def test_unmodellable_row_is_skipped_not_fatal(self) -> None: + """A row B wrote that C cannot model must not abort the batch.""" + sqla.session.add_all([_row("a"), _row("b", source_type="carrier-pigeon")]) + sqla.session.commit() + + with self.assertLogs( + "application.utils.librarian.knowledge_source", level="WARNING" + ): + items = list(DbKnowledgeSource(sqla.session).items()) + + self.assertEqual([i.id for i in items], ["a"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/pipeline_test.py b/application/tests/librarian/pipeline_test.py index 9bddd7ee1..38f491d77 100644 --- a/application/tests/librarian/pipeline_test.py +++ b/application/tests/librarian/pipeline_test.py @@ -20,12 +20,21 @@ RUN = "run-7" -def _row(text="Verify the JWT signature.", label="KNOWLEDGE"): +def _row(text="Verify the JWT signature.", label="KNOWLEDGE", row_id="1"): return KnowledgeQueueItem( - id="1", + id=row_id, + content_hash=f"hash-{row_id}", + chunk_id=f"chk:art:owasp/x:a.md:{row_id}", + artifact_id="art:owasp/x:a.md", + pipeline_run_id=RUN, + schema_version="0.2.0", + source_type="github", source_repo="owasp/x", - source_path="a.md", source_commit_sha="abcdef1", + locator_kind="repo_path", + locator_path="a.md", + span_index=0, + span_total=1, text=text, confidence=0.9, llm_label=label, @@ -189,6 +198,53 @@ def test_errored_is_counted_separately_from_skipped(self): self.assertEqual(result.stats.errored, 1) +class RowOutcomeTest(unittest.TestCase): + """Per-row outcomes (W8): what the queue write-back keys its decision on. + + ``RunStats`` counts alone cannot say *which* rows finished, and retiring an + errored row would silently drop that chunk from the pipeline forever. + """ + + def _run(self, rows, scaler_confidence=0.95): + return LibrarianPipeline( + _Source(rows), + _Retriever(), + _Reranker(TOP), + _Scaler(scaler_confidence), + threshold=0.8, + pipeline_run_id=RUN, + ).run(at=AT) + + def test_outcome_per_row_carries_the_queue_id(self): + result = self._run([_row(row_id="a"), _row(row_id="b")]) + self.assertEqual([o.row_id for o in result.outcomes], ["a", "b"]) + self.assertEqual({o.status.value for o in result.outcomes}, {"linked"}) + + def test_boundary_rejection_is_finished_with(self): + """A malformed row cannot be fixed by re-reading it, so it counts as + finished — otherwise it is re-read on every run, forever.""" + result = self._run([_row(row_id="bad", label="UNCERTAIN")]) + self.assertEqual([o.status.value for o in result.outcomes], ["skipped"]) + self.assertEqual(result.finished_row_ids(), ["bad"]) + + def test_errored_row_is_not_finished(self): + result = LibrarianPipeline( + _Source([_row(row_id="a")]), + failing_component_stub("retriever"), + _Reranker(TOP), + _Scaler(0.95), + threshold=0.8, + pipeline_run_id=RUN, + ).run(at=AT) + self.assertEqual([o.status.value for o in result.outcomes], ["errored"]) + self.assertEqual(result.finished_row_ids(), []) + + def test_finished_ids_mix_decisions_and_rejections_but_not_errors(self): + rows = [_row(row_id="ok"), _row(row_id="skip", label="UNCERTAIN")] + result = self._run(rows) + self.assertEqual(sorted(result.finished_row_ids()), ["ok", "skip"]) + + def failing_component_stub(kind): """A stub whose single method always raises, for the given seam.""" diff --git a/application/tests/librarian/queue_consumer_test.py b/application/tests/librarian/queue_consumer_test.py new file mode 100644 index 000000000..0c07cddcb --- /dev/null +++ b/application/tests/librarian/queue_consumer_test.py @@ -0,0 +1,141 @@ +"""Tests for C's write-back to Module B's queue (``mark_consumed``). + +In-memory SQLite via ``create_app(mode="test")``, matching ``db_test.py``. + +This is the only column Module C writes on B's table, so the tests are about +restraint: stamp exactly the rows asked for, never re-stamp one that is already +consumed, and never delete anything. +""" + +import unittest +from datetime import datetime, timezone + +from application import create_app, sqla +from application.database.db import KnowledgeQueueItem as KnowledgeQueueRow +from application.utils.librarian.queue_consumer import mark_consumed + +AT = datetime(2026, 8, 5, 12, 0, 0, tzinfo=timezone.utc) +EARLIER = datetime(2026, 8, 1, 9, 0, 0, tzinfo=timezone.utc) + +# B declared `consumed_at` as a plain `DateTime`, so the driver stores the UTC +# wall clock and hands back a naive value. Compare against that rather than +# against the aware input we wrote. +AT_NAIVE = AT.replace(tzinfo=None) +EARLIER_NAIVE = EARLIER.replace(tzinfo=None) + + +def _row(row_id: str, **overrides) -> KnowledgeQueueRow: + values = dict( + id=row_id, + content_hash=f"hash-{row_id}", + chunk_id=f"chk:{row_id}", + artifact_id="art:x", + pipeline_run_id="run-1", + schema_version="0.2.0", + source_type="github", + source_repo="OWASP/ASVS", + source_commit_sha="abc1234567890", + locator_kind="repo_path", + locator_path="a.md", + span_index=0, + span_total=1, + text="some security text", + llm_label="KNOWLEDGE", + confidence=0.9, + created_at=datetime(2026, 5, 25, tzinfo=timezone.utc), + ) + values.update(overrides) + return KnowledgeQueueRow(**values) + + +class MarkConsumedTest(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.ctx = self.app.app_context() + self.ctx.push() + sqla.create_all() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.ctx.pop() + + def _consumed_at(self, row_id: str): + return sqla.session.get(KnowledgeQueueRow, row_id).consumed_at + + def test_stamps_only_the_given_rows(self) -> None: + sqla.session.add_all([_row("a"), _row("b")]) + sqla.session.commit() + + stamped = mark_consumed(sqla.session, ["a"], at=AT) + + self.assertEqual(stamped, 1) + self.assertEqual(self._consumed_at("a"), AT_NAIVE) + self.assertIsNone(self._consumed_at("b")) + + def test_replay_does_not_move_an_existing_timestamp(self) -> None: + """Idempotence is the point: a re-run must not rewrite when a row was + first consumed, or the audit trail stops meaning anything.""" + sqla.session.add(_row("a", consumed_at=EARLIER)) + sqla.session.commit() + + stamped = mark_consumed(sqla.session, ["a"], at=AT) + + self.assertEqual(stamped, 0) + self.assertEqual(self._consumed_at("a"), EARLIER_NAIVE) + + def test_partial_stamp_is_reported_not_raised(self) -> None: + sqla.session.add_all([_row("a"), _row("b", consumed_at=EARLIER)]) + sqla.session.commit() + + with self.assertLogs( + "application.utils.librarian.queue_consumer", level="INFO" + ): + stamped = mark_consumed(sqla.session, ["a", "b"], at=AT) + + self.assertEqual(stamped, 1) + + def test_unknown_id_is_not_an_error(self) -> None: + sqla.session.add(_row("a")) + sqla.session.commit() + + with self.assertLogs( + "application.utils.librarian.queue_consumer", level="INFO" + ): + stamped = mark_consumed(sqla.session, ["a", "ghost"], at=AT) + + self.assertEqual(stamped, 1) + + def test_duplicate_ids_are_counted_once(self) -> None: + sqla.session.add(_row("a")) + sqla.session.commit() + + self.assertEqual(mark_consumed(sqla.session, ["a", "a"], at=AT), 1) + + def test_empty_input_touches_nothing(self) -> None: + sqla.session.add(_row("a")) + sqla.session.commit() + + self.assertEqual(mark_consumed(sqla.session, [], at=AT), 0) + self.assertIsNone(self._consumed_at("a")) + + def test_rows_are_never_deleted(self) -> None: + """The queue is also the audit trail of the B->C handover.""" + sqla.session.add_all([_row("a"), _row("b")]) + sqla.session.commit() + + mark_consumed(sqla.session, ["a", "b"], at=AT) + + self.assertEqual(sqla.session.query(KnowledgeQueueRow).count(), 2) + + def test_stamps_more_rows_than_one_chunk(self) -> None: + """The id list is stamped in chunks; the boundary must not drop rows.""" + ids = [f"r{i:04d}" for i in range(1200)] + sqla.session.add_all([_row(i) for i in ids]) + sqla.session.commit() + + self.assertEqual(mark_consumed(sqla.session, ids, at=AT), 1200) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/queue_runner_test.py b/application/tests/librarian/queue_runner_test.py new file mode 100644 index 000000000..2614162eb --- /dev/null +++ b/application/tests/librarian/queue_runner_test.py @@ -0,0 +1,373 @@ +"""End-to-end tests for the live B->C path (``run_librarian_queue``). + +Real Module B rows in an in-memory SQLite DB, real C.0->C.4 code, stub C.1/C.2/C.3 +— so the wiring under test is exactly the wiring that runs in production, minus +the embedding API and the cross-encoder. + +The claims worth proving here are the ones that only appear once both ends are +connected: a queue row becomes an envelope, a finished row gets retired, an +errored row does not, and a second run is a no-op rather than a re-run. +""" + +import unittest +from datetime import datetime, timezone + +from application import create_app, sqla +from application.database.db import KnowledgeQueueItem as KnowledgeQueueRow +from application.utils.librarian.config_loader import LibrarianConfig +from application.utils.librarian.envelope_sink import NullEnvelopeSink +from application.utils.librarian.factory import LibrarianComponents +from application.utils.librarian.queue_runner import RunSummary, run_librarian_queue +from application.utils.librarian.schemas import CreCandidate, RetrievalAudit + +AT = datetime(2026, 8, 5, 12, 0, 0, tzinfo=timezone.utc) +RUN = "run-1" + + +def _config(threshold: float = 0.80) -> LibrarianConfig: + return LibrarianConfig( + crossencoder_model="stub", + retriever_backend="in_memory", + top_k_retrieval=20, + top_k_rerank=5, + link_threshold=threshold, + temperature=1.0, + batch_size=32, + ece_target=0.10, + conformal_alpha=0.10, + ) + + +def _row(row_id: str, **overrides) -> KnowledgeQueueRow: + values = dict( + id=row_id, + content_hash=f"hash-{row_id}", + chunk_id=f"chk:art:OWASP/ASVS:a.md:{row_id}", + artifact_id="art:OWASP/ASVS:a.md", + pipeline_run_id=RUN, + schema_version="0.2.0", + source_type="github", + source_repo="OWASP/ASVS", + source_commit_sha="abc1234567890", + source_committed_at="2026-05-24T18:02:11Z", + locator_kind="repo_path", + locator_path="a.md", + span_index=0, + span_total=1, + text="Verify that passwords are at least 12 characters.", + llm_label="KNOWLEDGE", + confidence=0.9, + created_at=datetime(2026, 5, 25, 2, 25, 0, tzinfo=timezone.utc), + ) + values.update(overrides) + return KnowledgeQueueRow(**values) + + +class _Retriever: + def retrieve(self, text: str) -> RetrievalAudit: + return RetrievalAudit( + retriever="stub/1.0.0", + candidates=[CreCandidate(cre_id="616-305", score_vector=0.9)], + reranked=[], + threshold=0.0, + ) + + +class _Reranker: + def __init__(self, logit: float = 20.0) -> None: + self._logit = logit + + def rerank(self, text: str, audit: RetrievalAudit) -> RetrievalAudit: + return audit.model_copy( + update={ + "reranked": [ + CreCandidate(cre_id="616-305", score_rerank=self._logit), + CreCandidate(cre_id="999-999", score_rerank=0.0), + ] + } + ) + + +class _Scaler: + """Returns a fixed confidence, so the auto-link branch is chosen by the test.""" + + def __init__(self, confidence: float) -> None: + self._confidence = confidence + + def confidence(self, logits) -> float: + return self._confidence + + +class _ExplodingRetriever: + def retrieve(self, text: str) -> RetrievalAudit: + raise RuntimeError("embedding API timed out") + + +class _RecordingSink: + """A persisting sink that keeps the batch in memory for assertions.""" + + def __init__(self) -> None: + self.envelopes: list = [] + + @property + def persists(self) -> bool: + return True + + def write(self, envelopes) -> int: + self.envelopes.extend(envelopes) + return len(envelopes) + + +class _ExplodingSink: + @property + def persists(self) -> bool: + return True + + def write(self, envelopes) -> int: + raise IOError("disk full") + + +def _components(confidence: float = 0.95, retriever=None) -> LibrarianComponents: + return LibrarianComponents( + retriever=retriever or _Retriever(), + reranker=_Reranker(), + scaler=_Scaler(confidence), + known_cre_ids=frozenset({"616-305"}), + ) + + +class RunLibrarianQueueTest(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.ctx = self.app.app_context() + self.ctx.push() + sqla.create_all() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.ctx.pop() + + def _consumed_at(self, row_id: str): + return sqla.session.get(KnowledgeQueueRow, row_id).consumed_at + + def _run(self, **kwargs): + kwargs.setdefault("sink", _RecordingSink()) + return run_librarian_queue( + sqla.session, + RUN, + kwargs.pop("components", None) or _components(), + _config(), + at=AT, + **kwargs, + ) + + def test_queue_row_becomes_a_link_and_is_consumed(self) -> None: + sqla.session.add(_row("a")) + sqla.session.commit() + + summary = self._run() + + self.assertEqual((summary.read, summary.linked, summary.review), (1, 1, 0)) + self.assertEqual(summary.consumed, 1) + self.assertIsNotNone(self._consumed_at("a")) + + def test_low_confidence_routes_to_review_and_still_consumes(self) -> None: + """A review is a completed decision, not a failure — the row is done.""" + sqla.session.add(_row("a")) + sqla.session.commit() + + summary = self._run(components=_components(confidence=0.10)) + + self.assertEqual((summary.linked, summary.review), (0, 1)) + self.assertEqual(summary.consumed, 1) + + def test_envelope_carries_the_rows_own_identity(self) -> None: + """The point of the W8 schema fix, asserted end to end.""" + sqla.session.add(_row("a")) + sqla.session.commit() + + run_librarian_queue( + sqla.session, RUN, _components(), _config(), at=AT, dry_run=True + ) + # dry_run leaves state alone; re-read through the pipeline to inspect. + from application.utils.librarian.knowledge_source import DbKnowledgeSource + from application.utils.librarian.section_validator import ( + section_from_queue_row, + ) + + item = next(iter(DbKnowledgeSource(sqla.session).items())) + section = section_from_queue_row(item) + self.assertEqual(section.chunk_id, "chk:art:OWASP/ASVS:a.md:a") + self.assertEqual(section.artifact_id, "art:OWASP/ASVS:a.md") + + def test_errored_row_is_left_unconsumed_for_retry(self) -> None: + """A timeout is transient; retiring the row would lose the chunk.""" + sqla.session.add(_row("a")) + sqla.session.commit() + + summary = self._run(components=_components(retriever=_ExplodingRetriever())) + + self.assertEqual(summary.errored, 1) + self.assertEqual(summary.consumed, 0) + self.assertIsNone(self._consumed_at("a")) + + def test_second_run_is_a_no_op(self) -> None: + """Consumption is what stops the queue being reprocessed forever.""" + sqla.session.add_all([_row("a"), _row("b")]) + sqla.session.commit() + + first = self._run() + second = self._run() + + self.assertEqual((first.read, first.consumed), (2, 2)) + self.assertEqual((second.read, second.consumed), (0, 0)) + + def test_dry_run_reads_and_decides_but_stamps_nothing(self) -> None: + sqla.session.add(_row("a")) + sqla.session.commit() + + summary = self._run(dry_run=True) + + self.assertEqual((summary.read, summary.linked), (1, 1)) + self.assertEqual(summary.consumed, 0) + self.assertIsNone(self._consumed_at("a")) + + def test_only_the_named_run_is_drained(self) -> None: + sqla.session.add_all([_row("a"), _row("b", pipeline_run_id="other-run")]) + sqla.session.commit() + + summary = self._run() + + self.assertEqual(summary.read, 1) + self.assertIsNone(self._consumed_at("b")) + + def test_uncertain_rows_are_never_touched(self) -> None: + """They are Module D's queue; C must not drain them.""" + sqla.session.add_all([_row("a"), _row("d", llm_label="UNCERTAIN")]) + sqla.session.commit() + + summary = self._run() + + self.assertEqual(summary.read, 1) + self.assertIsNone(self._consumed_at("d")) + + def test_boundary_rejection_is_consumed_not_retried(self) -> None: + """A row C can never link is finished with; re-reading it forever is + worse than retiring it, and the count keeps it visible.""" + sqla.session.add(_row("a", text=" ")) + sqla.session.commit() + + summary = self._run() + + self.assertEqual((summary.skipped, summary.linked), (1, 0)) + self.assertEqual(summary.consumed, 1) + + def test_real_run_without_a_sink_is_refused(self) -> None: + """The rule that keeps a drain lossless: no consumption without + somewhere for the envelopes to land.""" + sqla.session.add(_row("a")) + sqla.session.commit() + + with self.assertRaises(ValueError) as ctx: + run_librarian_queue(sqla.session, RUN, _components(), _config(), at=AT) + + self.assertIn("EnvelopeSink", str(ctx.exception)) + self.assertIsNone(self._consumed_at("a")) + + def test_real_run_behind_a_non_persisting_sink_is_refused(self) -> None: + sqla.session.add(_row("a")) + sqla.session.commit() + + with self.assertRaises(ValueError): + run_librarian_queue( + sqla.session, + RUN, + _components(), + _config(), + at=AT, + sink=NullEnvelopeSink(), + ) + + self.assertIsNone(self._consumed_at("a")) + + def test_a_failing_sink_consumes_nothing(self) -> None: + """Persist first, retire second: if the write fails the rows stay B's + to hand back, and the whole run is retried.""" + sqla.session.add(_row("a")) + sqla.session.commit() + + with self.assertRaises(IOError): + self._run(sink=_ExplodingSink()) + + self.assertIsNone(self._consumed_at("a")) + + def test_envelopes_reach_the_sink_before_rows_are_retired(self) -> None: + sqla.session.add_all([_row("a"), _row("b")]) + sqla.session.commit() + sink = _RecordingSink() + + summary = self._run(sink=sink) + + self.assertEqual(summary.persisted, 2) + self.assertEqual( + sorted(e.chunk_id for e in sink.envelopes), + ["chk:art:OWASP/ASVS:a.md:a", "chk:art:OWASP/ASVS:a.md:b"], + ) + + def test_unevaluated_safety_path_is_reported_not_hidden(self) -> None: + """No SafetyGuard exists yet, so every row is decided without it. That + has to show up in the summary rather than look like a clean result.""" + sqla.session.add_all([_row("a"), _row("b")]) + sqla.session.commit() + + with self.assertLogs( + "application.utils.librarian.queue_runner", level="WARNING" + ) as logs: + summary = self._run() + + self.assertEqual(summary.safety_unevaluated, 2) + self.assertIn("safety path", "\n".join(logs.output)) + + def test_summary_serializes_for_the_orchestrator(self) -> None: + import json + + sqla.session.add(_row("a")) + sqla.session.commit() + + payload = json.loads(self._run().to_json()) + + self.assertEqual(payload["run_id"], RUN) + self.assertEqual(payload["linked"], 1) + + def test_status_declares_a_degraded_run(self) -> None: + """The orchestrator branches on ``status``; a constant "ok" would hide + exactly the runs worth noticing. + + Every row today is decided behind ``NullSafetyGuard``, so a real run is + genuinely degraded until a detector exists — and the field says which + rows and why, rather than leaving the reader to know that a non-zero + ``safety_unevaluated`` is bad news. + """ + sqla.session.add(_row("a")) + sqla.session.commit() + + summary = self._run() + + self.assertTrue(summary.safety_unevaluated) + self.assertIn("degraded", summary.status) + self.assertIn("safety path", summary.status) + + def test_status_is_ok_when_there_is_nothing_to_declare(self) -> None: + summary = RunSummary(run_id=RUN, read=3, linked=3) + summary.finalize_status() + self.assertEqual(summary.status, "ok") + + def test_status_names_errored_rows(self) -> None: + summary = RunSummary(run_id=RUN, read=3, linked=2, errored=1) + summary.finalize_status() + self.assertIn("1 errored", summary.status) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/safety_guard_test.py b/application/tests/librarian/safety_guard_test.py new file mode 100644 index 000000000..e6a9d3ac9 --- /dev/null +++ b/application/tests/librarian/safety_guard_test.py @@ -0,0 +1,148 @@ +"""Tests for the C.4 safety seam. + +The behaviour under test is a distinction, not a detector: an unevaluated +verdict must never be readable as a clean one. That is the same failure mode +W5's review caught in the calibration gate, which skipped and still reported +success. +""" + +import unittest +from datetime import datetime, timezone + +from application.utils.librarian.pipeline import LibrarianPipeline +from application.utils.librarian.safety_guard import NullSafetyGuard, SafetyVerdict +from application.utils.librarian.schemas import ( + CreCandidate, + KnowledgeQueueItem, + ReasonCode, + RetrievalAudit, + ReviewItem, +) + +AT = datetime(2026, 1, 1, tzinfo=timezone.utc) +RUN = "run-safety" + + +def _row(row_id: str = "1") -> KnowledgeQueueItem: + return KnowledgeQueueItem( + id=row_id, + content_hash=f"hash-{row_id}", + chunk_id=f"chk:{row_id}", + artifact_id="art:1", + pipeline_run_id=RUN, + schema_version="0.2.0", + source_type="github", + source_repo="owasp/x", + source_commit_sha="abcdef1", + locator_kind="repo_path", + locator_path="a.md", + span_index=0, + span_total=1, + text="Verify the JWT signature.", + llm_label="KNOWLEDGE", + confidence=0.9, + created_at="2026-01-01T00:00:00Z", + ) + + +class _Source: + def __init__(self, rows): + self._rows = rows + + def items(self): + return iter(self._rows) + + +class _Retriever: + def retrieve(self, text): + return RetrievalAudit( + retriever="stub", + candidates=[CreCandidate(cre_id="616-305", score_vector=0.9)], + reranked=[], + threshold=0.0, + ) + + +class _Reranker: + def rerank(self, text, audit): + return audit.model_copy( + update={"reranked": [CreCandidate(cre_id="616-305", score_rerank=9.0)]} + ) + + +class _Scaler: + def confidence(self, logits): + return 0.99 # comfortably over the bar, so only a flag can block it + + +class _FlaggingGuard: + def __init__(self, **flags) -> None: + self._verdict = SafetyVerdict(evaluated=True, **flags) + + def evaluate(self, section) -> SafetyVerdict: + return self._verdict + + +def _run(guard=None): + return LibrarianPipeline( + _Source([_row()]), + _Retriever(), + _Reranker(), + _Scaler(), + threshold=0.8, + pipeline_run_id=RUN, + safety_guard=guard, + ).run(at=AT) + + +class SafetyVerdictTest(unittest.TestCase): + def test_default_verdict_is_unevaluated(self) -> None: + verdict = SafetyVerdict() + self.assertFalse(verdict.evaluated) + self.assertFalse(verdict.blocks_auto_link) + + def test_either_flag_blocks(self) -> None: + self.assertTrue(SafetyVerdict(adversarial=True).blocks_auto_link) + self.assertTrue(SafetyVerdict(update_ambiguous=True).blocks_auto_link) + + +class NullSafetyGuardTest(unittest.TestCase): + def test_reports_that_it_evaluated_nothing(self) -> None: + verdict = NullSafetyGuard().evaluate(section=None) + self.assertFalse(verdict.evaluated) + self.assertFalse(verdict.adversarial) + self.assertFalse(verdict.update_ambiguous) + + +class PipelineSafetyWiringTest(unittest.TestCase): + """#991: `decide()` was called without the flags, so these reason codes + could never fire from the pipeline. They can now.""" + + def test_adversarial_flag_forces_review(self) -> None: + result = _run(_FlaggingGuard(adversarial=True)) + self.assertEqual(result.stats.linked, 0) + envelope = result.envelopes[0] + self.assertIsInstance(envelope, ReviewItem) + self.assertEqual(envelope.reason_code, ReasonCode.adversarial_flag) + + def test_update_ambiguous_forces_review(self) -> None: + result = _run(_FlaggingGuard(update_ambiguous=True)) + envelope = result.envelopes[0] + self.assertIsInstance(envelope, ReviewItem) + self.assertEqual(envelope.reason_code, ReasonCode.update_ambiguous) + + def test_evaluated_and_clean_still_auto_links(self) -> None: + result = _run(_FlaggingGuard()) + self.assertEqual(result.stats.linked, 1) + self.assertEqual(result.stats.safety_unevaluated, 0) + + def test_default_guard_links_but_records_the_gap(self) -> None: + """Without a guard the row still links — but the run says the safety + path did not run for it, rather than reporting a clean check.""" + result = _run() + self.assertEqual(result.stats.linked, 1) + self.assertEqual(result.stats.safety_unevaluated, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/schemas_test.py b/application/tests/librarian/schemas_test.py index 6a5a63b1a..09f1d89a9 100644 --- a/application/tests/librarian/schemas_test.py +++ b/application/tests/librarian/schemas_test.py @@ -314,31 +314,73 @@ def test_module_c_librarian_md_example_round_trips(self): class TestKnowledgeQueueItem(unittest.TestCase): """Internal model — mirrors B's SQL row. Not an RFC contract.""" - def test_minimal_row(self): - item = KnowledgeQueueItem( + @staticmethod + def _row(**overrides): + row = dict( id="uuid-1", + content_hash="deadbeef", + chunk_id="chk:art:OWASP/ASVS:4.0/en/0x11.md:0", + artifact_id="art:OWASP/ASVS:4.0/en/0x11.md", + pipeline_run_id="run-1", + schema_version="0.2.0", + source_type="github", source_repo="OWASP/ASVS", - source_path="4.0/en/0x11.md", source_commit_sha="abc1234567890", + locator_kind="repo_path", + locator_path="4.0/en/0x11.md", + span_index=0, + span_total=1, text="Verify X.", confidence=0.9, llm_label="KNOWLEDGE", created_at="2026-05-25T02:25:00Z", ) + row.update(overrides) + return row + + def test_minimal_row(self): + item = KnowledgeQueueItem(**self._row()) self.assertIsNone(item.consumed_at) def test_confidence_bounds(self): with self.assertRaises(ValidationError): - KnowledgeQueueItem( - id="x", - source_repo="r", - source_path="p", - source_commit_sha="c", - text="t", - confidence=1.5, - llm_label="KNOWLEDGE", - created_at="2026-05-25T02:25:00Z", - ) + KnowledgeQueueItem(**self._row(confidence=1.5)) + + def test_extra_columns_are_ignored_so_b_can_extend_the_table(self): + item = KnowledgeQueueItem(**self._row(some_column_c_has_never_heard_of=1)) + self.assertEqual(item.id, "uuid-1") + + def test_heading_path_decodes(self): + item = KnowledgeQueueItem( + **self._row(span_heading_path='["V2 Authentication","V2.1 Passwords"]') + ) + self.assertEqual(item.heading_path(), ["V2 Authentication", "V2.1 Passwords"]) + + def test_heading_path_degrades_on_junk(self): + for junk in ("", None, "not json", '{"a": 1}', "[]"): + with self.subTest(junk=junk): + item = KnowledgeQueueItem(**self._row(span_heading_path=junk)) + self.assertEqual(item.heading_path(), []) + + def test_source_type_must_agree_with_the_populated_columns(self): + """B nulls repo/sha on rss rows and feed_url on github rows; a row that + contradicts its own source_type is rejected here rather than deeper in.""" + for name, overrides in ( + ("github without repo", {"source_repo": None}), + ("github without sha", {"source_commit_sha": None}), + ( + "rss without feed_url", + { + "source_type": "rss", + "source_repo": None, + "source_commit_sha": None, + "feed_url": None, + }, + ), + ): + with self.subTest(name): + with self.assertRaises(ValidationError): + KnowledgeQueueItem(**self._row(**overrides)) class TestGoldenDataset(unittest.TestCase): diff --git a/application/tests/librarian/section_validator_test.py b/application/tests/librarian/section_validator_test.py index f9dbff76e..434164a9b 100644 --- a/application/tests/librarian/section_validator_test.py +++ b/application/tests/librarian/section_validator_test.py @@ -23,11 +23,25 @@ def valid_queue_row(**overrides) -> dict: + """A `knowledge_queue` row in Module B's merged v0.2 shape (#989).""" row = { "id": "4a8c1b2e-1d2f-4e3a-9b4c-5d6e7f8a9b0c", + "content_hash": "9f2a1c7d3e5b8a04c6d1e2f3a4b5c6d7", + "chunk_id": "chk:art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md:0", + "artifact_id": "art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md", + "pipeline_run_id": "run-001", + "schema_version": "0.2.0", + "source_type": "github", "source_repo": "OWASP/ASVS", - "source_path": "4.0/en/0x11-V2-Authentication.md", "source_commit_sha": "abc123def456789012345678901234567890abcd", + "source_committed_at": "2026-05-24T18:02:11Z", + "feed_url": None, + "post_guid": None, + "locator_kind": "repo_path", + "locator_path": "4.0/en/0x11-V2-Authentication.md", + "span_index": 0, + "span_total": 3, + "span_heading_path": '["V2 Authentication","V2.1 Password Security"]', "text": "Verify that user-set passwords are at least 12 characters long.", "confidence": 0.93, "llm_label": "KNOWLEDGE", @@ -39,6 +53,27 @@ def valid_queue_row(**overrides) -> dict: return row +def valid_rss_row(**overrides) -> dict: + """The other shape B writes: a feed post, with no repo or commit at all.""" + row = valid_queue_row( + id="7dbf4e51-4a5c-4b6d-ce7f-8a9b0c1d2e3f", + chunk_id="chk:art:owasp-blog:session-fixation:0", + artifact_id="art:owasp-blog:session-fixation", + source_type="rss", + source_repo=None, + source_commit_sha=None, + source_committed_at=None, + feed_url="https://owasp.org/blog/feed.xml", + post_guid="https://owasp.org/blog/2026/05/20/session-fixation", + locator_kind="feed_item", + locator_path="https://owasp.org/blog/2026/05/20/session-fixation.html", + span_heading_path='["Preventing session fixation"]', + text="Regenerate the session id after authentication.", + ) + row.update(overrides) + return row + + def valid_knowledge_item(**overrides) -> dict: item = { "schema_version": "0.2.0", @@ -75,25 +110,62 @@ def valid_knowledge_item(**overrides) -> dict: class QueueRowBoundaryTest(unittest.TestCase): - def test_valid_row_builds_section_with_synthesized_identity(self) -> None: - section = section_from_queue_row(valid_queue_row()) + def test_identity_is_read_from_the_row_not_synthesized(self) -> None: + """The W8 contract fix: A's ids pass through C untouched. + + Through W7 these were built out of repo/path/sha, which produced ids + matching nothing upstream. Pinning both to the row's own values is what + lets a link join back to the artifact Module A harvested. + """ + row = valid_queue_row() + section = section_from_queue_row(row) self.assertIsInstance(section, Section) - self.assertEqual( - section.chunk_id, - "chk:OWASP/ASVS@abc123def456789012345678901234567890abcd:" - "4.0/en/0x11-V2-Authentication.md", - ) - self.assertEqual( - section.artifact_id, "art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md" - ) + self.assertEqual(section.chunk_id, row["chunk_id"]) + self.assertEqual(section.artifact_id, row["artifact_id"]) + + def test_github_row_maps_source_and_locator(self) -> None: + section = section_from_queue_row(valid_queue_row()) + self.assertEqual(section.source.type.value, "github") self.assertEqual(section.source.repo, "OWASP/ASVS") + # A's real commit time, not B's classification time. self.assertEqual( section.source.committed_at, - datetime(2026, 5, 25, 2, 25, tzinfo=timezone.utc), + datetime(2026, 5, 24, 18, 2, 11, tzinfo=timezone.utc), ) self.assertEqual(section.locator.path, "4.0/en/0x11-V2-Authentication.md") self.assertEqual(section.language, "en") + def test_committed_at_falls_back_to_created_at(self) -> None: + """`source_committed_at` is github-only and nullable; created_at is the + best provenance a row without one carries.""" + section = section_from_queue_row(valid_queue_row(source_committed_at=None)) + self.assertEqual( + section.source.committed_at, + datetime(2026, 5, 25, 2, 25, tzinfo=timezone.utc), + ) + + def test_rss_row_is_accepted_with_no_repo_or_sha(self) -> None: + """The whole RSS path was unrepresentable before W8: C required a repo + and a commit sha that B leaves NULL on every feed row.""" + section = section_from_queue_row(valid_rss_row()) + self.assertEqual(section.source.type.value, "rss") + self.assertIsNone(section.source.repo) + self.assertIsNone(section.source.commit_sha) + self.assertEqual(section.locator.kind.value, "feed_item") + # The guid is the stable identity for a feed item; the path is the URL. + self.assertEqual( + section.locator.id, "https://owasp.org/blog/2026/05/20/session-fixation" + ) + + def test_title_hint_comes_from_the_heading_path(self) -> None: + section = section_from_queue_row(valid_queue_row()) + self.assertEqual(section.title_hint, "V2.1 Password Security") + + def test_unparseable_heading_path_degrades_to_no_title(self) -> None: + """A cosmetic field must not cost an otherwise linkable row.""" + section = section_from_queue_row(valid_queue_row(span_heading_path="{oops")) + self.assertIsNone(section.title_hint) + def test_volatile_metadata_not_carried_into_section(self) -> None: section = section_from_queue_row( valid_queue_row(llm_reasoning="audit-only rationale") @@ -113,7 +185,7 @@ def test_rejection_table(self) -> None: ), ( "missing field", - {k: v for k, v in valid_queue_row().items() if k != "source_repo"}, + {k: v for k, v in valid_queue_row().items() if k != "chunk_id"}, MalformedKnowledgeItemError, ), ( @@ -122,6 +194,38 @@ def test_rejection_table(self) -> None: MalformedKnowledgeItemError, ), ("not a mapping", "just a string", MalformedKnowledgeItemError), + # source_type and the populated source columns must agree: B nulls + # repo/sha for rss, so a github row without them is a contract + # breach, and it has to surface as one typed boundary rejection + # rather than a raw error out of the RFC SourceRef. + ( + "github row with no repo", + valid_queue_row(source_repo=None), + MalformedKnowledgeItemError, + ), + ( + "github row with no sha", + valid_queue_row(source_commit_sha=None), + MalformedKnowledgeItemError, + ), + ( + "rss row with no feed url", + valid_rss_row(feed_url=None), + MalformedKnowledgeItemError, + ), + # Module A's contract allows a 4-character sha; the RFC SourceRef + # requires 7. That row is malformed for C, and must not escape as a + # raw Pydantic error from outside the validation call. + ( + "sha shorter than the RFC allows", + valid_queue_row(source_commit_sha="abcd"), + MalformedKnowledgeItemError, + ), + ( + "feed item whose locator is not a url", + valid_rss_row(locator_path="not-a-url"), + MalformedKnowledgeItemError, + ), ] for name, row, expected_error in cases: with self.subTest(name): diff --git a/application/utils/librarian/README.md b/application/utils/librarian/README.md new file mode 100644 index 000000000..23a5e6f7b --- /dev/null +++ b/application/utils/librarian/README.md @@ -0,0 +1,106 @@ +# Module C — The Librarian + +Module C is the decision stage of the OWASP Integrated Ecosystem (OIE) pipeline. +Module A harvests changes from OWASP repositories, Module B filters the noise and +writes what survives to `knowledge_queue`, and **Module C reads that queue and +decides what each chunk means**: link it to a CRE automatically, or route it to a +human. + +```text +A (harvester) ──▶ harvest_input ──▶ B (noise filter) ──▶ knowledge_queue ──▶ C (librarian) ──▶ LinkProposal + └▶ ReviewItem ──▶ D (HITL) +``` + +C never guesses quietly. Every chunk leaves as one of two RFC envelopes, each +carrying the full retrieval audit that produced it, so a decision can always be +explained after the fact. + +## The stages + +| Stage | Module | What it does | +|---|---|---| +| **C.-1** | `schemas.py`, `config_loader.py` | RFC contracts, config, the read-only mirror of B's `knowledge_queue` row | +| **C.0** | `section_validator.py` | Input boundary — validates and adapts a queue row into an internal `Section` without re-normalizing text | +| **C.0.5** | `explicit_link_resolver.py` | Deterministic path: a chunk that cites a CRE id resolves with no ML at all | +| **C.1** | `candidate_retriever.py` | Embedding retrieval over the CRE hub — produces a shortlist | +| **C.2** | `cross_encoder.py` | Cross-encoder reranker — re-sorts that shortlist | +| **C.3** | `calibration/temperature.py` | Temperature scaling — turns a rerank logit into an honest probability, gated at ECE < 0.10 | +| **C.4** | `decision_engine.py`, `emitter.py` | `decide()` thresholds the confidence; the emitter builds the `LinkProposal` or `ReviewItem` | + +Supporting the live path: + +| Module | Role | +|---|---| +| `pipeline.py` | Runs C.0 → C.4 over a batch. Persistence-free and hermetic | +| `knowledge_source.py` | Where rows come from — `DbKnowledgeSource` (live) or `FixtureKnowledgeSource` (JSONL) | +| `envelope_sink.py` | Where envelopes go — `JsonlEnvelopeSink` (durable) or `NullEnvelopeSink` (dry runs) | +| `queue_consumer.py` | Stamps `consumed_at` back on B's queue. Idempotent; never deletes | +| `queue_runner.py` | The live entry point: drain → decide → persist → retire | +| `factory.py` | Builds the live C.1/C.2/C.3 components from config + the OpenCRE database | +| `safety_guard.py` | The blocking-flag seam `decide()` accepts. Ships as `NullSafetyGuard` | +| `hub_firewall.py` | TRACT hub firewall — strips candidates that leak the answer during evaluation | + +## The two rules that matter + +**1. A row is only retired if its envelope survived.** Marking a queue row +consumed tells Module B never to offer that chunk again. Doing that while the +envelope goes nowhere destroys the chunk outright. So `queue_runner` refuses to +stamp anything unless it was given a sink that reports `persists=True` *and* that +sink accepted the batch. Dry runs use `NullEnvelopeSink`, which reports `False`, +so a dry run can never consume. + +**2. Errors and refusals are different.** A row rejected at the C.0 boundary is +consumed — re-reading a malformed row forever helps nobody. A row that *errored* +mid-pipeline (an embedding timeout, a cross-encoder hiccup) is left unconsumed, +because the next run should retry it. `RunStats` counts them separately on +purpose. + +## Design constraints + +**Hermetic by default.** Nothing in this package imports the database at module +scope. `factory.py` is the single boundary where that stops being true, and even +there the imports are function-local. That is why the whole package is testable +without a DB, an API key, or a model download. + +**Seams, not implementations.** C.1 takes an `embed_fn`, C.2 a `score_fn`, C.3 a +scaler, C.4 a safety guard. Each is a `Protocol`, so the live components and the +test stubs are interchangeable and the decision logic stays model-free. + +**Declared-degraded over silently-degraded.** `NullSafetyGuard` evaluates nothing +and *says so* — its verdict carries `evaluated=False`, the pipeline counts those +rows, and the runner reports the count. An unevaluated safety path must never +look identical to a clean one. + +## Running it + +See [the runbook](../../../docs/gsoc_2026_module_c/runbook.md) for setup, live +runs, and troubleshooting. The short version: + +```bash +# hermetic regression harness — no DB, no key, no model +python scripts/evaluate_librarian.py \ + --dataset application/tests/librarian/fixtures/golden_dataset.json + +# dry run over a JSONL fixture +python cre.py --run_librarian --librarian_dry_run \ + --librarian_source application/tests/librarian/fixtures/sample_knowledge_queue.jsonl + +# the full test suite +python -m pytest application/tests/librarian/ +``` + +## Contracts + +- **B → C:** [`docs/gsoc_2026_module_b/module_c_contract.md`](../../../docs/gsoc_2026_module_b/module_c_contract.md) + — the `knowledge_queue` table, column by column. +- **C → D:** the `LinkProposal` / `ReviewItem` envelopes in `schemas.py`, pinned + to the vendored RFC schemas under `_rfc_schemas/`. + +## Not built yet + +- **The graph / review-queue writers (W8b).** C emits envelopes to JSONL; nothing + commits a link into the graph. The rule those writers must honour is already + stated in `safety_guard.py`: a writer that commits links **must refuse to run + behind a guard that reports `evaluated=False`.** +- **The SafetyGuard detector.** The seam is wired; the out-of-distribution + scoring, conformal prediction, and update detection behind it are future work. diff --git a/application/utils/librarian/__init__.py b/application/utils/librarian/__init__.py index a787f9ac7..755fb05a8 100644 --- a/application/utils/librarian/__init__.py +++ b/application/utils/librarian/__init__.py @@ -4,8 +4,10 @@ auto-links them or routes them to human review. Contracts (v0.2.0, RFC #734): - B -> C : KnowledgeItem (RFC envelope — what B emits) - internal: KnowledgeQueueItem (mirror of B's SQL row, master guide §1.2) + B -> C : knowledge_queue row (the live handover; see + docs/gsoc_2026_module_b/module_c_contract.md) + B -> C : KnowledgeItem (RFC envelope — the fixture/offline shape) + internal: KnowledgeQueueItem (read-side mirror of B's SQL row) C -> graph : LinkProposal (confident auto-link, status=linked) C -> D : ReviewItem (low-confidence / flagged, routed to HITL) @@ -18,12 +20,19 @@ W4 (C.2): cross-encoder reranker — re-sorts the C.1 shortlist, fills reranked[]. W5 (C.3): confidence calibration — temperature scaling maps a rerank logit to an honest probability (fit by NLL on the golden set, gated ECE < 0.10). - W6 (C.4): decision engine — thresholds the calibrated confidence to auto-link - (LinkProposal) or route to human review (ReviewItem), with a reason. - W6b (C.4): envelope emitter — builds the RFC LinkProposal / ReviewItem from a - DecisionResult — plus the C.0->C.4 pipeline glue that runs a batch - of queue rows end to end (dry-run: nothing is persisted). -The live queue drain and the graph / review-queue writers (W8) are not built yet. + W6 (C.4): decision engine — decide() turns a calibrated confidence into + auto-link vs. review, with a reason code. + W6b (C.4): envelope emitter + the C.0->C.4 LibrarianPipeline (dry-run). + W7: threshold sweep over the golden set; tau holds at 0.80. Analysis + only, no code. + W8: live B->C integration — the queue mirror reconciled against B's + merged table, a DB-backed source, the envelope sink, the + consumed_at write-back, the component factory the orchestrator + builds C from, and the safety seam wired into decide(). + +Not built yet: the SafetyGuard detector itself (the seam ships with +NullSafetyGuard, which evaluates nothing and says so), and the graph / review +writers — W8b. C still commits no links. Vendored RFC JSON schemas live under ``_rfc_schemas/``. They are pinned to upstream/owasp-graph @ 2b1437987768d5ed20fe9ee721ab9a898c4b84af (PR #734). diff --git a/application/utils/librarian/config_loader.py b/application/utils/librarian/config_loader.py index 945a210c1..5646e94e1 100644 --- a/application/utils/librarian/config_loader.py +++ b/application/utils/librarian/config_loader.py @@ -1,9 +1,20 @@ """Loads CRE_LIBRARIAN_* environment variables into a typed config. -Loader only — nothing consumes these yet. Defaults match the OIE design doc so -later weeks (retriever W3, cross-encoder W4, SafetyGuard W5) read one source. +One source of truth for every tunable in Module C: the retriever (C.1), the +cross-encoder (C.2), the calibration temperature (C.3), and the auto-link +threshold (C.4). Defaults match the OIE design doc. + +Two of these are fitted numbers, not preferences, and both come off a run of +``scripts/evaluate_librarian.py --use_live_embeddings``: + +- ``CRE_LIBRARIAN_TEMPERATURE`` is C.3's ``T``. The harness fits it and prints + it; there is no other place it is stored. The default of 1.0 is the identity + transform — an *uncalibrated* softmax, honest about being unfitted rather + than pretending to a temperature nobody measured. +- ``CRE_LIBRARIAN_LINK_THRESHOLD`` is τ, held at 0.80 by the W7 sweep. """ +import math import os from dataclasses import dataclass @@ -20,6 +31,7 @@ class LibrarianConfig: top_k_retrieval: int top_k_rerank: int link_threshold: float + temperature: float batch_size: int ece_target: float conformal_alpha: float @@ -33,6 +45,7 @@ def load_config() -> LibrarianConfig: top_k_retrieval = int(os.getenv("CRE_LIBRARIAN_TOP_K_RETRIEVAL", "20")) top_k_rerank = int(os.getenv("CRE_LIBRARIAN_TOP_K_RERANK", "5")) link_threshold = float(os.getenv("CRE_LIBRARIAN_LINK_THRESHOLD", "0.8")) + temperature = float(os.getenv("CRE_LIBRARIAN_TEMPERATURE", "1.0")) batch_size = int(os.getenv("CRE_LIBRARIAN_BATCH_SIZE", "32")) ece_target = float(os.getenv("CRE_LIBRARIAN_ECE_TARGET", "0.10")) conformal_alpha = float(os.getenv("CRE_LIBRARIAN_CONFORMAL_ALPHA", "0.10")) @@ -59,6 +72,12 @@ def load_config() -> LibrarianConfig: raise ValueError( f"CRE_LIBRARIAN_LINK_THRESHOLD must be in [0.0, 1.0], got {link_threshold}" ) + # Mirrors TemperatureScaler's own guard: T divides the logits, so zero or + # negative is not a bad setting, it is undefined. + if not math.isfinite(temperature) or temperature <= 0: + raise ValueError( + f"CRE_LIBRARIAN_TEMPERATURE must be finite and > 0, got {temperature}" + ) if not 0.0 <= ece_target <= 1.0: raise ValueError( f"CRE_LIBRARIAN_ECE_TARGET must be in [0.0, 1.0], got {ece_target}" @@ -74,6 +93,7 @@ def load_config() -> LibrarianConfig: top_k_retrieval=top_k_retrieval, top_k_rerank=top_k_rerank, link_threshold=link_threshold, + temperature=temperature, batch_size=batch_size, ece_target=ece_target, conformal_alpha=conformal_alpha, diff --git a/application/utils/librarian/envelope_sink.py b/application/utils/librarian/envelope_sink.py new file mode 100644 index 000000000..0b6815930 --- /dev/null +++ b/application/utils/librarian/envelope_sink.py @@ -0,0 +1,114 @@ +"""Where C's envelopes go once they are built. + +Retiring a queue row is only safe if the envelope built from it survived +somewhere. A run that stamps ``consumed_at`` and drops the ``LinkProposal`` on +the floor has destroyed that chunk: B will not re-offer the row, and nothing +downstream ever saw the decision. So consumption is defined against a sink, not +against the pipeline having finished. + +Two implementations here: + +- ``JsonlEnvelopeSink`` — appends one JSON envelope per line. Durable, greppable, + and enough to make a live drain lossless before the graph writer exists. +- ``NullEnvelopeSink`` — counts and discards, for dry runs. It reports + ``persists=False``, and the runner refuses to consume behind it. + +The graph / review-queue writers (W8b) become further implementations of the +same protocol, so the consumption rule does not have to change when they land. +""" + +import logging +import os +from typing import List, Protocol, Sequence, Union + +from application.utils.librarian.schemas import LinkProposal, ReviewItem + +logger = logging.getLogger(__name__) + +Envelope = Union[LinkProposal, ReviewItem] + + +class EnvelopeSink(Protocol): + """Somewhere an envelope can be durably put.""" + + @property + def persists(self) -> bool: + """True when a written envelope outlives the process. + + The runner reads this to decide whether retiring the source row would + lose work; a sink that answers False can never trigger consumption. + """ + ... + + def write(self, envelopes: Sequence[Envelope]) -> int: + """Persist the batch; return how many were written.""" + ... + + +class NullEnvelopeSink: + """Accepts envelopes and keeps nothing. For dry runs and tests.""" + + def __init__(self) -> None: + self.written = 0 + + @property + def persists(self) -> bool: + return False + + def write(self, envelopes: Sequence[Envelope]) -> int: + self.written += len(envelopes) + return len(envelopes) + + +class JsonlEnvelopeSink: + """Appends envelopes to a JSONL file, one RFC envelope per line. + + Append rather than truncate: several runs over different + ``pipeline_run_id``s share one output file, and each envelope already + carries its own run id. ``model_dump_json`` is used so the file holds the + same RFC shape Module D will consume, datetimes and enums included. + + ``exclude_none=True`` is not cosmetic. The vendored RFC schemas type their + optional fields as plain ``"string"`` and leave them out of ``required``, + so an absent value must be an *absent key* — emitting ``"repo": null`` + fails validation against the very schema this file is supposed to satisfy. + Every rss envelope trips it (no ``repo``/``commit_sha``), and so does every + github one (no ``feed_url``/``post_guid``). Writing the null would hand + Module D a file its own validator rejects. + """ + + def __init__(self, path: str) -> None: + self._path = path + + @property + def persists(self) -> bool: + return True + + def write(self, envelopes: Sequence[Envelope]) -> int: + if not envelopes: + return 0 + parent = os.path.dirname(os.path.abspath(self._path)) + os.makedirs(parent, exist_ok=True) + lines: List[str] = [e.model_dump_json(exclude_none=True) for e in envelopes] + # One open/flush for the batch, and the newline goes after every record + # so a partially written run still parses line by line. + with open(self._path, "a", encoding="utf-8") as fh: + for line in lines: + fh.write(line + "\n") + fh.flush() + os.fsync(fh.fileno()) + logger.info("wrote %d envelopes to %s", len(lines), self._path) + return len(lines) + + +def envelope_id(envelope: Envelope) -> str: + """The chunk an envelope speaks for — handy for logs and tests.""" + return envelope.chunk_id + + +__all__ = [ + "EnvelopeSink", + "JsonlEnvelopeSink", + "NullEnvelopeSink", + "envelope_id", +] diff --git a/application/utils/librarian/factory.py b/application/utils/librarian/factory.py new file mode 100644 index 000000000..6096d6c9c --- /dev/null +++ b/application/utils/librarian/factory.py @@ -0,0 +1,151 @@ +"""Builds the live C.1/C.2/C.3 components from config + the OpenCRE database. + +Until now the only place that knew how to construct a real retriever and +reranker was ``cre_main.run_librarian``, inline. That is why the OIE +orchestrator (#996) could not drive ``LibrarianPipeline`` — it had the pipeline +class but no way to produce the components it takes. This module is that seam. + +Everything here is *construction only*: no rows are read, no run is executed. +Callers that want a run use ``queue_runner.run_librarian_queue``. + +The DB and embedding imports are deliberately function-local. The rest of the +librarian package is hermetically testable precisely because it never imports +the database at module scope, and this module is the boundary where that stops +being true — keeping the imports inside the call preserves it for everyone else. +""" + +import logging +from dataclasses import dataclass +from typing import Any, Callable, FrozenSet, Optional, Sequence + +from application.utils.librarian.config_loader import LibrarianConfig, load_config +from application.utils.librarian.pipeline import Reranker, Retriever, Scaler + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class LibrarianComponents: + """The three live stages plus the CRE id registry, built once and reused. + + ``known_cre_ids`` is the set of ids present in the embedding hub — the only + ids C is allowed to link to, and what the explicit-reference fast path (C.0.5) + validates a cited id against. + """ + + retriever: Retriever + reranker: Reranker + scaler: Scaler + known_cre_ids: FrozenSet[str] + + +def build_scaler(config: Optional[LibrarianConfig] = None) -> Scaler: + """The C.3 calibrator at the configured temperature. + + ``T`` is not learned here. It is fitted offline by + ``scripts/evaluate_librarian.py --use_live_embeddings`` and set as + ``CRE_LIBRARIAN_TEMPERATURE``; the default 1.0 means *uncalibrated*, so a + deployment that never ran the fit gets a plain softmax rather than a + confidence dressed up as calibrated. + """ + from application.utils.librarian.calibration.temperature import TemperatureScaler + + config = config or load_config() + if config.temperature == 1.0: + logger.warning( + "CRE_LIBRARIAN_TEMPERATURE is 1.0 (uncalibrated). Fit it with " + "scripts/evaluate_librarian.py --use_live_embeddings and set the " + "value it prints, or the C.4 threshold is being applied to an " + "uncalibrated confidence." + ) + return TemperatureScaler(config.temperature) + + +def build_components( + database: Any, + *, + config: Optional[LibrarianConfig] = None, + embed_fn: Optional[Callable[[str], Sequence[float]]] = None, +) -> LibrarianComponents: + """Construct C.1 + C.2 + C.3 against a connected OpenCRE database. + + Args: + database: a connected ``db.Node_collection`` (the caller owns it, the + same way Module B's ``run_noise_filter`` takes a session it did not + open). + config: Module C settings; defaults to ``load_config()``. + embed_fn: text -> embedding. Defaults to the prompt handler's embedder, + which calls the paid embedding API; injectable so a caller can + supply a local or fake embedder. + + Raises whatever the pgvector guard raises when that backend is configured + but unavailable — deliberately, rather than falling back to a different + retrieval path and producing silently different rankings. + """ + from application.defs import cre_defs as defs + from application.utils.librarian.candidate_retriever import ( + CandidatePool, + RetrieverBackend, + build_retriever, + ) + from application.utils.librarian.cross_encoder import ( + CrossEncoderReranker, + build_cross_encoder_score_fn, + ) + + config = config or load_config() + backend = RetrieverBackend(config.retriever_backend) + + if backend is RetrieverBackend.pgvector: + from application.database.pgvector_utils import fail_pgvector_unavailable + + if not database.can_use_pgvector_similarity(): + fail_pgvector_unavailable( + context="CRE_LIBRARIAN_RETRIEVER_BACKEND=pgvector" + ) + + if embed_fn is None: + from application.prompt_client import prompt_client + + embed_fn = prompt_client.PromptHandler(database=database).get_text_embeddings + + cre_embeddings = database.get_embeddings_by_doc_type(defs.Credoctypes.CRE.value) + # in_memory holds the hub matrix in RAM; pgvector ranks in the DB over the + # embedding_vec column and needs no pool. Both satisfy the same retrieve(). + pool = ( + CandidatePool.from_mapping(cre_embeddings) + if backend is RetrieverBackend.in_memory + else None + ) + retriever = build_retriever( + backend, + embed_fn=embed_fn, + top_k=config.top_k_retrieval, + threshold=config.link_threshold, + pool=pool, + connection=( + database.session.connection() + if backend is RetrieverBackend.pgvector + else None + ), + ) + + # C.2 scores each (section, candidate) pair against the CRE's + # embeddings_content — the same text the hub vectors were built from. + reranker = CrossEncoderReranker( + score_fn=build_cross_encoder_score_fn(config.crossencoder_model), + top_n=config.top_k_rerank, + cre_texts=database.get_embedding_contents_by_doc_type( + defs.Credoctypes.CRE.value + ), + ) + + return LibrarianComponents( + retriever=retriever, + reranker=reranker, + scaler=build_scaler(config), + known_cre_ids=frozenset(cre_embeddings.keys()), + ) + + +__all__ = ["LibrarianComponents", "build_components", "build_scaler"] diff --git a/application/utils/librarian/knowledge_source.py b/application/utils/librarian/knowledge_source.py index 9a87ce104..90364d6d3 100644 --- a/application/utils/librarian/knowledge_source.py +++ b/application/utils/librarian/knowledge_source.py @@ -1,14 +1,23 @@ """Where Module C reads accepted chunks from. -Defines the source interface plus a fixture-backed stub for testing. The real -DB-backed source (polling Module B's knowledge_queue table) lands W8 and yields -the same KnowledgeQueueItem rows; C synthesizes the RFC KnowledgeItem envelope -from each row at processing time (master guide §1.2). +Defines the source interface plus two implementations: + +- ``FixtureKnowledgeSource`` — a JSONL file, for tests and offline dry-runs. +- ``DbKnowledgeSource`` — the live reader over Module B's ``knowledge_queue`` + table (merged in #989), which is what the orchestrator runs against. + +Both yield the same ``KnowledgeQueueItem`` mirror, so the pipeline cannot tell +them apart; C synthesizes the RFC envelope from each row downstream. + +**Only ``KNOWLEDGE`` rows are read.** B writes two labels: ``KNOWLEDGE`` (C's +work) and ``UNCERTAIN``, which exists for Module D's human review. Filtering in +the query rather than at the C.0 boundary is deliberate — a row C never reads is +a row C never marks consumed, so D's queue stays intact. """ import logging from abc import ABC, abstractmethod -from typing import Iterator +from typing import Iterator, Optional from pydantic import ValidationError @@ -16,6 +25,9 @@ logger = logging.getLogger(__name__) +# The one label Module C acts on; see the module docstring. +KNOWLEDGE_LABEL = "KNOWLEDGE" + class KnowledgeSource(ABC): @abstractmethod @@ -46,3 +58,62 @@ def items(self) -> Iterator[KnowledgeQueueItem]: exc.errors(include_input=False), ) continue + + +class DbKnowledgeSource(KnowledgeSource): + """Reads unconsumed ``KNOWLEDGE`` rows from Module B's live queue. + + The caller owns the session (mirroring Module B's ``run_noise_filter``), so + this class never opens, commits, or closes a transaction — it only reads. + Marking a row consumed is a separate, explicit step; see ``queue_consumer``. + + ``pipeline_run_id`` scopes a run to one orchestrator pass. Left unset, C + drains every unconsumed row regardless of which run produced it, which is + what a standalone catch-up run wants. ``limit`` caps one batch. + + Rows are ordered by ``created_at`` then ``id``: the timestamp alone is not + unique (B inserts a batch inside one transaction), and an unstable order + would make a ``limit``ed run non-reproducible. + """ + + def __init__( + self, + session: object, + *, + pipeline_run_id: Optional[str] = None, + limit: Optional[int] = None, + ) -> None: + self._session = session + self._run_id = pipeline_run_id + self._limit = limit + + def _query(self) -> object: + # Imported lazily: the schemas/pipeline layers stay DB-free by design, + # and this keeps `import knowledge_source` cheap for hermetic tests. + from application.database.db import KnowledgeQueueItem as KnowledgeQueueRow + + query = self._session.query(KnowledgeQueueRow).filter( # type: ignore[attr-defined] + KnowledgeQueueRow.consumed_at.is_(None), + KnowledgeQueueRow.llm_label == KNOWLEDGE_LABEL, + ) + if self._run_id: + query = query.filter(KnowledgeQueueRow.pipeline_run_id == self._run_id) + query = query.order_by(KnowledgeQueueRow.created_at, KnowledgeQueueRow.id) + if self._limit is not None: + query = query.limit(self._limit) + return query + + def items(self) -> Iterator[KnowledgeQueueItem]: + for row in self._query(): # type: ignore[attr-defined] + try: + yield KnowledgeQueueItem.model_validate(row) + except ValidationError as exc: + # A row B wrote that C cannot model is a contract breach worth + # seeing, but it must not abort the batch. Ids are safe to log; + # the row's text is not. + logger.warning( + "Skipping unmodellable knowledge_queue row id=%s: %s", + getattr(row, "id", ""), + exc.errors(include_input=False), + ) + continue diff --git a/application/utils/librarian/pipeline.py b/application/utils/librarian/pipeline.py index 8ecbec1ec..5f7bde36b 100644 --- a/application/utils/librarian/pipeline.py +++ b/application/utils/librarian/pipeline.py @@ -6,22 +6,29 @@ C.1 retriever.retrieve text -> RetrievalAudit.candidates (top-K) C.2 reranker.rerank text -> RetrievalAudit.reranked (top-N logits) C.3 scaler.confidence logits -> one calibrated confidence - C.4 decide + emit confidence -> LinkProposal | ReviewItem + C.4 safety_guard.evaluate section -> blocking flags + C.4 decide + emit confidence + flags -> LinkProposal | ReviewItem Every stage is an injected seam (``source``/``retriever``/``reranker``/``scaler``), so the whole pipeline runs hermetically with stubs — no DB, embedding model, or -cross-encoder. It is inherently **dry-run**: it builds envelopes and never persists -(the queue write-back and graph writes are W8). ``pipeline_run_id`` and the ``at`` -timestamp are injected, never read from the clock, so a run is reproducible. +cross-encoder. ``pipeline_run_id`` and the ``at`` timestamp are injected, never +read from the clock, so a run is reproducible. + +This module stays **persistence-free**: it builds envelopes and writes nothing. +What it does report, as of W8, is a ``RowOutcome`` per row, which is what lets +``queue_runner`` mark the finished rows consumed without this layer ever holding +a session. Graph writes remain out (W8b). """ import logging -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime -from typing import Any, Dict, Iterable, List, Protocol, Sequence, Union +from enum import Enum +from typing import Any, Dict, Iterable, List, Optional, Protocol, Sequence, Union from application.utils.librarian.decision_engine import decide from application.utils.librarian.emitter import emit +from application.utils.librarian.safety_guard import NullSafetyGuard, SafetyGuard from application.utils.librarian.schemas import ( KnowledgeQueueItem, LinkProposal, @@ -38,6 +45,20 @@ Envelope = Union[LinkProposal, ReviewItem] +def _row_id(item: Union[KnowledgeQueueItem, Dict[str, Any]]) -> Optional[str]: + """The queue row's primary key, read before C.0 may reject the row. + + Taken off the raw item because a row that fails validation still has to be + marked consumed — otherwise every malformed row is re-read forever. + """ + if isinstance(item, KnowledgeQueueItem): + return item.id + if isinstance(item, dict): + value = item.get("id") + return value if isinstance(value, str) else None + return None + + # The injected seams, as Protocols rather than bare duck-typing: each stage is # structurally one method, so a stub only has to provide that method, while # ``make mypy --strict`` can still check the call sites and every implementation @@ -84,12 +105,52 @@ class RunStats: review: int skipped: int errored: int = 0 + #: Rows whose safety verdict came back unevaluated. Non-zero means the + #: ADVERSARIAL_FLAG / UPDATE_AMBIGUOUS path did not run for them, so their + #: clean verdicts are defaults, not findings. + safety_unevaluated: int = 0 + + +class RowStatus(str, Enum): + """What the pipeline did with one queue row.""" + + linked = "linked" + review = "review" + skipped = "skipped" # refused at the C.0 boundary + errored = "errored" # a later stage raised + + +@dataclass(frozen=True) +class RowOutcome: + """Per-row result, so a caller can act on individual rows after the run. + + The queue write-back needs this: a row that reached a decision (or was + definitively refused at the boundary) is finished and may be marked + consumed, while an ``errored`` row must stay unconsumed so the next run + retries it. ``RunStats`` counts alone cannot express that distinction. + + ``row_id`` is the ``knowledge_queue`` primary key and is None when the + source yielded something without one (a hand-built fixture dict). + """ + + row_id: Optional[str] + chunk_id: Optional[str] + status: RowStatus @dataclass(frozen=True) class RunResult: envelopes: List[Envelope] stats: RunStats + outcomes: List[RowOutcome] = field(default_factory=list) + + def finished_row_ids(self) -> List[str]: + """Ids of rows that are done with — everything except ``errored``.""" + return [ + o.row_id + for o in self.outcomes + if o.row_id is not None and o.status != RowStatus.errored + ] class LibrarianPipeline: @@ -113,7 +174,8 @@ def __init__( scaler: Scaler, *, threshold: float, - pipeline_run_id: str + pipeline_run_id: str, + safety_guard: Optional[SafetyGuard] = None ) -> None: self._source = source self._retriever = retriever @@ -121,16 +183,24 @@ def __init__( self._scaler = scaler self._threshold = threshold self._run_id = pipeline_run_id + # Defaults to the declared-degraded guard rather than to nothing, so + # `decide()` is always called with the safety arguments and the run can + # report how many rows went unevaluated. + self._safety_guard: SafetyGuard = safety_guard or NullSafetyGuard() def run(self, *, at: datetime) -> RunResult: envelopes: List[Envelope] = [] + outcomes: List[RowOutcome] = [] linked = review = skipped = errored = total = 0 + safety_unevaluated = 0 for item in self._source.items(): total += 1 + row_id = _row_id(item) try: section = section_from_queue_row(item) except SectionValidationError: skipped += 1 # rejected at the boundary; not a decision + outcomes.append(RowOutcome(row_id, None, RowStatus.skipped)) continue # Contain failures per row. With hermetic stubs nothing here raises, @@ -145,12 +215,22 @@ def run(self, *, at: datetime) -> RunResult: cre_ids = [c.cre_id for c in reranked] confidence = self._scaler.confidence(logits) if logits else 0.0 - result = decide(confidence, cre_ids, threshold=self._threshold) + verdict = self._safety_guard.evaluate(section) + result = decide( + confidence, + cre_ids, + threshold=self._threshold, + adversarial=verdict.adversarial, + update_ambiguous=verdict.update_ambiguous, + ) + if not verdict.evaluated: + safety_unevaluated += 1 envelope = emit( section, audit, result, pipeline_run_id=self._run_id, at=at ) except Exception: errored += 1 + outcomes.append(RowOutcome(row_id, section.chunk_id, RowStatus.errored)) logger.warning( "librarian pipeline: chunk %s (artifact %s) failed after the C.0 " "boundary; skipping this row", @@ -163,8 +243,11 @@ def run(self, *, at: datetime) -> RunResult: envelopes.append(envelope) if isinstance(envelope, LinkProposal): linked += 1 + status = RowStatus.linked else: review += 1 + status = RowStatus.review + outcomes.append(RowOutcome(row_id, section.chunk_id, status)) return RunResult( envelopes=envelopes, @@ -174,5 +257,7 @@ def run(self, *, at: datetime) -> RunResult: review=review, skipped=skipped, errored=errored, + safety_unevaluated=safety_unevaluated, ), + outcomes=outcomes, ) diff --git a/application/utils/librarian/queue_consumer.py b/application/utils/librarian/queue_consumer.py new file mode 100644 index 000000000..a58926053 --- /dev/null +++ b/application/utils/librarian/queue_consumer.py @@ -0,0 +1,80 @@ +"""Module C's write-back to Module B's queue: stamp ``consumed_at``. + +The mirror image of Module B's ``queue_writer``. B inserts rows and C retires +them — ``db.KnowledgeQueueItem``'s own docstring assigns this side to C: +*"Module C reads unconsumed rows and sets consumed_at."* + +This is the only place Module C writes to Module B's table, and it writes +exactly one column. It never deletes a row: the queue doubles as the audit trail +of what B handed over and what C did with it. + +Idempotent by construction — the update is filtered on ``consumed_at IS NULL``, +so replaying a run cannot move a timestamp that is already set. The count +returned is rows actually stamped, which is why it can be lower than the number +of ids passed in (a concurrent run got there first, or the id no longer exists). + +The caller owns the transaction, matching ``run_noise_filter`` on B's side. +""" + +import logging +from datetime import datetime, timezone +from typing import Any, Iterable, List, Sequence + +logger = logging.getLogger(__name__) + +# Postgres has a bind-parameter ceiling and a huge IN (...) plans badly, so the +# id list is stamped in chunks rather than one statement. +_CHUNK_SIZE = 500 + + +def _chunks(values: Sequence[str], size: int) -> Iterable[Sequence[str]]: + for start in range(0, len(values), size): + yield values[start : start + size] + + +def mark_consumed(session: Any, row_ids: Iterable[str], *, at: datetime) -> int: + """Stamp ``consumed_at = at`` on the given rows; return how many were stamped. + + ``at`` is injected rather than read from the clock so a run is reproducible + and so every row in one batch carries the same timestamp — that shared value + is what makes a run's consumption identifiable after the fact. + """ + unique: List[str] = list(dict.fromkeys(rid for rid in row_ids if rid)) + if not unique: + return 0 + + from application.database.db import KnowledgeQueueItem as KnowledgeQueueRow + + # `consumed_at` is a plain `DateTime`, and the B->C contract is explicit that + # the UTC wall clock is "stored and read back timezone-naive". Handing an + # aware datetime to a naive column is dialect-dependent: SQLite keeps the + # offset in the string, Postgres drops it. Converting to UTC first and then + # stripping tzinfo makes the stored instant correct under both, and matches + # the `created_at` values B writes alongside it. + stamp = at.astimezone(timezone.utc).replace(tzinfo=None) if at.tzinfo else at + + stamped = 0 + for chunk in _chunks(unique, _CHUNK_SIZE): + stamped += ( + session.query(KnowledgeQueueRow) + .filter( + KnowledgeQueueRow.id.in_(list(chunk)), + KnowledgeQueueRow.consumed_at.is_(None), + ) + .update({KnowledgeQueueRow.consumed_at: stamp}, synchronize_session=False) + ) + + if stamped != len(unique): + # Not an error: another worker may have taken the row, or B may have + # pruned it. Worth a line, because a persistent gap means C is + # re-reading rows it believes it finished. + logger.info( + "librarian queue write-back: stamped %d of %d rows consumed " + "(the rest were already consumed or no longer exist)", + stamped, + len(unique), + ) + return stamped + + +__all__ = ["mark_consumed"] diff --git a/application/utils/librarian/queue_runner.py b/application/utils/librarian/queue_runner.py new file mode 100644 index 000000000..4c6461467 --- /dev/null +++ b/application/utils/librarian/queue_runner.py @@ -0,0 +1,200 @@ +"""Module C's live entry point: knowledge_queue -> C.0..C.4 -> consumed. + +The counterpart to Module B's ``run_noise_filter``, and deliberately the same +shape — ``(session, pipeline_run_id, ..., dry_run) -> RunSummary`` with a +``to_json()`` the CLI prints for the orchestrator to read. B reads +``harvest_input`` and fills ``knowledge_queue``; C drains ``knowledge_queue`` +and stamps ``consumed_at``. + +What this module adds over ``LibrarianPipeline`` is only the two DB-facing ends. +The pipeline stays persistence-free and hermetic; this wraps it with a live +source on one side and the write-back on the other. + +**Consumption is gated on persistence.** Marking a row consumed tells Module B +never to offer that chunk again, so doing it while the envelope goes nowhere +destroys the chunk outright. The runner therefore refuses to stamp anything +unless it was given a sink that reports ``persists=True`` and that sink accepted +the batch. Graph writes are still W8b; ``JsonlEnvelopeSink`` is what makes a live +drain lossless in the meantime. + +**Which rows get retired.** Everything the pipeline finished with: rows that +linked, rows that routed to review, and rows refused at the C.0 boundary (a +definitive refusal — re-reading a malformed row forever helps nobody). Rows that +*errored* mid-pipeline are left unconsumed, because those are the transient +failures — an embedding timeout, a cross-encoder hiccup — that the next run +should retry. ``RowOutcome`` is what carries that distinction out of the pipeline. + +**The safety path is declared, not assumed.** No detector exists yet, so the +pipeline runs behind ``NullSafetyGuard`` and every row comes back unevaluated. +That count is carried into the summary rather than left to look like a clean +result, and W8b's graph writer must refuse to run while it is non-zero. +""" + +import json +import logging +from dataclasses import asdict, dataclass +from datetime import datetime +from typing import Any, Optional + +from application.utils.librarian.config_loader import LibrarianConfig, load_config +from application.utils.librarian.envelope_sink import EnvelopeSink +from application.utils.librarian.factory import LibrarianComponents +from application.utils.librarian.knowledge_source import DbKnowledgeSource +from application.utils.librarian.pipeline import LibrarianPipeline, RunResult +from application.utils.librarian.queue_consumer import mark_consumed + +logger = logging.getLogger(__name__) + + +@dataclass +class RunSummary: + """Outcome of one Module C run; the CLI emits this as JSON. + + ``read`` counts rows drawn from the queue, ``consumed`` counts rows actually + stamped. They differ when rows errored (left for retry) or when the run was + a dry run. + """ + + run_id: str + read: int = 0 + linked: int = 0 + review: int = 0 + skipped: int = 0 + errored: int = 0 + persisted: int = 0 + consumed: int = 0 + #: Rows decided without the safety path having run. Reported rather than + #: hidden: an unevaluated guard must not look like a clean one. + safety_unevaluated: int = 0 + dry_run: bool = False + #: ``ok`` only when the run has nothing to declare. A constant ``"ok"`` would + #: be the same failure this module keeps fixing elsewhere: a field that looks + #: like a verdict while measuring nothing. The orchestrator reads this JSON, + #: so a run that dropped rows to errors, or decided them without the safety + #: path, has to say so in the field a consumer actually branches on — the + #: counts alone require the reader to know which ones are bad news. + status: str = "ok" + + def finalize_status(self) -> None: + """Derive ``status`` from the counts. Called once, after the run.""" + reasons = [] + if self.errored: + reasons.append(f"{self.errored} errored") + if self.safety_unevaluated: + reasons.append(f"{self.safety_unevaluated} decided without the safety path") + self.status = "degraded: " + "; ".join(reasons) if reasons else "ok" + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + +def run_librarian_queue( + session: Any, + pipeline_run_id: str, + components: LibrarianComponents, + config: Optional[LibrarianConfig] = None, + *, + at: datetime, + sink: Optional[EnvelopeSink] = None, + limit: Optional[int] = None, + dry_run: bool = False, +) -> RunSummary: + """Drain one pipeline run's unconsumed queue rows through C.0 -> C.4. + + Args: + session: SQLAlchemy session (caller owns connect/commit/teardown, as on + Module B's side). + pipeline_run_id: the run to process. Required, and it scopes both ends: + only that run's rows are read, and it is the id stamped on every + envelope. Draining several B runs under one id would misattribute + provenance, so the API does not allow it. + components: live C.1/C.2/C.3, from ``factory.build_components``. + config: Module C settings; defaults to ``load_config()``. + at: the run timestamp, injected rather than read from the clock so a run + is reproducible and every envelope and every ``consumed_at`` in the + batch share one value. + sink: where the envelopes are persisted. Required for a real run — + consuming a row whose envelope was discarded loses the chunk — and + it must report ``persists=True``. A dry run may omit it. + limit: cap on rows read in this batch; None drains the run. + dry_run: build envelopes, persist nothing, leave ``consumed_at`` alone. + + Returns a ``RunSummary``; it does not raise on individual bad rows, which + are counted as ``skipped`` or ``errored``. It *does* raise on a caller that + asks for a real run with nowhere to put the results. + """ + config = config or load_config() + summary = RunSummary(run_id=pipeline_run_id, dry_run=dry_run) + + if not dry_run: + if sink is None: + raise ValueError( + "a real run needs an EnvelopeSink: marking rows consumed while " + "discarding their envelopes would lose those chunks. Pass a " + "sink, or set dry_run=True." + ) + if not sink.persists: + raise ValueError( + f"{type(sink).__name__} does not persist envelopes, so this run " + "must not mark rows consumed. Use dry_run=True with it." + ) + + source = DbKnowledgeSource(session, pipeline_run_id=pipeline_run_id, limit=limit) + pipeline = LibrarianPipeline( + source, + components.retriever, + components.reranker, + components.scaler, + threshold=config.link_threshold, + pipeline_run_id=pipeline_run_id, + ) + + result: RunResult = pipeline.run(at=at) + summary.read = result.stats.total + summary.linked = result.stats.linked + summary.review = result.stats.review + summary.skipped = result.stats.skipped + summary.errored = result.stats.errored + summary.safety_unevaluated = result.stats.safety_unevaluated + + if summary.safety_unevaluated: + logger.warning( + "librarian run %s: %d of %d rows were decided without the safety " + "path (no SafetyGuard implementation yet), so ADVERSARIAL_FLAG and " + "UPDATE_AMBIGUOUS could not fire. Their clean verdicts are defaults, " + "not findings.", + pipeline_run_id, + summary.safety_unevaluated, + summary.read, + ) + + if dry_run: + summary.finalize_status() + return summary + + # Persist first, retire second. If the sink raises, nothing is consumed and + # the whole run is retried — the rows are still B's to hand back. + assert sink is not None # guarded above; narrows the Optional for mypy + summary.persisted = sink.write(result.envelopes) + + finished = result.finished_row_ids() + summary.consumed = mark_consumed(session, finished, at=at) + session.commit() + + logger.info( + "librarian run %s: read %d, linked %d, review %d, skipped %d, " + "errored %d, persisted %d, consumed %d", + pipeline_run_id, + summary.read, + summary.linked, + summary.review, + summary.skipped, + summary.errored, + summary.persisted, + summary.consumed, + ) + summary.finalize_status() + return summary + + +__all__ = ["RunSummary", "run_librarian_queue"] diff --git a/application/utils/librarian/safety_guard.py b/application/utils/librarian/safety_guard.py new file mode 100644 index 000000000..d79bf3041 --- /dev/null +++ b/application/utils/librarian/safety_guard.py @@ -0,0 +1,67 @@ +"""The C.4 safety seam: the blocking flags ``decide()`` already accepts. + +``decide()`` has taken ``adversarial`` / ``update_ambiguous`` since W6, but no +caller ever passed them, so ``ADVERSARIAL_FLAG`` and ``UPDATE_AMBIGUOUS`` could +not fire from the pipeline — flagged on #991, and called out there as something +that must be wired before any write-back. This module is that wiring. + +What is **not** here is a detector. The real guard (out-of-distribution scoring, +conformal prediction, update detection) is later work. So the seam ships with +``NullSafetyGuard``, which evaluates nothing and says so: its verdict carries +``evaluated=False``, the pipeline counts those rows, and the runner reports the +count. That distinction is the whole point — W5's review turned on a gate that +skipped and still reported success, and an unevaluated safety path that looks +identical to a clean one is the same failure wearing a different hat. + +The rule this establishes for W8b: **a writer that commits links into the graph +must refuse to run behind a guard that reports ``evaluated=False``.** Retiring a +queue row without the safety path is recoverable — the envelope is still on +disk. Committing a link into a graph other tools read as truth is not. +""" + +import logging +from dataclasses import dataclass +from typing import Protocol + +from application.utils.librarian.section_validator import Section + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class SafetyVerdict: + """Blocking flags for one chunk, plus whether anything actually looked. + + ``evaluated=False`` means no detector ran, so the two flags below are + defaults rather than findings. A caller must never read ``adversarial=False`` + from an unevaluated verdict as "this chunk is safe". + """ + + adversarial: bool = False + update_ambiguous: bool = False + evaluated: bool = False + + @property + def blocks_auto_link(self) -> bool: + return self.adversarial or self.update_ambiguous + + +class SafetyGuard(Protocol): + """Scores one section for the conditions that force human review.""" + + def evaluate(self, section: Section) -> SafetyVerdict: ... + + +class NullSafetyGuard: + """The declared-degraded guard: wires the seam, detects nothing. + + Deliberately not a lambda returning ``(False, False)`` — the named type and + the ``evaluated=False`` verdict are what keep "we checked and it is clean" + distinguishable from "nobody checked" everywhere downstream. + """ + + def evaluate(self, section: Section) -> SafetyVerdict: + return SafetyVerdict(evaluated=False) + + +__all__ = ["NullSafetyGuard", "SafetyGuard", "SafetyVerdict"] diff --git a/application/utils/librarian/schemas.py b/application/utils/librarian/schemas.py index 12dc0a56d..4a3c116f1 100644 --- a/application/utils/librarian/schemas.py +++ b/application/utils/librarian/schemas.py @@ -11,6 +11,7 @@ from __future__ import annotations +import json import re from datetime import datetime from enum import Enum @@ -277,25 +278,87 @@ def _schema_version_pattern(self) -> "ReviewItem": class KnowledgeQueueItem(BaseModel): - """Read-side mirror of Module B's `knowledge_queue` Postgres row. + """Read-side mirror of Module B's `knowledge_queue` row — contract v0.2. - Per master guide §1.2: C reads these rows and synthesizes the RFC - `KnowledgeItem` envelope from them. Not a wire contract; tolerates extra - fields so B can extend the row without breaking C. + Mirrors `application.database.db.KnowledgeQueueItem` (the SQLAlchemy model B + merged in #989) column for column, including its nullability: `source_repo` + and `source_commit_sha` are NULL on every `rss` row, so neither may be + required here. `from_attributes` lets a SQLAlchemy row validate directly. + + `chunk_id` / `artifact_id` are Module A's real identity, carried through B. + C consumes them verbatim — it must never mint its own, or the graph ends up + keyed on ids that never join back to A's artifacts. + + Not a wire contract: `extra="ignore"` so B can add columns without breaking C. + See docs/gsoc_2026_module_b/module_c_contract.md. """ - model_config = ConfigDict(extra="ignore") + model_config = ConfigDict(extra="ignore", from_attributes=True) id: str - source_repo: str - source_path: str - source_commit_sha: str + content_hash: str # B's dedup key + # Provenance / traceability — Module A's v0.3 record, passed through by B. + chunk_id: str + artifact_id: str + pipeline_run_id: str + schema_version: str + # Source. Which of these are populated is a function of `source_type`; + # see the `_source_fields_match_type` validator below. + source_type: SourceType + source_repo: Optional[str] = None + source_commit_sha: Optional[str] = None + source_committed_at: Optional[datetime] = None + feed_url: Optional[str] = None + post_guid: Optional[str] = None + # Locator + position of this chunk within its artifact. + locator_kind: LocatorKind + locator_path: str + span_index: int = Field(ge=0) + span_total: int = Field(ge=1) + span_heading_path: Optional[str] = None # JSON-encoded list[str] + # Payload + B's verdict. text: str - confidence: float = Field(ge=0, le=1) llm_label: str + confidence: float = Field(ge=0, le=1) llm_reasoning: Optional[str] = None - created_at: str - consumed_at: Optional[str] = None + created_at: datetime + consumed_at: Optional[datetime] = None + + @model_validator(mode="after") + def _source_fields_match_type(self) -> "KnowledgeQueueItem": + """Reject a row whose populated source fields contradict `source_type`. + + Checked here rather than in the adapter so the failure surfaces as one + typed boundary rejection: the adapter builds an RFC `SourceRef`, whose + own github rule would otherwise raise a raw Pydantic error from outside + the validation call. + """ + if self.source_type == SourceType.github and not ( + self.source_repo and self.source_commit_sha + ): + raise ValueError( + "source_type='github' requires source_repo and source_commit_sha" + ) + if self.source_type == SourceType.rss and not self.feed_url: + raise ValueError("source_type='rss' requires feed_url") + return self + + def heading_path(self) -> List[str]: + """`span_heading_path` decoded; empty when absent or not decodable. + + B stores A's heading list as a JSON string. A malformed value is a + cosmetic loss (it only feeds `title_hint`), so it degrades to empty + rather than rejecting a row that is otherwise linkable. + """ + if not self.span_heading_path: + return [] + try: + decoded = json.loads(self.span_heading_path) + except (ValueError, TypeError): + return [] + if not isinstance(decoded, list): + return [] + return [str(part) for part in decoded if str(part).strip()] # ---------- Golden dataset (internal, harness only) ---------- diff --git a/application/utils/librarian/section_validator.py b/application/utils/librarian/section_validator.py index 22927ba0f..1793ae211 100644 --- a/application/utils/librarian/section_validator.py +++ b/application/utils/librarian/section_validator.py @@ -9,19 +9,20 @@ Two entry points, one per upstream shape: -- ``section_from_queue_row`` — Module B's reduced ``knowledge_queue`` row - (master guide §1.2). The RFC identity fields C needs downstream are - synthesized from the row:: - - artifact_id = "art:{source_repo}:{source_path}" - chunk_id = "chk:{source_repo}@{source_commit_sha}:{source_path}" - - This chunk_id format differs from Module B's ``ChangeRecord`` - (``chk:art:{repo}:{path}:{index}``); align the two when the live - B->C pipeline wiring lands (W8). Not blocking W2 — no shared consumer yet. +- ``section_from_queue_row`` — Module B's live ``knowledge_queue`` row + (contract v0.2, table merged in #989). ``chunk_id`` and ``artifact_id`` + are **read straight off the row**: they are Module A's identity, carried + through B, and C consuming them verbatim is what lets a link join back to + the artifact it came from. + + Through W7 this function *synthesized* those ids from repo/path/sha, which + produced strings that matched nothing upstream. W8 removed that: the queue + row is the identity, and the source/locator shape now follows B's + ``source_type`` (``github`` carries repo+sha, ``rss`` carries a feed url) + rather than assuming every row is a GitHub commit. - ``section_from_knowledge_item`` — the full RFC ``KnowledgeItem`` - envelope (fixtures today; the live B->C path lands W8). + envelope (fixtures; B writes the flat queue row in practice). Volatile / audit-only metadata (``llm_reasoning``, ``filtered_at``, ``pipeline_run_id``, filter stages) is intentionally not carried into @@ -32,7 +33,7 @@ """ from dataclasses import dataclass -from typing import Any, Dict, Optional, Type, TypeVar, Union +from typing import Any, Callable, Dict, Optional, Type, TypeVar, Union from pydantic import BaseModel, ValidationError @@ -117,6 +118,87 @@ def _require_language(language: Optional[str]) -> str: return language +def _rfc_or_raise(build: Callable[[], _ModelT]) -> _ModelT: + """Build an RFC sub-model, converting its Pydantic error to a typed one. + + The row-level validator has already checked the field/`source_type` pairing, + but the RFC models carry rules of their own that B's column types cannot + express — ``SourceRef.commit_sha`` requires 7+ characters while Module A's + contract allows 4, and a ``feed_item`` locator requires a parseable URL. Any + such row is malformed *for C* and must leave as a SectionValidationError. + """ + try: + return build() + except ValidationError as exc: + raise MalformedKnowledgeItemError(str(exc)) from exc + + +def _source_ref(row: KnowledgeQueueItem) -> SourceRef: + """B's flat source columns -> the RFC source-ref, keyed on ``source_type``. + + ``source_committed_at`` is A's real commit time and is github-only; every + other source type falls back to ``created_at`` (B's classification time), + which is the best provenance the row carries. + """ + committed_at = row.source_committed_at or row.created_at + if row.source_type == SourceType.github: + return _rfc_or_raise( + lambda: SourceRef( + type=SourceType.github, + repo=row.source_repo, + commit_sha=row.source_commit_sha, + committed_at=committed_at, + ) + ) + # Validated from a mapping rather than constructed: `url` is an ``AnyUrl`` + # and B stores a plain string, so this routes the coercion (and any failure) + # through Pydantic, where ``_rfc_or_raise`` can type it. + return _rfc_or_raise( + lambda: SourceRef.model_validate( + { + "type": row.source_type, + "url": row.feed_url, + "committed_at": committed_at, + } + ) + ) + + +def _locator(row: KnowledgeQueueItem) -> Locator: + """B's ``locator_kind``/``locator_path`` -> the RFC locator. + + ``repo_path`` addresses by path; ``url``/``feed_item`` address by URL, where + the stable identity is the post guid when B carried one. + """ + if row.locator_kind == LocatorKind.repo_path: + return _rfc_or_raise( + lambda: Locator( + kind=LocatorKind.repo_path, + id=row.locator_path, + path=row.locator_path, + ) + ) + return _rfc_or_raise( + lambda: Locator.model_validate( + { + "kind": row.locator_kind, + "id": row.post_guid or row.locator_path, + "url": row.locator_path, + } + ) + ) + + +def _title_hint(row: KnowledgeQueueItem) -> Optional[str]: + """The deepest heading above this chunk, when A recorded one. + + Nothing downstream keys a decision on ``title_hint`` — retrieval reads + ``text`` — so this is provenance carried forward, not a scoring change. + """ + headings = row.heading_path() + return headings[-1] if headings else None + + def section_from_queue_row( row: Union[KnowledgeQueueItem, Dict[str, Any]], ) -> Section: @@ -132,27 +214,15 @@ def section_from_queue_row( ) _require_text(row.text) - # B's reduced row has no commit timestamp; created_at (B's classification - # time) is the best available provenance until the live B->C path lands. - source = SourceRef( - type=SourceType.github, - repo=row.source_repo, - commit_sha=row.source_commit_sha, - committed_at=row.created_at, - ) - locator = Locator( - kind=LocatorKind.repo_path, - id=row.source_path, - path=row.source_path, - ) return Section( - chunk_id=f"chk:{row.source_repo}@{row.source_commit_sha}:{row.source_path}", - artifact_id=f"art:{row.source_repo}:{row.source_path}", + # A's real identity, carried through B — never synthesized here. + chunk_id=row.chunk_id, + artifact_id=row.artifact_id, text=row.text, - title_hint=None, + title_hint=_title_hint(row), language=_DEFAULT_LANGUAGE, - source=source, - locator=locator, + source=_source_ref(row), + locator=_locator(row), ) diff --git a/cre.py b/cre.py index 5cbe86ae5..4dff20fc4 100644 --- a/cre.py +++ b/cre.py @@ -213,20 +213,28 @@ def main() -> None: parser.add_argument( "--run_librarian", action="store_true", - help="run Module C (the Librarian): for each knowledge-queue section, " - "resolve explicit CRE ids or retrieve the top-K semantic CRE candidates", + help="run Module C (the Librarian). With --run_id, drains Module B's " + "knowledge_queue for that run through C.0-C.4 and marks the rows " + "consumed; without it, walks a JSONL fixture and logs candidates", ) parser.add_argument( "--librarian_dry_run", action="store_true", - help="run the Librarian without writing any links (the only supported " - "mode pre-W8; logs the candidate shortlist per section)", + help="run the Librarian without persisting anything: no links (never " + "written pre-W8b) and no consumed_at stamp on the queue", ) parser.add_argument( "--librarian_source", default=None, help="path to a knowledge_queue JSONL for --run_librarian " - "(defaults to the bundled sample fixture)", + "(defaults to the bundled sample fixture; not valid with --run_id)", + ) + parser.add_argument( + "--librarian_envelopes_out", + default=None, + help="JSONL path the Librarian appends its LinkProposal / ReviewItem " + "envelopes to. Required for a real --run_id run: queue rows are only " + "marked consumed once their envelopes have been persisted", ) parser.add_argument( "--populate_neo4j_db", @@ -309,7 +317,8 @@ def main() -> None: parser.add_argument( "--run_id", default="", - help="pipeline_run_id to process (required with --run_noise_filter)", + help="pipeline_run_id to process (required with --run_noise_filter; " + "with --run_librarian, selects the live knowledge_queue path)", ) parser.add_argument( "--noise_filter_dry_run", @@ -322,6 +331,13 @@ def main() -> None: parser.error("--export requires --csv ") if args.run_noise_filter and not args.run_id.strip(): parser.error("--run_noise_filter requires --run_id ") + # The live queue path takes its rows from the DB, so a fixture path would be + # silently ignored rather than doing what the caller plainly asked for. + if args.librarian_source and args.run_id.strip(): + parser.error( + "--librarian_source reads a fixture and cannot be combined " + "with --run_id (which drains the live knowledge_queue)" + ) from application.cmd import cre_main diff --git a/docs/gsoc_2026_module_c/final_metrics.md b/docs/gsoc_2026_module_c/final_metrics.md new file mode 100644 index 000000000..c0c577738 --- /dev/null +++ b/docs/gsoc_2026_module_c/final_metrics.md @@ -0,0 +1,141 @@ +# Module C — final metrics + +Every number here comes from one command over the committed golden set. Nothing +is hand-copied from a notebook. + +```bash +python scripts/evaluate_librarian.py \ + --dataset application/tests/librarian/fixtures/golden_dataset.json \ + --use_live_embeddings --cache_file standards_cache.sqlite +``` + +**Environment:** 428 CRE hub vectors, `gemini/gemini-embedding-001` (dim 3072), +cross-encoder `ms-marco-MiniLM-L-6-v2`, hub-firewall ON, τ = 0.80. + +## The golden set + +319 hand-labelled rows drawn from OWASP standards already linked into OpenCRE. + +| Slice | Rows | What it tests | +|---|---|---| +| `positive` | 292 | a chunk that should link to a known CRE | +| `hard_negative` | 12 | plausible-looking chunks that must *not* link | +| `explicit` | 5 | chunks citing a CRE id — must resolve deterministically | +| `update` | 5 | chunks restating an existing link | +| `ambiguous` | 5 | chunks that must route to review | + +## Results + +### C.0 — input boundary + +| Slice | Validated | +|---|---| +| positive | 292/292 (100%) | +| hard_negative | 12/12 (100%) | +| explicit | 5/5 (100%) | +| update | 5/5 (100%) | +| ambiguous | 5/5 (100%) | + +Hub firewall stripped **319** leaking hub entries — one per row, as designed. The +golden standards are themselves linked into OpenCRE, so without the firewall every +row could retrieve itself and every metric below would be inflated. + +### C.0.5 — explicit resolver + +**5/5 — gate 100%: PASS.** A chunk that cites a CRE id resolves with no ML in the +path at all. This is a hard gate; the run fails if it is not perfect. + +### C.1 — candidate retrieval (recall@20) + +| Measure | Result | +|---|---| +| any-hit | 285/292 (**98%**) | +| all-hit | 274/292 (**94%**) | + +The correct CRE is in the shortlist 98% of the time. Retrieval is not the +bottleneck. + +### C.2 — cross-encoder rerank (top-1, top_n=5) + +**220/292 (75%).** + +### C.3 — confidence calibration + +| Measure | Result | +|---|---| +| fitted `T` | **1.105** | +| ECE raw (T=1) | 0.053 | +| ECE calibrated | **0.046** | +| Gate (ECE < 0.10) | **PASS** | + +Fitted on 304 rows (positive + hard_negative). A `T` near 1.0 means the reranker +was already close to honest; calibration tightened it rather than rescuing it. + +### C.4 — decision accuracy @ τ = 0.80 + +| Measure | Result | +|---|---| +| overall | 181/319 (57%) | +| auto-link recall (expected-linked) | 176/314 (56%) | +| **review recall (expected-review)** | **5/5 (100%)** | +| reason_code match | 4/5 (80%) | + +**Read this by direction — a single accuracy number hides the story.** + +*Review recall is 5/5.* Every chunk that should reach a human does. The engine +never wrongly auto-links something that needed review. For a gate whose job is +protecting the graph, this is the number that matters. + +*Auto-link recall is 56%.* At τ=0.80, many correct-but-close positives fall under +the bar and route to review instead. **That is the safe direction**: the cost is a +human looking at something the machine could have handled, not a wrong link +entering the graph. + +The one `reason_code` miss is a flag-based code that needs the SafetyGuard +detector, which is declared-not-built. + +## Against the plan's targets + +| Target | Planned | Actual | | +|---|---|---|---| +| C.0.5 explicit gate | 100% | 100% | ✅ | +| C.3 ECE | < 0.10 | 0.046 | ✅ | +| W4 top-1 | ≥ 0.80 | 0.75 | ❌ | +| W8 top-1 | ≥ 0.90 | 0.75 | ❌ | + +**The top-1 targets were not met, and the reason is now well understood.** + +Week 8's reranker investigation tried thirteen separate levers — model swaps, +prompt shapes, score fusion, candidate-pool changes. **All thirteen regressed.** +C.2 as shipped scores net −7 against plain cosine similarity on the same +shortlist. + +The root cause is not the model. **427 of 428 CREs have empty `description` +fields.** A cross-encoder scores a (query, document) pair; when the document side +is effectively just a title, there is almost nothing to cross-attend to. C.1 gets +98% recall from embeddings over that same thin corpus, so the information needed +to *rank* within the shortlist largely is not present. + +Week 9's selective-reranking experiment — gating C.2 to fire only where it helps +— reached 250/319 top-1 against the shipped 238, holding precision at τ=0.80. It +is deliberately not shipped: it needs two calibrators and held-out validation +that there was not time to do honestly. + +**The path to ≥0.90 runs through populating CRE descriptions, not through a +better reranker.** That is a corpus problem and belongs upstream of Module C. + +## Reproducing + +The hermetic subset needs no DB, key, or model, and is what CI gates on: + +```bash +python scripts/evaluate_librarian.py \ + --dataset application/tests/librarian/fixtures/golden_dataset.json +python -m pytest application/tests/librarian/ # 223 tests +``` + +For the live numbers, migrate a legacy cache first: + +```bash +python scripts/rewrite_sqlite_embeddings_to_vec.py --db standards_cache.sqlite +``` diff --git a/docs/gsoc_2026_module_c/final_report.md b/docs/gsoc_2026_module_c/final_report.md new file mode 100644 index 000000000..45c0580f5 --- /dev/null +++ b/docs/gsoc_2026_module_c/final_report.md @@ -0,0 +1,193 @@ +# Module C — The Librarian · GSoC 2026 final report + +**Project:** OWASP Integrated Ecosystem (OIE) — a living knowledge graph of OWASP +security knowledge +**Module:** C, the Librarian — the decision stage +**Contributor:** Prateek Singh +**Organisation:** OWASP · OpenCRE + +--- + +## What Module C does + +OpenCRE links security requirements across standards. Keeping those links current +by hand does not scale: OWASP repositories change constantly, and every change is +a candidate link nobody has time to review. + +The OIE pipeline automates the pass. Module A harvests changes, Module B filters +the noise, **Module C decides what each surviving chunk means**, and Module D puts +a human in front of what C could not decide alone. + +```text +A (harvester) ──▶ harvest_input ──▶ B (noise filter) ──▶ knowledge_queue ──▶ C (librarian) + │ + ┌─────────────┴─────────────┐ + LinkProposal ReviewItem + (auto-linked) (human review) ──▶ D +``` + +C's whole job is deciding **when not to decide**. Auto-linking a wrong CRE +pollutes a graph other tools read as truth; sending everything to a human defeats +the automation. The value is in the boundary between those two. + +--- + +## What was built + +Seven stages, each a separately reviewed and merged pull request. + +| Week | Stage | PR | What landed | +|---|---|---|---| +| 1 | C.-1 | [#922](https://github.com/OWASP/OpenCRE/pull/922) | RFC contracts, config, eval harness, 319-row golden dataset | +| 2 | C.0 | [#925](https://github.com/OWASP/OpenCRE/pull/925) | Input boundary — `SectionValidator`, `ExplicitLinkResolver` | +| 3 | C.1 | [#937](https://github.com/OWASP/OpenCRE/pull/937) | Candidate retriever (in-memory + pgvector) | +| 4 | C.2 | [#957](https://github.com/OWASP/OpenCRE/pull/957) | Cross-encoder reranker | +| 5 | C.3 | [#974](https://github.com/OWASP/OpenCRE/pull/974) | Confidence calibration — temperature scaling, ECE gate | +| 6 | C.4 | [#990](https://github.com/OWASP/OpenCRE/pull/990) | Decision engine — `decide()` | +| 6b | C.4 | [#991](https://github.com/OWASP/OpenCRE/pull/991) | Envelope emitter + C.0→C.4 pipeline glue | +| 7 | — | *analysis* | Auto-link threshold sweep; τ held at 0.80 | +| 8 | — | *this PR* | Live B→C integration — queue drain, sink, write-back | + +**223 tests**, all hermetic — no database, API key, or model download required. + +### Design decisions worth defending + +**Seams, not implementations.** C.1 takes an `embed_fn`, C.2 a `score_fn`, C.3 a +scaler, C.4 a safety guard — each a `Protocol`. Live components and test stubs are +interchangeable, so decision logic is tested without ever loading a model. This is +why the suite runs in under four seconds. + +**Nothing imports the database at module scope.** `factory.py` is the single +boundary where that stops being true, and even there the imports are +function-local. Hermetic testability was a constraint held for the whole project, +not an afterthought. + +**Declared-degraded, never silently-degraded.** The recurring failure mode this +project kept finding — in its own code — is a check that skips and reports +success. It was caught three times: + +1. The C.3 calibration gate returned 0 on a degenerate set, so a live run could + exit green without the ECE gate ever running. Now returns non-zero. +2. `NullSafetyGuard` evaluates nothing, and *says so* — its verdict carries + `evaluated=False`, counted and reported rather than looking clean. +3. When Module B's table shape moved, the eval harness rejected every row at the + C.0 boundary — so the explicit gate had nothing to count, skipped itself, and + the run still exited 0. Now a collapsed boundary fails the run explicitly. + +Each was a case of a gate that looked green while measuring nothing. + +**Consumption is gated on persistence.** Marking a queue row consumed tells +Module B never to offer that chunk again. Doing so while the envelope goes +nowhere destroys the chunk. The runner therefore refuses to retire anything +unless a sink reporting `persists=True` accepted the batch first. + +--- + +## Results + +Full numbers and methodology: [`final_metrics.md`](final_metrics.md). + +| Stage | Result | Target | | +|---|---|---|---| +| C.0 boundary validation | 319/319 (100%) | — | | +| C.0.5 explicit resolver | 5/5 (100%) | 100% | ✅ | +| C.1 retrieval recall@20 | 285/292 (98%) | — | | +| C.2 rerank top-1 | 220/292 (75%) | ≥ 90% | ❌ | +| C.3 calibration ECE | 0.046 (T=1.105) | < 0.10 | ✅ | +| C.4 review recall | 5/5 (100%) | — | ✅ | +| C.4 auto-link recall | 176/314 (56%) | — | | + +**The result I am most confident in is review recall: 5/5.** Every chunk that +should reach a human does. The engine never wrongly auto-links something that +needed review. For a component whose failure mode is polluting a shared graph, +that is the number that matters, and τ=0.80 buys it. + +**The target I missed is top-1 accuracy: 75% against a planned 90%.** I want to be +precise about why, because the honest answer is more useful than the flattering +one. + +Week 8's investigation tried thirteen reranker levers. All thirteen regressed — +C.2 as shipped scores net −7 against plain cosine similarity. The cause is not the +model: **427 of 428 CREs have empty `description` fields.** A cross-encoder scores +a (query, document) pair, and when the document side is effectively a bare title, +there is nothing to cross-attend to. C.1 still reaches 98% recall from embeddings +over that same thin corpus, which localises the problem precisely — the +information needed to *rank within* a shortlist mostly is not in the corpus. + +Week 9's selective-reranking experiment reached 250/319 top-1 versus the shipped +238 by gating C.2 to fire only where it helps. It is deliberately unshipped: it +needs two calibrators and held-out validation there was not time to do honestly. +Shipping it on in-sample numbers would have been exactly the greenwashing this +project spent three separate fixes eliminating. + +**The path to 90% runs through populating CRE descriptions, not through a better +reranker.** That is a corpus problem, upstream of Module C. + +--- + +## Integration status + +**B → C is live.** Module B writes `knowledge_queue`; C drains it, decides, and +stamps `consumed_at`. C's model matches B's table on all 23 columns. + +This did not start out true. B's merged table ([#989]) was substantially richer +than the flat mirror C had been built against — `chunk_id`, `artifact_id`, +`content_hash`, the `locator_*` and span columns, RSS provenance. C failed on +100% of real rows and fabricated chunk ids it should have carried through +verbatim. Reconciling that is the bulk of Week 8, and the ids are now used as A +minted them. + +**A → B is not yet connected.** Module A builds the harvester internals — repo +cloning, change detection, diff retrieval and parsing — but nothing writes +`harvest_input`, and A's `DiffBlock` is several transforms short of the +`ChangeRecord` shape B parses: chunking, id minting, run scoping. This is Module +A's remaining work, noted here because it means the end-to-end chain is not yet +demonstrable outside a fixture. + +**C → D is specified but unconsumed.** C emits `ReviewItem` envelopes against the +vendored RFC schema. Module D has no active implementation, so nothing reads +them today. + +--- + +## What is not built + +Stated plainly, because a known gap is worth more than a vague claim. + +- **Graph / review-queue writers (W8b).** C emits envelopes to JSONL; no link is + committed to the graph. The rule those writers must honour is already written + into `safety_guard.py`: a writer that commits links **must refuse to run behind + a guard reporting `evaluated=False`**. Retiring a queue row without the safety + path is recoverable — the envelope is still on disk. Committing a wrong link + into a graph other tools trust is not. +- **The SafetyGuard detector.** The seam is wired into `decide()`; the + out-of-distribution scoring, conformal prediction, and update detection behind + it are future work. +- **Selective reranking (W9).** Measured, promising, unvalidated. Do not ship + without held-out data. + +--- + +## Recommendations + +1. **Populate CRE descriptions.** The single highest-leverage change available. + It unblocks the top-1 target that no amount of reranker work reached. +2. **Do not lower τ below 0.80** without re-running the Week 7 sweep. The 100% + review recall is bought with it. +3. **Re-fit `T` after any retriever, reranker, or corpus change.** It is a + measured number, not a preference, and the live harness is the only place it + is produced. +4. **Build W8b's writers behind the safety guard**, not beside it. +5. **Finish A → B** before claiming an end-to-end pipeline. + +--- + +## Links + +- [Package README](../../application/utils/librarian/README.md) +- [Runbook](runbook.md) +- [Final metrics](final_metrics.md) +- [B → C contract](../gsoc_2026_module_b/module_c_contract.md) +- [OIE RFC #734](https://github.com/OWASP/OpenCRE/pull/734) + +[#989]: https://github.com/OWASP/OpenCRE/pull/989 diff --git a/docs/gsoc_2026_module_c/runbook.md b/docs/gsoc_2026_module_c/runbook.md new file mode 100644 index 000000000..d51449cee --- /dev/null +++ b/docs/gsoc_2026_module_c/runbook.md @@ -0,0 +1,178 @@ +# Module C (The Librarian) — runbook + +How to run Module C, what each knob does, and what to do when a run goes wrong. + +The package overview lives in +[`application/utils/librarian/README.md`](../../application/utils/librarian/README.md). +The table C reads from is specified in +[`module_c_contract.md`](../gsoc_2026_module_b/module_c_contract.md). + +--- + +## 1. The three ways to run C + +### a. Hermetic regression harness — no DB, no key, no model + +This is what CI runs and what you should run before every push. + +```bash +python scripts/evaluate_librarian.py \ + --dataset application/tests/librarian/fixtures/golden_dataset.json +``` + +Exercises the C.0 boundary and the C.0.5 explicit resolver over all 319 golden +rows. Exits non-zero if the explicit-slice gate fails, or if the boundary rejects +everything (see §5). + +The semantic reports stay off here deliberately: there are no CRE vectors +offline, and seeding the candidate pool from golden text is exactly the leakage +the hub firewall exists to strip. + +### b. Live evaluation — measures C.1 through C.4 + +Needs a populated embedding cache and an embedding-capable LLM. + +```bash +python scripts/evaluate_librarian.py \ + --dataset application/tests/librarian/fixtures/golden_dataset.json \ + --use_live_embeddings --cache_file standards_cache.sqlite +``` + +Adds four reports on top of the hermetic ones: C.1 recall@k, C.2 rerank top-1, +the C.3 ECE gate, and C.4 decision accuracy. All four share one retrieve+rerank +pass and one fitted `T` — the pipeline is built once per run, not once per +report. + +**Only the C.3 gate sets the exit status.** A failed *or skipped* ECE gate +returns non-zero, so a live run cannot pass without calibration having actually +run. C.4 is informational. + +### c. Live queue drain — the real pipeline + +```bash +# dry run first: reads, decides, persists nothing, retires nothing +python cre.py --run_librarian --librarian_dry_run --run_id + +# for real: envelopes are written, and only then are rows retired +python cre.py --run_librarian --run_id \ + --librarian_envelopes_out envelopes.jsonl +``` + +Or against a JSONL fixture instead of the live queue: + +```bash +python cre.py --run_librarian --librarian_dry_run \ + --librarian_source application/tests/librarian/fixtures/sample_knowledge_queue.jsonl +``` + +--- + +## 2. CLI flags + +| Flag | Meaning | +|---|---| +| `--run_librarian` | Run Module C. With `--run_id`, drains B's live `knowledge_queue`; without it, walks a JSONL fixture | +| `--run_id` | The `pipeline_run_id` to drain. Scopes the run to one orchestrator pass | +| `--librarian_dry_run` | Read and decide, but persist nothing and stamp nothing | +| `--librarian_source` | Path to a `knowledge_queue` JSONL. **Not valid with `--run_id`** — the CLI rejects the combination rather than silently ignoring one | +| `--librarian_envelopes_out` | JSONL path envelopes are appended to. **Required for a real `--run_id` run** | + +`--librarian_envelopes_out` is mandatory for a real run by design. Without a sink +that persists, retiring a row would destroy the chunk, so the runner refuses. + +--- + +## 3. Configuration + +All tunables are `CRE_LIBRARIAN_*` environment variables. + +| Variable | Default | What it is | +|---|---|---| +| `CRE_LIBRARIAN_CROSSENCODER_MODEL` | `cross-encoder/ms-marco-MiniLM-L-6-v2` | C.2 reranker model | +| `CRE_LIBRARIAN_RETRIEVER_BACKEND` | `in_memory` | `in_memory` or `pgvector` | +| `CRE_LIBRARIAN_TOP_K_RETRIEVAL` | `20` | C.1 shortlist size | +| `CRE_LIBRARIAN_TOP_K_RERANK` | `5` | How many C.2 re-sorts | +| `CRE_LIBRARIAN_LINK_THRESHOLD` | `0.8` | **τ** — the auto-link bar | +| `CRE_LIBRARIAN_TEMPERATURE` | `1.0` | **T** — C.3's fitted temperature | +| `CRE_LIBRARIAN_BATCH_SIZE` | `32` | Rows per batch | +| `CRE_LIBRARIAN_ECE_TARGET` | `0.10` | Calibration gate | + +**Two of these are measured, not chosen.** + +`CRE_LIBRARIAN_TEMPERATURE` is fitted by the live harness, which prints it. There +is nowhere else it is stored. The default of `1.0` is the identity transform — an +*uncalibrated* softmax that is honest about being unfitted rather than pretending +to a temperature nobody measured. **After any change to the retriever, the +reranker, or the CRE corpus, re-fit it and update the variable.** + +`CRE_LIBRARIAN_LINK_THRESHOLD` is τ, held at 0.80 by the Week 7 sweep. Lowering it +trades review precision for auto-link recall. The sweep found no better value — +do not lower it without re-running that analysis. + +--- + +## 4. Setup + +### The embedding cache + +C.1 retrieves over CRE-node vectors. After the pgvector migration (`c7d8e9f0a1b2`), +vectors live only in `embedding_vec`. A legacy `standards_cache.sqlite` with the +old CSV `embeddings` column will be refused by the ORM: + +```bash +python scripts/rewrite_sqlite_embeddings_to_vec.py --db standards_cache.sqlite +``` + +Rerun-safe: it reports `already embedding_vec-only; nothing to do` if applied. + +### The cross-encoder + +Downloaded from the HF Hub on first use. Set `HF_TOKEN` to avoid rate limits. + +--- + +## 5. Troubleshooting + +**`embeddings.embedding_vec is required; the legacy CSV embeddings column is no longer a supported store`** +Your cache predates the pgvector migration. Run the rewrite script in §4. + +**`validation (C.0): 0/319 rows validated ... FAILED (gates did not run)`** +Every golden row was rejected at the boundary. Almost always means the synthetic +row shape in `queue_row_from_golden` has drifted from Module B's live +`knowledge_queue`. This exact failure happened when B's table moved in #989 — the +adapter kept minting the old flat row. The guard exists because the *old* +behaviour was worse: with nothing validated, the explicit gate had nothing to +count, skipped itself, and the run still exited 0. + +**`--librarian_source reads a fixture and cannot be combined with --run_id`** +Pick one. A fixture path with a run id would silently ignore one of them. + +**A real run reports rows decided but nothing consumed** +The sink refused the batch, or you passed no `--librarian_envelopes_out`. +Consumption is gated on persistence — this is the safe failure, not a bug. + +**Rows keep reappearing across runs** +They errored mid-pipeline rather than being decided. Errored rows are left +unconsumed on purpose so the next run retries them. Check the logs for the chunk +and artifact id; `RunStats.errored` counts them separately from `skipped`. + +**ECE gate fails or is skipped** +A degenerate calibration set — all top-1 labels the same class — cannot identify +`T`. The harness fails rather than skipping, so this is reported, not hidden. +Widen the slice selection so both the positive and hard_negative slices are in. + +--- + +## 6. Tests + +```bash +# everything (hermetic — no DB, key, or model needed) +python -m pytest application/tests/librarian/ + +# the live-drain path specifically +python -m pytest application/tests/librarian/queue_runner_test.py \ + application/tests/librarian/queue_consumer_test.py +``` + +The queue tests use a real SQLAlchemy session with real row inserts and real +`consumed_at` assertions. Only the ML seams are stubbed. diff --git a/scripts/evaluate_librarian.py b/scripts/evaluate_librarian.py index 66ee69b1f..34b945ebc 100644 --- a/scripts/evaluate_librarian.py +++ b/scripts/evaluate_librarian.py @@ -26,6 +26,7 @@ """ import argparse +import hashlib import json import os import sys @@ -54,8 +55,13 @@ # Harness-only synthetic provenance: golden rows are not queue rows, so we # synthesize the minimum B-shaped row needed to exercise the C.0 boundary. +# The shape tracks Module B's live ``knowledge_queue`` (v0.2, merged in #989), +# not the flat pre-W8 mirror — if it drifts again, every row fails validation +# and the slice gates below stop grading anything. _SYNTHETIC_SHA = "0" * 40 _SYNTHETIC_CREATED_AT = "2026-06-01T00:00:00Z" +_SYNTHETIC_RUN_ID = "golden-harness" +_SCHEMA_VERSION = "0.2.0" def load_dataset(path: str) -> List[GoldenDatasetRow]: @@ -81,13 +87,39 @@ def load_dataset(path: str) -> List[GoldenDatasetRow]: def queue_row_from_golden(row: GoldenDatasetRow) -> dict: - """Adapt a golden row into the knowledge_queue shape C.0 validates.""" + """Adapt a golden row into the knowledge_queue shape C.0 validates. + + Golden rows carry no chunk or artifact identity — they predate the A->B->C + id-space — so the harness mints deterministic stand-ins from the row id. + Deterministic, not random: two runs over the same dataset must produce the + same ids, or the shared audits keyed off them stop lining up. + + Everything here is a synthetic github-repo-path row. The golden set is drawn + from checked-in OWASP standards, so ``repo_path`` is the honest locator kind; + an rss-shaped variant would exercise a provenance branch the dataset has no + examples of. + """ standard = row.input.source_standard.value if row.input.source_standard else "OTHER" + path = row.provenance.section_path or "unknown.md" + artifact_id = f"art:golden/{standard}:{path}" return { "id": row.id, + # The row id is already unique (load_dataset enforces it), so it is a + # sound dedup key without hashing the text. + "content_hash": hashlib.sha256(row.id.encode("utf-8")).hexdigest(), + "chunk_id": f"chk:{artifact_id}:0", + "artifact_id": artifact_id, + "pipeline_run_id": _SYNTHETIC_RUN_ID, + "schema_version": _SCHEMA_VERSION, + "source_type": "github", "source_repo": f"golden/{standard}", - "source_path": row.provenance.section_path or "unknown.md", "source_commit_sha": _SYNTHETIC_SHA, + "locator_kind": "repo_path", + "locator_path": path, + # One span per golden row: the dataset stores whole sections, already + # chunked by hand, so there is no second span to point at. + "span_index": 0, + "span_total": 1, "text": row.input.text, "confidence": 0.99, "llm_label": "KNOWLEDGE", @@ -502,6 +534,29 @@ def main(argv: List[str]) -> int: f"hub-firewall: {'ON' if firewall_on else 'OFF'}; " f"stripped {stripped} leaking hub entries" ) + # A boundary that rejects everything silently un-grades every report below + # it: nothing reaches the resolver, so the explicit gate has nothing to + # count, reports PASS-by-vacuum, and the run exits 0. That is the same + # skipped-gate-reports-success failure the C.3 calibration gate was fixed + # for, and it is how a knowledge_queue schema drift would slip through. + validated_total = sum(validated_per_slice.values()) + if rows and not validated_total: + print( + f"validation (C.0): 0/{len(rows)} rows validated — the whole golden " + "set was rejected at the boundary, so no report below this line " + "graded anything; FAILED (gates did not run). Most likely the " + "synthetic row shape has drifted from Module B's knowledge_queue" + ) + return 1 + + explicit_expected = per_slice.get("explicit", 0) + if explicit_expected and not explicit_total: + print( + f"explicit slice (C.0.5 resolver): 0/{explicit_expected} reached the " + "resolver — every explicit row was rejected at the C.0 boundary; " + "FAILED (gate did not run)" + ) + return 1 if explicit_total: gate_ok = explicit_correct == explicit_total print(