diff --git a/.gitignore b/.gitignore index bfd3699d3..7f6e1f3d9 100644 --- a/.gitignore +++ b/.gitignore @@ -72,6 +72,7 @@ standards_cache.sqlite !docs/Mid_eval_blog_gsoc2026/module_B_mideval_blog.md !docs/Final_eval_blog_gsoc2026/module_B_final_blog.md !application/utils/librarian/README.md +!application/utils/reviewer/README.md !docs/gsoc_2026_module_c/*.md !docs/gsoc_2026_module_a/*.md !docs/Final_eval_blog_gsoc2026/*.md diff --git a/application/tests/reviewer/__init__.py b/application/tests/reviewer/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/application/tests/reviewer/_envelopes.py b/application/tests/reviewer/_envelopes.py new file mode 100644 index 000000000..a52761f7e --- /dev/null +++ b/application/tests/reviewer/_envelopes.py @@ -0,0 +1,85 @@ +"""Shared fixtures for Module D tests. + +Rows are seeded through the real ``DbEnvelopeSink`` wherever possible, so these +tests exercise the actual C -> D handoff: D reads exactly what C writes, or the +test fails. The envelope builders mirror the librarian's own test fixtures. +""" + +from datetime import datetime, timezone + +from application.utils.librarian.schemas import ( + SCHEMA_VERSION, + CreCandidate, + KnowledgeSnapshot, + LinkProposal, + Locator, + ProposedLink, + ReasonCode, + RetrievalAudit, + ReviewItem, + SourceRef, + UpdateDetection, +) + +AT = datetime(2026, 9, 17, 12, 0, 0, tzinfo=timezone.utc) +RUN = "run-1" + + +def _knowledge() -> KnowledgeSnapshot: + return 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"), + ) + + +def _audit() -> RetrievalAudit: + return 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, + ) + + +def review(chunk_id: str = "chk:1", run: str = RUN) -> ReviewItem: + return ReviewItem( + schema_version=SCHEMA_VERSION, + review_id=f"review:{chunk_id}", + chunk_id=chunk_id, + artifact_id="art:1", + pipeline_run_id=run, + created_at=AT, + reason_code=ReasonCode.below_threshold, + knowledge=_knowledge(), + retrieval=_audit(), + suggested_links=[ + ProposedLink(cre_id="616-305", link_type="Related to", confidence=0.42) + ], + update_detection=UpdateDetection(is_update=False), + ) + + +def linked(chunk_id: str = "chk:9", run: str = RUN) -> LinkProposal: + return LinkProposal( + schema_version=SCHEMA_VERSION, + chunk_id=chunk_id, + artifact_id="art:1", + pipeline_run_id=run, + classified_at=AT, + knowledge=_knowledge(), + retrieval=_audit(), + links=[ + ProposedLink( + cre_id="616-305", + link_type="Automatically linked to", + confidence=0.95, + ) + ], + update_detection=UpdateDetection(is_update=False), + ) diff --git a/application/tests/reviewer/corrections_log_test.py b/application/tests/reviewer/corrections_log_test.py new file mode 100644 index 000000000..8a41ef58f --- /dev/null +++ b/application/tests/reviewer/corrections_log_test.py @@ -0,0 +1,192 @@ +"""Tests for the corrections log: the durable half of a decision. + +The property under test is not formatting, it is the retirement gate: what this +log records is what permits ``mark_reviewed`` to run, so a record must land +whole, survive reopening, match the pinned ``HumanDecision`` schema, and refuse +a decision that contradicts itself. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +import json +import os +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from unittest import mock + +from jsonschema import Draft202012Validator, FormatChecker +from pydantic import ValidationError +from referencing import Registry, Resource + +from application.utils.librarian.schemas import ProposedLink +from application.utils.reviewer import corrections_log +from application.utils.reviewer.corrections_log import ( + HumanDecision, + JsonlCorrectionsLog, + LossEvent, + RejectionDetail, + UiMeta, +) + +AT = datetime(2026, 9, 17, 12, 0, 0, tzinfo=timezone.utc) +LINK = ProposedLink(cre_id="616-305", link_type="Related", confidence=1.0) +SCHEMA_DIR = os.path.join( + os.path.dirname(__file__), + "..", + "..", + "..", + "docs", + "owasp-graph", + "apis", + "schemas", +) + + +def _decision(**overrides) -> HumanDecision: + values = dict( + review_id="review:chk:1", + chunk_id="chk:1", + pipeline_run_id="run-1", + decided_at=AT, + decided_by="prateek", + action="approve", + final_links=[LINK], + ) + values.update(overrides) + return HumanDecision(**values) + + +def _rejected(**overrides) -> HumanDecision: + return _decision( + action="reject", + final_links=None, + rejection=RejectionDetail(reason_code="NOT_SECURITY"), + **overrides, + ) + + +def _validator() -> Draft202012Validator: + schemas = [] + for name in ("human-decision.json", "proposed-link.json"): + with open(os.path.join(SCHEMA_DIR, name), encoding="utf-8") as fh: + schemas.append(json.load(fh)) + registry = Registry().with_resources( + [(s["$id"], Resource.from_contents(s)) for s in schemas] + ) + return Draft202012Validator( + schemas[0], registry=registry, format_checker=FormatChecker() + ) + + +class HumanDecisionTest(unittest.TestCase): + def test_all_three_contract_actions_are_representable(self) -> None: + self.assertEqual("approve", _decision().action) + self.assertEqual("correct", _decision(action="correct").action) + self.assertEqual("reject", _rejected().action) + + def test_unknown_action_is_refused(self) -> None: + with self.assertRaises(ValidationError): + _decision(action="maybe") + + def test_contradictory_decisions_are_refused(self) -> None: + for bad in ( + dict(final_links=None), + dict(final_links=[]), + dict(rejection=RejectionDetail(reason_code="X")), + dict(action="reject"), + dict(action="reject", rejection=RejectionDetail(reason_code="X")), + dict(decided_by=""), + dict(decision_id="not-a-uuid"), + dict(schema_version="1.0.0"), + ): + with self.assertRaises(ValidationError, msg=bad): + _decision(**bad) + + +class JsonlCorrectionsLogTest(unittest.TestCase): + def setUp(self) -> None: + self.dir = tempfile.TemporaryDirectory() + self.path = os.path.join(self.dir.name, "nested", "corrections.jsonl") + + def tearDown(self) -> None: + self.dir.cleanup() + + def _lines(self): + with open(self.path, encoding="utf-8") as fh: + return [json.loads(line) for line in fh] + + def test_every_line_matches_the_pinned_schema(self) -> None: + """A dump with nulls or extra keys still looks like JSONL; only + validating against the contract can catch it.""" + log = JsonlCorrectionsLog(self.path) + log.record(_decision()) + log.record(_rejected()) + log.record( + _decision( + action="correct", + ui=UiMeta(review_duration_ms=2100, keybind_used="c"), + loss_event=LossEvent( + module="C", + input_snapshot={"text": "x"}, + wrong_prediction={"cre_id": "111-111"}, + correct_label={"cre_id": "616-305"}, + ), + ) + ) + validator = _validator() + for line in self._lines(): + self.assertEqual([], [e.message for e in validator.iter_errors(line)]) + + def test_each_decision_gets_its_own_id(self) -> None: + log = JsonlCorrectionsLog(self.path) + log.record(_decision()) + log.record(_decision()) + first, second = self._lines() + self.assertNotEqual(first["decision_id"], second["decision_id"]) + + def test_survives_reopening_the_log(self) -> None: + """Two service restarts, one file: earlier decisions must remain.""" + JsonlCorrectionsLog(self.path).record(_decision(review_id="r1")) + JsonlCorrectionsLog(self.path).record(_decision(review_id="r2")) + + self.assertEqual([line["review_id"] for line in self._lines()], ["r1", "r2"]) + + def test_decided_at_is_recorded_in_utc(self) -> None: + ist = timezone(timedelta(hours=5, minutes=30)) + JsonlCorrectionsLog(self.path).record(_decision(decided_at=AT.astimezone(ist))) + + (line,) = self._lines() + self.assertEqual(line["decided_at"], "2026-09-17T12:00:00Z") + + def test_a_torn_last_line_does_not_swallow_the_next_record(self) -> None: + log = JsonlCorrectionsLog(self.path) + log.record(_decision(review_id="r1")) + with open(self.path, "a", encoding="utf-8") as fh: + fh.write('{"review_id": "torn') # a crash mid-write + log.record(_decision(review_id="r3")) + + with open(self.path, encoding="utf-8") as fh: + lines = fh.read().splitlines() + self.assertEqual(3, len(lines)) + self.assertEqual("r3", json.loads(lines[2])["review_id"]) + + def test_directory_entry_is_synced_up_the_chain(self) -> None: + opened = [] + real_open = os.open + + def tracking_open(path, flags, *args, **kwargs): + opened.append(path) + return real_open(path, flags, *args, **kwargs) + + with mock.patch.object(corrections_log.os, "open", tracking_open): + JsonlCorrectionsLog(self.path).record(_decision()) + parent = os.path.abspath(os.path.dirname(self.path)) + self.assertIn(parent, opened) + self.assertIn(os.path.dirname(parent), opened) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/reviewer/queue_consumer_test.py b/application/tests/reviewer/queue_consumer_test.py new file mode 100644 index 000000000..d723a3b03 --- /dev/null +++ b/application/tests/reviewer/queue_consumer_test.py @@ -0,0 +1,137 @@ +"""Tests for D's write-back: the stamp that retires a review row. + +The claim that matters most: a ``linked`` row can NEVER be stamped by this path, +whatever id it is handed. That guard is what makes the human queue structurally +unable to retire the graph writer's rows. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +import unittest +from datetime import datetime, timezone +from unittest.mock import MagicMock + +from application import create_app, sqla +from application.database.db import DecisionQueueItem +from application.utils.librarian.envelope_sink import DbEnvelopeSink +from application.utils.reviewer.corrections_log import HumanDecision, RejectionDetail +from application.utils.reviewer.queue_consumer import mark_reviewed, record_and_retire +from application.utils.reviewer.review_source import DbReviewSource +from application.tests.reviewer._envelopes import RUN, linked, review + +AT = datetime(2026, 9, 17, 12, 0, 0, tzinfo=timezone.utc) + + +class MarkReviewedTest(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.ctx = self.app.app_context() + self.ctx.push() + sqla.create_all() + DbEnvelopeSink(sqla.session, RUN).write([review("chk:1"), linked("chk:9")]) + sqla.session.commit() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.ctx.pop() + + def _row(self, chunk_id: str) -> DecisionQueueItem: + return ( + sqla.session.query(DecisionQueueItem) + .filter(DecisionQueueItem.chunk_id == chunk_id) + .one() + ) + + def test_stamps_review_rows_idempotently(self) -> None: + rid = self._row("chk:1").id + + self.assertEqual(mark_reviewed(sqla.session, [rid], at=AT), 1) + sqla.session.commit() + # A replay cannot move a timestamp that is already set. + self.assertEqual(mark_reviewed(sqla.session, [rid], at=AT), 0) + self.assertIsNotNone(self._row("chk:1").consumed_at) + + def test_a_linked_row_can_never_be_stamped_by_the_review_path(self) -> None: + """linked rows are the graph writer's. Even handed the right id, the + human-review write-back must refuse — silently, counted in the gap + between ids passed and rows stamped.""" + rid = self._row("chk:9").id + + self.assertEqual(mark_reviewed(sqla.session, [rid], at=AT), 0) + sqla.session.commit() + + self.assertIsNone(self._row("chk:9").consumed_at) + + def test_stamp_is_stored_naive_utc(self) -> None: + """Aware datetime in, naive UTC stored — the pipeline-wide convention, + dialect-portable.""" + rid = self._row("chk:1").id + mark_reviewed(sqla.session, [rid], at=AT) + sqla.session.commit() + + stored = self._row("chk:1").consumed_at + self.assertIsNone(stored.tzinfo) + self.assertEqual(stored, AT.astimezone(timezone.utc).replace(tzinfo=None)) + + def test_empty_and_falsy_ids_are_a_no_op(self) -> None: + self.assertEqual(mark_reviewed(sqla.session, [], at=AT), 0) + self.assertEqual(mark_reviewed(sqla.session, ["", None], at=AT), 0) + + +class RecordAndRetireTest(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.ctx = self.app.app_context() + self.ctx.push() + sqla.create_all() + sink = DbEnvelopeSink(sqla.session, RUN) + sink.write([review("chk:1")]) + sqla.session.commit() + self.row = sqla.session.query(DecisionQueueItem).one() + (self.review,) = DbReviewSource(sqla.session).items() + self.log = MagicMock() + self.decision = HumanDecision( + review_id="review:chk:1", + chunk_id="chk:1", + pipeline_run_id=RUN, + decided_at=AT, + decided_by="prateek", + action="reject", + rejection=RejectionDetail(reason_code="NOT_SECURITY"), + ) + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.ctx.pop() + + def test_records_then_stamps(self) -> None: + stamped = record_and_retire( + sqla.session, self.log, self.decision, self.review, at=AT + ) + self.assertEqual(1, stamped) + self.log.record.assert_called_once_with(self.decision) + sqla.session.refresh(self.row) + self.assertIsNotNone(self.row.consumed_at) + + def test_a_failed_record_leaves_the_row_unretired(self) -> None: + self.log.record.side_effect = OSError("disk full") + with self.assertRaises(OSError): + record_and_retire(sqla.session, self.log, self.decision, self.review, at=AT) + sqla.session.refresh(self.row) + self.assertIsNone(self.row.consumed_at) + + def test_a_decision_about_another_row_changes_nothing(self) -> None: + other = self.decision.model_copy(update={"chunk_id": "chk:2"}) + with self.assertRaises(ValueError): + record_and_retire(sqla.session, self.log, other, self.review, at=AT) + self.log.record.assert_not_called() + sqla.session.refresh(self.row) + self.assertIsNone(self.row.consumed_at) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/reviewer/review_source_test.py b/application/tests/reviewer/review_source_test.py new file mode 100644 index 000000000..2ae7e91e2 --- /dev/null +++ b/application/tests/reviewer/review_source_test.py @@ -0,0 +1,174 @@ +"""Tests for Module D's read side over ``decision_queue``. + +Rows are seeded through the real ``DbEnvelopeSink``, so what these tests prove +is the actual handoff: the envelope C persisted validates back into the same +RFC model on D's side. The behaviour that matters is what the source refuses to +read — ``linked`` rows (the graph writer's), consumed rows, other runs' rows — +and that a row D cannot model is surfaced, skipped, and never retired. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +import unittest +from datetime import datetime +from unittest.mock import patch + +from application import create_app, sqla +from application.database.db import DecisionQueueItem +from application.utils.librarian.envelope_sink import DbEnvelopeSink +from application.utils.reviewer.queue_consumer import mark_reviewed +from application.utils.reviewer.review_source import DbReviewSource +from application.tests.reviewer._envelopes import RUN, linked, review + + +def _seed(*envelopes, run: str = RUN) -> None: + sink = DbEnvelopeSink(sqla.session, run) + sink.write(list(envelopes)) + sqla.session.commit() + + +class DbReviewSourceTest(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_only_review_required_rows(self) -> None: + """linked rows belong to the graph writer; a human inbox that showed + them would invite retiring an auto-link by hand.""" + _seed(review("chk:1"), linked("chk:9")) + + items = list(DbReviewSource(sqla.session).items()) + + self.assertEqual([i.chunk_id for i in items], ["chk:1"]) + + def test_envelope_round_trips_through_the_real_sink(self) -> None: + """What C persisted validates back into the same pinned RFC model.""" + _seed(review("chk:1")) + + (item,) = DbReviewSource(sqla.session).items() + + self.assertEqual(item.envelope.review_id, "review:chk:1") + self.assertEqual(item.envelope.reason_code.value, "BELOW_THRESHOLD") + self.assertEqual(item.envelope.suggested_links[0].cre_id, "616-305") + self.assertEqual(item.reason_code, "BELOW_THRESHOLD") + self.assertEqual(item.source_label, None) + + def test_consumed_rows_are_not_re_read(self) -> None: + from datetime import datetime + + _seed(review("chk:1"), review("chk:2")) + row = ( + sqla.session.query(DecisionQueueItem) + .filter(DecisionQueueItem.chunk_id == "chk:1") + .one() + ) + row.consumed_at = datetime(2026, 9, 17) + sqla.session.commit() + + items = list(DbReviewSource(sqla.session).items()) + + self.assertEqual([i.chunk_id for i in items], ["chk:2"]) + + def test_scopes_to_one_pipeline_run(self) -> None: + _seed(review("chk:1")) + _seed(review("chk:2", run="run-2"), run="run-2") + + items = list(DbReviewSource(sqla.session, pipeline_run_id="run-2").items()) + + self.assertEqual([i.chunk_id for i in items], ["chk:2"]) + + def test_limit_is_applied_in_a_stable_order(self) -> None: + _seed(review("chk:3"), review("chk:1"), review("chk:2")) + + first = [i.row_id for i in DbReviewSource(sqla.session, limit=2).items()] + again = [i.row_id for i in DbReviewSource(sqla.session, limit=2).items()] + + self.assertEqual(len(first), 2) + self.assertEqual(first, again) + + def test_unreadable_envelope_is_skipped_reported_and_never_consumed(self) -> None: + """A decision row D cannot parse is C's audit record of a decision. + Retiring it would hide a C-side contract breach from the one queue a + human watches, so it is surfaced instead — the opposite of C's + poison-row policy, deliberately.""" + _seed(review("chk:1")) + sqla.session.add( + DecisionQueueItem( + chunk_id="chk:bad", + artifact_id="art:1", + pipeline_run_id=RUN, + schema_version="0.2.0", + status="review_required", + envelope={"not": "an envelope"}, + ) + ) + sqla.session.commit() + + source = DbReviewSource(sqla.session) + with self.assertLogs( + "application.utils.reviewer.review_source", level="WARNING" + ): + items = list(source.items()) + + self.assertEqual([i.chunk_id for i in items], ["chk:1"]) + self.assertEqual(len(source.unreadable_row_ids), 1) + bad = sqla.session.get(DecisionQueueItem, source.unreadable_row_ids[0]) + self.assertIsNone(bad.consumed_at) + + def _spoil_first(self, *chunk_ids: str) -> None: + """Make rows unreadable and oldest, so they head every page.""" + for chunk_id in chunk_ids: + row = sqla.session.query(DecisionQueueItem).filter_by(chunk_id=chunk_id) + row.update({"envelope": {"bad": True}, "created_at": datetime(2000, 1, 1)}) + sqla.session.commit() + + def test_unreadable_rows_cannot_starve_a_limited_page(self) -> None: + _seed(*[review(f"chk:{i}") for i in range(5)]) + self._spoil_first("chk:0", "chk:1") + + source = DbReviewSource(sqla.session, limit=2) + items = list(source.items()) + + self.assertEqual(2, len(items)) + self.assertEqual(2, len(source.unreadable_row_ids)) + + def test_walks_past_a_batch_boundary(self) -> None: + _seed(*[review(f"chk:{i}") for i in range(7)]) + with patch("application.utils.reviewer.review_source._BATCH_SIZE", 3): + items = list(DbReviewSource(sqla.session).items()) + self.assertEqual(7, len({i.row_id for i in items})) + + def test_a_row_stamped_during_the_walk_is_not_yielded(self) -> None: + _seed(*[review(f"chk:{i}") for i in range(4)]) + with patch("application.utils.reviewer.review_source._BATCH_SIZE", 2): + source = DbReviewSource(sqla.session).items() + first = next(source) + pending = sqla.session.query(DecisionQueueItem).filter( + DecisionQueueItem.id != first.row_id + ) + last_id = sorted(r.id for r in pending)[-1] + mark_reviewed(sqla.session, [last_id], at=datetime(2026, 1, 1)) + rest = [i.row_id for i in source] + self.assertEqual(2, len(rest)) + self.assertNotIn(last_id, rest) + + def test_reading_twice_does_not_double_count_unreadable_rows(self) -> None: + _seed(review("chk:0"), review("chk:1")) + self._spoil_first("chk:0") + source = DbReviewSource(sqla.session) + list(source.items()) + list(source.items()) + self.assertEqual(1, len(source.unreadable_row_ids)) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/reviewer/README.md b/application/utils/reviewer/README.md new file mode 100644 index 000000000..805727802 --- /dev/null +++ b/application/utils/reviewer/README.md @@ -0,0 +1,79 @@ +# Module D: The Reviewer + +The fourth stage of the OIE pipeline: human-in-the-loop review over +`decision_queue`. Module C writes every decision there; the rows it could not +decide alone arrive as `status='review_required'` with a full RFC `ReviewItem` +envelope. This package consumes exactly those rows and nothing else. + +```text +A (harvester) -> B (filter) -> C (librarian) -> decision_queue -> D (reviewer) + | + linked rows -> graph writer (not D) +``` + +## The two rules + +**`linked` rows are not D's.** They belong to the graph writer. Every read +(`DbReviewSource`) and every write-back (`mark_reviewed`) is filtered on +`status='review_required'`, so a human queue is structurally unable to retire an +auto-link, even handed the right id. + +**The decision is persisted before the row is retired.** `record_and_retire` +records the human's decision (fsync'd) through `JsonlCorrectionsLog` first and +stamps `consumed_at` second. A crash between the two leaves the row unconsumed +and the decision re-recordable, never the reverse. A re-record appends a second +line for the same row; readers take the last one per `(review_id, +pipeline_run_id)`, because `review_id` alone repeats across runs. This is the same rule +Module C follows against Module B's queue, one handoff downstream. + +## What is in this package (D0) + +| piece | file | what it does | +|---|---|---| +| Read side | `review_source.py` | `DbReviewSource` yields `PendingReview` rows, envelope validated back through the same pinned RFC model C emitted it with | +| Write-back | `queue_consumer.py` | `mark_reviewed` stamps `consumed_at`, idempotent, review rows only; `record_and_retire` is the fixed order | +| Corrections log | `corrections_log.py` | append-only JSONL, one `HumanDecision` per line as pinned in `docs/owasp-graph/apis/schemas/human-decision.json`, the RFC's "no db bloat" decision | + +A row whose envelope does not validate is skipped, logged by id, surfaced in +`unreadable_row_ids`, and **left unconsumed**. That is deliberately the opposite of C's +poison-row policy, because a decision row D cannot parse is C's audit record of +a decision, and retiring it would hide a C-side contract breach from the one +queue a human watches. + +## The click-speed prototype + +`docs/gsoc_2026_module_d/click_speed_prototype.html` is the founding RFC's +pre-code experiment: open it in a browser, review ten fake items with `y`, `n` +and `c` (press space to start), and it measures whether the flow stays under the RFC's 3-second bar. It +passes only if the average and every single item are under 3s and at least 8 of +10 answers are right, so mashing a key or undoing a slow item does not count. +No measured human run is recorded yet; record one here before D1 is held to it. + +## Not built yet + +- **Where the log lives.** `JsonlCorrectionsLog` writes a local file, and a Heroku + dyno's disk does not survive a restart, while `consumed_at` lives in Postgres. + D1 must not run on a dyno until the log is on durable storage (the RFC's S3/Blob + target), or rows would stay retired with their decisions gone. +- **The review flow (D1).** Endpoints in the existing `/admin/` pattern wiring + source -> human -> corrections log -> `mark_reviewed`. +- **The graph writer behind review (D2).** Approved rows go to a staging + changeset (per `docs/designs/easier-importing.md`) and are applied through + the same accept/apply pattern the imports admin API already uses. A human + verdict is the safety evaluation that the W8 rule requires; the automated + SafetyGuard detector (D3) comes last and lets high-confidence `linked` rows + earn the right to skip the human. + +## Contracts + +- **C -> D:** [`docs/gsoc_2026_module_c/module_d_contract.md`](../../../docs/gsoc_2026_module_c/module_d_contract.md) +- The envelope model is imported from `application.utils.librarian.schemas`, + so C and D cannot drift on what a `ReviewItem` is without one side failing loudly. + +## Running the tests + +```bash +python -m pytest application/tests/reviewer/ -q +``` + +Hermetic: in-memory SQLite, no API key, no model download. diff --git a/application/utils/reviewer/__init__.py b/application/utils/reviewer/__init__.py new file mode 100644 index 000000000..14cc98d40 --- /dev/null +++ b/application/utils/reviewer/__init__.py @@ -0,0 +1,18 @@ +"""Module D — The Reviewer: human-in-the-loop review over ``decision_queue``. + +The fourth stage of the OIE pipeline. Module C writes every decision to +``decision_queue``; the rows it could not decide alone carry +``status='review_required'`` and a full RFC ``ReviewItem`` envelope. This +package is the consumer of exactly those rows and nothing else. + +Two rules, both inherited from the contract +(``docs/gsoc_2026_module_c/module_d_contract.md``): + +- **``linked`` rows are not D's.** They belong to the graph writer. Every read + and every write-back in this package is filtered on + ``status='review_required'`` so a human queue can never retire an auto-link. +- **The decision is persisted before the row is retired.** Stamping + ``consumed_at`` tells the pipeline this chunk is finished; doing it before the + human's decision is durably recorded would destroy the one thing the review + produced. Same rule Module C follows against Module B's queue. +""" diff --git a/application/utils/reviewer/corrections_log.py b/application/utils/reviewer/corrections_log.py new file mode 100644 index 000000000..9533dc730 --- /dev/null +++ b/application/utils/reviewer/corrections_log.py @@ -0,0 +1,180 @@ +"""The corrections log: every human decision, append-only, one JSON per line. + +The founding RFC (``docs/designs/owasp-pane-of-glass.md``, Module D) decides +this shape: "simple human oversight without db bloat", so decisions append to a +JSONL log rather than growing new tables. Each line is a ``HumanDecision``, the +D -> feedback contract pinned in ``docs/owasp-graph/apis/module-d-hitl.md`` and +``schemas/human-decision.json``, and a test validates real lines against that +schema. ``chunk_id`` plus ``pipeline_run_id`` joins a line back to its one +``decision_queue`` row (``review_id`` alone repeats across runs), and those rows +are never deleted, so (input, prediction, human label) stays reconstructable even +for a line that carries no ``loss_event``. + +Replaying a review appends a second line for the same row. The log stays +append-only, so readers take the **last** line per ``(review_id, pipeline_run_id)``. + +This log is the durable half of the two-step verdict flow: **record here first, +stamp ``consumed_at`` second.** A crash in between leaves the row unconsumed +and the verdict re-recordable. The reverse order would retire the row and lose +the one thing the human produced. + +Like C's ``JsonlEnvelopeSink``, this makes no cross-process claims: two +services appending to one path can interleave. One reviewer service per log +file, which is the same single-consumer contract the queue itself carries. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Literal, Optional + + +def _fsync_dir_chain(path: str) -> None: + """fsync every directory from ``path`` up to the filesystem root. + + ``os.fsync`` on the file makes its *contents* durable, not its *directory + entry*: a crash after creating the file can lose the file itself unless the + parent directory is synced, and the same holds for each directory + ``os.makedirs`` created above it. Which ancestors are newly created cannot + be detected without racing a concurrent ``record()``, so the whole chain is + synced on every record. Records are human-paced, so the cost is noise. + + Platforms that cannot fsync a directory (or deny it, e.g. on ``/``) raise + ``OSError``; those levels are skipped, which leaves durability no worse + than it was before the attempt. + """ + current = os.path.abspath(path) + while True: + try: + fd = os.open(current, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + except OSError: + pass + parent = os.path.dirname(current) + if parent == current: + return + current = parent + + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from application.utils.librarian.schemas import SCHEMA_VERSION, ProposedLink + + +class RejectionDetail(BaseModel): + model_config = ConfigDict(extra="forbid") + + reason_code: str + reason_message: Optional[str] = None + + +class LossEvent(BaseModel): + """Pro-mode training capture: what the pipeline got wrong, and the fix.""" + + model_config = ConfigDict(extra="allow") + + module: Literal["B", "C"] + input_snapshot: Dict[str, Any] + wrong_prediction: Dict[str, Any] + correct_label: Dict[str, Any] + notes: Optional[str] = None + + +class UiMeta(BaseModel): + """Timing for the RFC's 3-second experiment.""" + + model_config = ConfigDict(extra="allow") + + review_duration_ms: Optional[int] = Field(default=None, ge=0) + keybind_used: Optional[str] = None + + +class HumanDecision(BaseModel): + """One human decision about one review item (``human-decision.json``).""" + + model_config = ConfigDict(extra="forbid") + + schema_version: str = Field(default=SCHEMA_VERSION, pattern=r"^0\.\d+\.\d+$") + decision_id: uuid.UUID = Field(default_factory=uuid.uuid4) + review_id: str + chunk_id: str + pipeline_run_id: str + decided_at: datetime + decided_by: str = Field(min_length=1) + action: Literal["approve", "reject", "correct"] + #: Ground truth CRE links, on approve and correct. + final_links: Optional[List[ProposedLink]] = None + #: Why, on reject. + rejection: Optional[RejectionDetail] = None + loss_event: Optional[LossEvent] = None + ui: Optional[UiMeta] = None + + @model_validator(mode="after") + def _action_matches_its_payload(self) -> "HumanDecision": + if self.action == "reject": + if self.rejection is None or self.final_links: + raise ValueError("reject needs rejection and no final_links") + elif not self.final_links or self.rejection is not None: + raise ValueError(f"{self.action} needs final_links and no rejection") + return self + + +class JsonlCorrectionsLog: + """Appends decisions to a JSONL file, creating parent directories as needed.""" + + def __init__(self, path: str) -> None: + self._path = path + + @property + def path(self) -> str: + return self._path + + def record(self, decision: HumanDecision) -> None: + """Append one decision durably (flush + fsync before returning). + + The stamp on decision_queue happens only after this returns, so this + write IS the persistence the retirement is gated on, the same rule as + C's "consumption is gated on persistence", one queue downstream. + """ + parent = os.path.dirname(self._path) + if parent: + os.makedirs(parent, exist_ok=True) + # The schema has no nulls: an absent value must be an absent key. + payload = decision.model_dump(mode="json", exclude_none=True) + # RFC 3339 in UTC. Aware input is converted; naive input is taken as UTC, + # the pipeline-wide convention. + decided_at = decision.decided_at + if decided_at.tzinfo: + decided_at = decided_at.astimezone(timezone.utc).replace(tzinfo=None) + payload["decided_at"] = decided_at.isoformat() + "Z" + line = (json.dumps(payload, sort_keys=True) + "\n").encode("utf-8") + with open(self._path, "a+b") as fh: + # A crash mid-write leaves a line with no newline. Without this the + # next record would join it and be lost to every reader. + if fh.seek(0, os.SEEK_END) > 0: + fh.seek(-1, os.SEEK_END) + if fh.read(1) != b"\n": + line = b"\n" + line + fh.write(line) + fh.flush() + os.fsync(fh.fileno()) + # The file's contents are durable; now make its directory entry (and + # any directories makedirs created above it) durable too. + _fsync_dir_chain(os.path.dirname(self._path) or ".") + + +__all__ = [ + "HumanDecision", + "JsonlCorrectionsLog", + "LossEvent", + "RejectionDetail", + "UiMeta", +] diff --git a/application/utils/reviewer/queue_consumer.py b/application/utils/reviewer/queue_consumer.py new file mode 100644 index 000000000..ef1cbe2f7 --- /dev/null +++ b/application/utils/reviewer/queue_consumer.py @@ -0,0 +1,114 @@ +"""Module D's write-back to ``decision_queue``: stamp ``consumed_at``. + +The mirror of Module C's ``queue_consumer`` against Module B's queue, and it +follows the same three rules: exactly one column is written, nothing is ever +deleted (the queue doubles as the audit trail), and the update is filtered on +``consumed_at IS NULL`` so a replay cannot move a timestamp that is already set. + +One rule is D's own: **the filter also requires ``status='review_required'``.** +``linked`` rows belong to the graph writer, and a human-review service must be +structurally unable to retire one, even by bug or by being handed the wrong id. +Passing a linked row's id here is therefore a silent no-op, counted in the gap +between ids passed and rows stamped, exactly like an id that no longer exists. + +Call order is fixed by the package rule: record the decision in the corrections +log first, stamp here second. ``record_and_retire`` is that order as code, and is +what a review flow should call. A crash between the two leaves the row unconsumed +and the decision re-recordable, never the other way round, where the row is +retired and the decision lost. A re-record appends a second line for the same +row, and readers take the last one per ``(review_id, pipeline_run_id)``. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +from datetime import datetime, timezone +from typing import Any, Iterable, List, Sequence + +from application.utils.reviewer.corrections_log import ( + HumanDecision, + JsonlCorrectionsLog, +) +from application.utils.reviewer.review_source import ( + REVIEWABLE_STATUS, + PendingReview, +) + +# Same bind-parameter ceiling reasoning as Module C's write-back. +_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_reviewed(session: Any, row_ids: Iterable[str], *, at: datetime) -> int: + """Stamp ``consumed_at = at`` on the given review rows; return rows stamped. + + ``at`` is injected rather than read from the clock so a run is reproducible + and every row retired by one review session carries the same timestamp. + """ + unique: List[str] = list(dict.fromkeys(rid for rid in row_ids if rid)) + if not unique: + return 0 + + from application.database.db import DecisionQueueItem + + # `consumed_at` is a plain DateTime, stored naive-UTC across both dialects, + # same as everywhere else in the pipeline. Convert-then-strip keeps the + # stored instant correct under SQLite and Postgres alike. + 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(DecisionQueueItem) + .filter( + DecisionQueueItem.id.in_(list(chunk)), + DecisionQueueItem.consumed_at.is_(None), + DecisionQueueItem.status == REVIEWABLE_STATUS, + ) + .update({DecisionQueueItem.consumed_at: stamp}, synchronize_session=False) + ) + + if stamped != len(unique): + # Not an error: an id may be already consumed, gone, or — the case this + # module exists to make impossible to act on — a linked row. + logger.info( + "reviewer write-back: stamped %d of %d rows consumed " + "(the rest were already consumed, linked, or no longer exist)", + stamped, + len(unique), + ) + return stamped + + +def record_and_retire( + session: Any, + log: JsonlCorrectionsLog, + decision: HumanDecision, + review: PendingReview, + *, + at: datetime, +) -> int: + """Persist ``decision``, then stamp ``review``'s row; return rows stamped. + + Refuses a decision made about a different row before writing anything, since + that row would retire with no decision of its own on record. The caller owns + the session and commits, as with ``mark_reviewed``. + """ + if (decision.chunk_id, decision.pipeline_run_id) != ( + review.chunk_id, + review.pipeline_run_id, + ): + raise ValueError( + f"decision for {decision.chunk_id} in run {decision.pipeline_run_id} " + f"does not belong to row {review.row_id}" + ) + log.record(decision) + return mark_reviewed(session, [review.row_id], at=at) + + +__all__ = ["mark_reviewed", "record_and_retire"] diff --git a/application/utils/reviewer/review_source.py b/application/utils/reviewer/review_source.py new file mode 100644 index 000000000..daaf30a5d --- /dev/null +++ b/application/utils/reviewer/review_source.py @@ -0,0 +1,163 @@ +"""Where Module D reads pending reviews from. + +``DbReviewSource`` is D's counterpart to Module C's ``DbKnowledgeSource``: the +read side of a queue handoff, deliberately the same shape so anyone who has read +one can read the other. It selects ``status='review_required' AND consumed_at IS +NULL`` from ``decision_queue`` — never ``linked`` rows, which belong to the +graph writer — validates each row's ``envelope`` column back into the RFC +``ReviewItem`` it was written from, and yields both together. + +The envelope is validated through the *same* pinned model Module C emitted it +with (``application.utils.librarian.schemas.ReviewItem``), so C and D cannot +drift apart on what a review item is without one of them failing loudly. + +**One consumer at a time.** The read is not atomic with the ``consumed_at`` +stamp, exactly as on the B->C queue, and for the same reason the same rule +applies: run a single reviewer service per queue. The row-locking lesson from +#1025 is documented there; D inherits the single-consumer contract rather than +paying for locks it does not need yet. + +**Rows D cannot model are skipped, reported, and left unconsumed.** This is the +opposite of C's poison-row policy, on purpose. A ``knowledge_queue`` row C could +not model was B's input, safe to retire because C's refusal was the decision. A +``decision_queue`` row D cannot parse is *C's output and the audit record of a +decision*: retiring it would hide a C-side contract breach from the one queue +where a human would eventually see it. The ids surface in +``unreadable_row_ids`` so a run can alert on them instead. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +from abc import ABC, abstractmethod +from datetime import datetime +from typing import Any, Iterator, List, Optional + +from pydantic import BaseModel, ConfigDict, ValidationError + +from application.utils.librarian.schemas import ReviewItem + +#: The only status this package may touch, in reads and write-backs alike. +REVIEWABLE_STATUS = "review_required" +#: Rows loaded per query while walking the queue. +_BATCH_SIZE = 100 + + +class PendingReview(BaseModel): + """One decision_queue row awaiting a human, envelope already validated. + + The scalar columns are projections C made for filtering; the envelope is the + document of record. Both are carried so the review UI can list cheaply and + drill into the full audit without a second query. + """ + + model_config = ConfigDict(extra="forbid") + + row_id: str + chunk_id: str + artifact_id: str + pipeline_run_id: str + source_label: Optional[str] + reason_code: Optional[str] + confidence: Optional[float] + created_at: datetime + envelope: ReviewItem + + +class ReviewSource(ABC): + @abstractmethod + def items(self) -> Iterator[PendingReview]: + """Yield decision_queue rows awaiting human review.""" + raise NotImplementedError + + +class DbReviewSource(ReviewSource): + """Reads unconsumed ``review_required`` rows from ``decision_queue``. + + The caller owns the session, as everywhere else in this pipeline: this class + never opens, commits, or closes a transaction. Retiring a row is a separate, + explicit step (``queue_consumer.mark_reviewed``) that only happens after the + verdict is durably recorded. + + ``pipeline_run_id`` scopes to one orchestrator pass; left unset, the source + drains every pending review, which is what a standing review inbox wants. + ``limit`` caps the readable items in one page, so unreadable rows, which stay + unconsumed and sort first forever, cannot starve it. Rows are ordered by + ``created_at`` then ``id`` so a limited page is stable and 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 + #: Rows whose envelope failed validation. Never consumed by D — see the + #: module docstring for why this deliberately differs from C. + self.unreadable_row_ids: List[str] = [] + + def _rows(self) -> Iterator[Any]: + """Walk the matching rows in order, in batches. + + The ordered ids are read first and rows loaded by id. OFFSET would shift + if a caller stamps rows between batches, and comparing ``created_at`` + fails on SQLite, where the stored text and the bound value differ. The + filters are re-applied per batch, so a row stamped meanwhile is skipped. + """ + # Lazy import, same reason as the librarian: this package stays DB-free + # at import time so its tests and models are hermetic. + from application.database.db import DecisionQueueItem + + query = self._session.query(DecisionQueueItem).filter( # type: ignore[attr-defined] + DecisionQueueItem.consumed_at.is_(None), + DecisionQueueItem.status == REVIEWABLE_STATUS, + ) + if self._run_id: + query = query.filter(DecisionQueueItem.pipeline_run_id == self._run_id) + query = query.order_by(DecisionQueueItem.created_at, DecisionQueueItem.id) + + ids = [row_id for (row_id,) in query.with_entities(DecisionQueueItem.id)] + for start in range(0, len(ids), _BATCH_SIZE): + batch = ids[start : start + _BATCH_SIZE] + yield from query.filter(DecisionQueueItem.id.in_(batch)).all() + + def items(self) -> Iterator[PendingReview]: + self.unreadable_row_ids = [] + if self._limit is not None and self._limit <= 0: + return + yielded = 0 + for row in self._rows(): + try: + item = PendingReview( + row_id=row.id, + chunk_id=row.chunk_id, + artifact_id=row.artifact_id, + pipeline_run_id=row.pipeline_run_id, + source_label=row.source_label, + reason_code=row.reason_code, + confidence=row.confidence, + created_at=row.created_at, + envelope=ReviewItem.model_validate(row.envelope), + ) + except ValidationError as exc: + # A C-side contract breach worth seeing, but one bad row must + # not take the review inbox down. Ids are safe to log; the + # envelope's text is not. + row_id = getattr(row, "id", None) + logger.warning( + "Skipping unreadable decision_queue row id=%s: %s", + row_id if row_id is not None else "", + exc.errors(include_input=False), + ) + if row_id is not None: + self.unreadable_row_ids.append(row_id) + continue + yield item + yielded += 1 + if self._limit is not None and yielded >= self._limit: + return diff --git a/docs/gsoc_2026_module_c/module_d_contract.md b/docs/gsoc_2026_module_c/module_d_contract.md index 5c53549e9..385e1a0a5 100644 --- a/docs/gsoc_2026_module_c/module_d_contract.md +++ b/docs/gsoc_2026_module_c/module_d_contract.md @@ -104,6 +104,14 @@ Two things worth knowing before writing a validator against it: the candidates, the reranked shortlist and the threshold, so a reviewer can see *why* a chunk landed where it did rather than being asked to trust a number. +## What D does with a row + +D records the human's `HumanDecision` in `corrections.jsonl` before it stamps +`consumed_at`, so a retired row always has its decision on record (see +[module-d-hitl.md](../owasp-graph/apis/module-d-hitl.md)). A row whose envelope D +cannot validate is skipped, logged by id, and left unconsumed, because it is the +audit record of a C-side contract breach. It does not count against a page limit. + ## Reason codes Review rows carry one, in this precedence order: diff --git a/docs/gsoc_2026_module_d/click_speed_prototype.html b/docs/gsoc_2026_module_d/click_speed_prototype.html new file mode 100644 index 000000000..892b91126 --- /dev/null +++ b/docs/gsoc_2026_module_d/click_speed_prototype.html @@ -0,0 +1,138 @@ + + + + +Reviewer click-speed prototype + + + + +
+
+
+

y approve top suggestion   n reject   + c correct (type a CRE id, Enter)   u undo last
+ target: under 3s on every item

+
+ + + diff --git a/docs/owasp-graph/apis/module-d-hitl.md b/docs/owasp-graph/apis/module-d-hitl.md index 8942662fe..70987e75a 100644 --- a/docs/owasp-graph/apis/module-d-hitl.md +++ b/docs/owasp-graph/apis/module-d-hitl.md @@ -55,7 +55,7 @@ Fast maintainer UI (approve / reject / correct link). Append-only correction log | `GET` | `/internal/oie/v0/review/items` | Next `ReviewItem` for UI | | `POST` | `/internal/oie/v0/review/items/{review_id}/decisions` | Submit `HumanDecision` | -MVP: UI reads review JSONL from blob store; appends one line per decision to `corrections.jsonl`. +MVP: the reviewer reads `review_required` rows from `decision_queue` (see [the C to D contract](../../gsoc_2026_module_c/module_d_contract.md)) and appends one line per decision to `corrections.jsonl`. A replayed review appends a second line for the same row; readers take the last one per `(review_id, pipeline_run_id)`, because `review_id` alone repeats across runs. ## Downstream use of corrections