From cf96fa1d25a03f22912dcbde69d4ae9252898965 Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Sun, 27 Sep 2026 22:50:17 +0530 Subject: [PATCH 1/4] module_D week_1: The Reviewer - read path, verdict log, and the click-speed gate Module D is the human-in-the-loop stage of the OIE pipeline. Module C writes every decision to decision_queue; the rows it could not decide alone arrive as status='review_required' with a full RFC ReviewItem envelope. Nothing has ever read those rows. This commit is the reader. DbReviewSource yields PendingReview rows (review_required AND consumed_at IS NULL, run-scoped, stable order), validating each envelope back through the SAME pinned RFC model Module C emitted it with, so C and D cannot drift silently. mark_reviewed stamps consumed_at: idempotent, chunked, naive-UTC, and filtered on status='review_required' so the human path is structurally unable to retire a linked row, even handed the right id. linked rows belong to the graph writer. JsonlCorrectionsLog records each human verdict append-only and fsync'd, per the founding RFC's "simple human oversight without db bloat". Verdict before retirement: the log write is the persistence the consumed_at stamp is gated on, the same rule C follows against B's queue one handoff upstream. One deliberate inversion of C's poison-row policy: a decision row D cannot parse is left UNCONSUMED and surfaced in unreadable_row_ids, because that row 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. docs/gsoc_2026_module_d/click_speed_prototype.html is the founding RFC's mandated pre-code experiment: ten realistic items, y/n keybinds, measured against the 3-second-per-item bar. The D1 UI is not allowed to be slower than this page. 14 hermetic tests; rows are seeded through the real DbEnvelopeSink so the tests exercise the actual C -> D handoff. One .gitignore negation for the package README, same pattern as the librarian's. --- .gitignore | 1 + application/tests/reviewer/__init__.py | 0 application/tests/reviewer/_envelopes.py | 85 +++++++++++ .../tests/reviewer/corrections_log_test.py | 99 ++++++++++++ .../tests/reviewer/queue_consumer_test.py | 82 ++++++++++ .../tests/reviewer/review_source_test.py | 126 +++++++++++++++ application/utils/reviewer/README.md | 71 +++++++++ application/utils/reviewer/__init__.py | 18 +++ application/utils/reviewer/corrections_log.py | 95 ++++++++++++ application/utils/reviewer/queue_consumer.py | 79 ++++++++++ application/utils/reviewer/review_source.py | 143 ++++++++++++++++++ .../click_speed_prototype.html | 97 ++++++++++++ 12 files changed, 896 insertions(+) create mode 100644 application/tests/reviewer/__init__.py create mode 100644 application/tests/reviewer/_envelopes.py create mode 100644 application/tests/reviewer/corrections_log_test.py create mode 100644 application/tests/reviewer/queue_consumer_test.py create mode 100644 application/tests/reviewer/review_source_test.py create mode 100644 application/utils/reviewer/README.md create mode 100644 application/utils/reviewer/__init__.py create mode 100644 application/utils/reviewer/corrections_log.py create mode 100644 application/utils/reviewer/queue_consumer.py create mode 100644 application/utils/reviewer/review_source.py create mode 100644 docs/gsoc_2026_module_d/click_speed_prototype.html 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..441f43cf6 --- /dev/null +++ b/application/tests/reviewer/corrections_log_test.py @@ -0,0 +1,99 @@ +"""Tests for the corrections log: the durable half of a verdict. + +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, and refuse verdicts outside the two the RFC defines. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +import json +import os +import tempfile +import unittest +from datetime import datetime, timezone + +from pydantic import ValidationError + +from application.utils.reviewer.corrections_log import ( + CORRECTIONS_SCHEMA_VERSION, + JsonlCorrectionsLog, + ReviewVerdict, +) + +AT = datetime(2026, 9, 17, 12, 0, 0, tzinfo=timezone.utc) + + +def _verdict(**overrides) -> ReviewVerdict: + values = dict( + decision_row_id="row-1", + review_id="review:chk:1", + chunk_id="chk:1", + pipeline_run_id="run-1", + reason_code="BELOW_THRESHOLD", + source_label="KNOWLEDGE", + verdict="approved", + approved_cre_id="616-305", + reviewer="prateek", + reviewed_at=AT, + ) + values.update(overrides) + return ReviewVerdict(**values) + + +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_appends_one_json_line_per_verdict(self) -> None: + log = JsonlCorrectionsLog(self.path) + log.record(_verdict()) + log.record( + _verdict(decision_row_id="row-2", verdict="rejected", approved_cre_id=None) + ) + + first, second = self._lines() + self.assertEqual(first["decision_row_id"], "row-1") + self.assertEqual(first["verdict"], "approved") + self.assertEqual(first["approved_cre_id"], "616-305") + self.assertEqual(first["schema_version"], CORRECTIONS_SCHEMA_VERSION) + self.assertEqual(second["verdict"], "rejected") + self.assertIsNone(second["approved_cre_id"]) + + def test_survives_reopening_the_log(self) -> None: + """Two service restarts, one file: earlier verdicts must remain.""" + JsonlCorrectionsLog(self.path).record(_verdict()) + JsonlCorrectionsLog(self.path).record(_verdict(decision_row_id="row-2")) + + self.assertEqual( + [line["decision_row_id"] for line in self._lines()], ["row-1", "row-2"] + ) + + def test_reviewed_at_is_recorded_in_utc(self) -> None: + from datetime import timedelta, timezone as tz + + ist = tz(timedelta(hours=5, minutes=30)) + JsonlCorrectionsLog(self.path).record(_verdict(reviewed_at=AT.astimezone(ist))) + + (line,) = self._lines() + # Canonical form: naive UTC ISO, byte-identical to what decision_queue + # stores, no offset suffix to strip before joining. + self.assertEqual(line["reviewed_at"], "2026-09-17T12:00:00") + + def test_only_the_two_rfc_verdicts_are_representable(self) -> None: + with self.assertRaises(ValidationError): + _verdict(verdict="maybe") + + +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..c868f31cc --- /dev/null +++ b/application/tests/reviewer/queue_consumer_test.py @@ -0,0 +1,82 @@ +"""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 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.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) + + +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..a95919144 --- /dev/null +++ b/application/tests/reviewer/review_source_test.py @@ -0,0 +1,126 @@ +"""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 application import create_app, sqla +from application.database.db import DecisionQueueItem +from application.utils.librarian.envelope_sink import DbEnvelopeSink +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) + + +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..bfee22ae1 --- /dev/null +++ b/application/utils/reviewer/README.md @@ -0,0 +1,71 @@ +# 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 verdict is persisted before the row is retired.** `JsonlCorrectionsLog` +records the human's decision (fsync'd) first; `mark_reviewed` stamps +`consumed_at` second. A crash between the two leaves the row unconsumed and the +verdict re-recordable, never the reverse. 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 | +| Verdict log | `corrections_log.py` | append-only JSONL, one `ReviewVerdict` per line, 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 realistically shaped +items with `y`/`n`, and it measures whether the flow stays under the RFC's +3-second-per-item bar. The D1 UI is not allowed to be slower than this page. + +## Not built yet + +- **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..70b3deb29 --- /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 verdict is persisted before the row is retired.** Stamping + ``consumed_at`` tells the pipeline this chunk is finished; doing it before the + human's verdict 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..3950d2b53 --- /dev/null +++ b/application/utils/reviewer/corrections_log.py @@ -0,0 +1,95 @@ +"""The corrections log: every human verdict, 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" — verdicts append to a +JSONL log rather than growing new tables. The RFC's bonus tier, loss +warehousing, falls out for free: each record carries the ids needed to join +back to the full envelope, and ``decision_queue`` rows are never deleted, so +(input, prediction, human label) is always reconstructable without duplicating +the envelope into every line. + +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 +from datetime import datetime, timezone +from typing import Literal, Optional + +from pydantic import BaseModel, ConfigDict + +#: Bumped when a record gains/loses fields; readers filter on it. +CORRECTIONS_SCHEMA_VERSION = "0.1.0" + + +class ReviewVerdict(BaseModel): + """One human decision about one ``review_required`` row.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: str = CORRECTIONS_SCHEMA_VERSION + #: decision_queue primary key — the join back to the full envelope. + decision_row_id: str + review_id: Optional[str] + chunk_id: str + pipeline_run_id: str + #: why C routed this to a human, carried so the log is analysable alone. + reason_code: Optional[str] + #: B's label on the source chunk, for the same reason. + source_label: Optional[str] + verdict: Literal["approved", "rejected"] + #: On approval: the CRE id the human confirmed (from suggested_links or + #: typed). On rejection: None. + approved_cre_id: Optional[str] = None + reviewer: str + reviewed_at: datetime + + +class JsonlCorrectionsLog: + """Appends verdicts 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, verdict: ReviewVerdict) -> None: + """Append one verdict 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 — 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) + payload = verdict.model_dump(mode="json") + # One canonical timestamp format: naive UTC ISO, exactly what + # decision_queue itself stores, so joining the log to the queue never + # fights offsets. Aware input is converted; naive input is taken as UTC, + # the pipeline-wide convention. + reviewed_at = verdict.reviewed_at + if reviewed_at.tzinfo: + reviewed_at = reviewed_at.astimezone(timezone.utc).replace(tzinfo=None) + payload["reviewed_at"] = reviewed_at.isoformat() + line = json.dumps(payload, sort_keys=True) + with open(self._path, "a", encoding="utf-8") as fh: + fh.write(line + "\n") + fh.flush() + os.fsync(fh.fileno()) + + +__all__ = ["CORRECTIONS_SCHEMA_VERSION", "JsonlCorrectionsLog", "ReviewVerdict"] diff --git a/application/utils/reviewer/queue_consumer.py b/application/utils/reviewer/queue_consumer.py new file mode 100644 index 000000000..04665f127 --- /dev/null +++ b/application/utils/reviewer/queue_consumer.py @@ -0,0 +1,79 @@ +"""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 verdict in the corrections +log first, stamp here second. A crash between the two leaves the row unconsumed +and the verdict re-recordable — never the other way round, where the row is +retired and the verdict lost. +""" + +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.review_source import REVIEWABLE_STATUS + +# 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 + + +__all__ = ["mark_reviewed"] diff --git a/application/utils/reviewer/review_source.py b/application/utils/reviewer/review_source.py new file mode 100644 index 000000000..4591f495b --- /dev/null +++ b/application/utils/reviewer/review_source.py @@ -0,0 +1,143 @@ +"""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 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" + + +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 one page. 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 _query(self) -> object: + # 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) + if self._limit is not None: + query = query.limit(self._limit) + return query + + def items(self) -> Iterator[PendingReview]: + for row in self._query(): # type: ignore[attr-defined] + try: + yield 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 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..9fbcb4c6b --- /dev/null +++ b/docs/gsoc_2026_module_d/click_speed_prototype.html @@ -0,0 +1,97 @@ + + + + +Reviewer click-speed prototype + + + + +
+
+
+

y approve   n reject   + u undo last   (target: under 3s per item)

+
+ + + From f48ac54f9e3f1a8e82735d5c2f0bf9823bb0363b Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Sun, 27 Sep 2026 23:16:01 +0530 Subject: [PATCH 2/4] module_D week_1: make the verdict's directory entry durable, guard key auto-repeat CodeRabbit's review caught both. The corrections log fsynced the file but not its directory entry: a crash after creating the file could lose the file itself unless the parent (and any directory makedirs created above it) is synced. The chain is now synced on every record; which ancestors are new cannot be detected without racing a concurrent record, and records are human-paced, so syncing the whole chain is the simple correct choice. Platforms that refuse to fsync a directory are skipped, leaving durability no worse than before the attempt. The click-speed prototype now ignores key auto-repeat: a held y or n must not verdict a run of items, or the measured average stops describing the workflow the RFC's 3-second bar is about. --- application/utils/reviewer/corrections_log.py | 34 +++++++++++++++++++ .../click_speed_prototype.html | 1 + 2 files changed, 35 insertions(+) diff --git a/application/utils/reviewer/corrections_log.py b/application/utils/reviewer/corrections_log.py index 3950d2b53..bba29ae66 100644 --- a/application/utils/reviewer/corrections_log.py +++ b/application/utils/reviewer/corrections_log.py @@ -27,6 +27,37 @@ from datetime import datetime, timezone from typing import 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 #: Bumped when a record gains/loses fields; readers filter on it. @@ -90,6 +121,9 @@ def record(self, verdict: ReviewVerdict) -> None: fh.write(line + "\n") 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__ = ["CORRECTIONS_SCHEMA_VERSION", "JsonlCorrectionsLog", "ReviewVerdict"] diff --git a/docs/gsoc_2026_module_d/click_speed_prototype.html b/docs/gsoc_2026_module_d/click_speed_prototype.html index 9fbcb4c6b..d095ced0b 100644 --- a/docs/gsoc_2026_module_d/click_speed_prototype.html +++ b/docs/gsoc_2026_module_d/click_speed_prototype.html @@ -87,6 +87,7 @@ setInterval(() => { if (i < ITEMS.length) timer.textContent = ((performance.now()-t0)/1000).toFixed(1) + "s"; }, 100); document.addEventListener("keydown", e => { + if (e.repeat) return; // held key must not auto-verdict a run of items if (e.key === "y") decide("y"); else if (e.key === "n") decide("n"); else if (e.key === "u") undo(); From cec9344673113c983107a656d3faf12a62c580b4 Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Thu, 5 Nov 2026 16:51:23 +0530 Subject: [PATCH 3/4] reviewer: fix what review found The corrections log now writes HumanDecision as pinned in human-decision.json, with all three actions, loss_event and ui timing, instead of an invented shape. A test validates real lines against the schema, and a decision that contradicts itself is refused. A torn last line no longer swallows the next record. record_and_retire is the fixed order as code. A limited page now counts readable items, so unreadable rows cannot starve the inbox, and unreadable_row_ids resets per read. The click-speed page keeps time across undo, fails on any slow item or too many wrong answers, and adds correct, multi-candidate and empty-shortlist items with real reason codes. Contract docs and README updated, including that the log needs durable storage before D1 runs on a dyno. --- .../tests/reviewer/corrections_log_test.py | 169 +++++++++++++----- .../tests/reviewer/queue_consumer_test.py | 47 ++++- .../tests/reviewer/review_source_test.py | 48 +++++ application/utils/reviewer/README.md | 27 +-- application/utils/reviewer/__init__.py | 4 +- application/utils/reviewer/corrections_log.py | 138 +++++++++----- application/utils/reviewer/queue_consumer.py | 32 +++- application/utils/reviewer/review_source.py | 38 +++- docs/gsoc_2026_module_c/module_d_contract.md | 8 + .../click_speed_prototype.html | 97 ++++++---- docs/owasp-graph/apis/module-d-hitl.md | 2 +- 11 files changed, 467 insertions(+), 143 deletions(-) diff --git a/application/tests/reviewer/corrections_log_test.py b/application/tests/reviewer/corrections_log_test.py index 441f43cf6..d2187c3b2 100644 --- a/application/tests/reviewer/corrections_log_test.py +++ b/application/tests/reviewer/corrections_log_test.py @@ -1,8 +1,9 @@ -"""Tests for the corrections log: the durable half of a verdict. +"""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, and refuse verdicts outside the two the RFC defines. +whole, survive reopening, match the pinned ``HumanDecision`` schema, and refuse +a decision that contradicts itself. """ from cre_logging import get_logger @@ -13,34 +14,92 @@ import os import tempfile import unittest -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone +from unittest import mock +from jsonschema import Draft202012Validator 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 ( - CORRECTIONS_SCHEMA_VERSION, + HumanDecision, JsonlCorrectionsLog, - ReviewVerdict, + 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 _verdict(**overrides) -> ReviewVerdict: +def _decision(**overrides) -> HumanDecision: values = dict( - decision_row_id="row-1", review_id="review:chk:1", chunk_id="chk:1", pipeline_run_id="run-1", - reason_code="BELOW_THRESHOLD", - source_label="KNOWLEDGE", - verdict="approved", - approved_cre_id="616-305", - reviewer="prateek", - reviewed_at=AT, + decided_at=AT, + decided_by="prateek", + action="approve", + final_links=[LINK], ) values.update(overrides) - return ReviewVerdict(**values) + 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) + + +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=""), + ): + with self.assertRaises(ValidationError, msg=bad): + _decision(**bad) class JsonlCorrectionsLogTest(unittest.TestCase): @@ -55,44 +114,74 @@ def _lines(self): with open(self.path, encoding="utf-8") as fh: return [json.loads(line) for line in fh] - def test_appends_one_json_line_per_verdict(self) -> None: + 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(_verdict()) + log.record(_decision()) + log.record(_rejected()) log.record( - _verdict(decision_row_id="row-2", verdict="rejected", approved_cre_id=None) + _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.assertEqual(first["decision_row_id"], "row-1") - self.assertEqual(first["verdict"], "approved") - self.assertEqual(first["approved_cre_id"], "616-305") - self.assertEqual(first["schema_version"], CORRECTIONS_SCHEMA_VERSION) - self.assertEqual(second["verdict"], "rejected") - self.assertIsNone(second["approved_cre_id"]) + self.assertNotEqual(first["decision_id"], second["decision_id"]) def test_survives_reopening_the_log(self) -> None: - """Two service restarts, one file: earlier verdicts must remain.""" - JsonlCorrectionsLog(self.path).record(_verdict()) - JsonlCorrectionsLog(self.path).record(_verdict(decision_row_id="row-2")) - - self.assertEqual( - [line["decision_row_id"] for line in self._lines()], ["row-1", "row-2"] - ) + """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")) - def test_reviewed_at_is_recorded_in_utc(self) -> None: - from datetime import timedelta, timezone as tz + self.assertEqual([line["review_id"] for line in self._lines()], ["r1", "r2"]) - ist = tz(timedelta(hours=5, minutes=30)) - JsonlCorrectionsLog(self.path).record(_verdict(reviewed_at=AT.astimezone(ist))) + 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() - # Canonical form: naive UTC ISO, byte-identical to what decision_queue - # stores, no offset suffix to strip before joining. - self.assertEqual(line["reviewed_at"], "2026-09-17T12:00:00") + self.assertEqual(line["decided_at"], "2026-09-17T12:00:00Z") - def test_only_the_two_rfc_verdicts_are_representable(self) -> None: - with self.assertRaises(ValidationError): - _verdict(verdict="maybe") + 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__": diff --git a/application/tests/reviewer/queue_consumer_test.py b/application/tests/reviewer/queue_consumer_test.py index c868f31cc..194a6e6cb 100644 --- a/application/tests/reviewer/queue_consumer_test.py +++ b/application/tests/reviewer/queue_consumer_test.py @@ -11,11 +11,13 @@ 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.queue_consumer import mark_reviewed +from application.utils.reviewer.corrections_log import HumanDecision, RejectionDetail +from application.utils.reviewer.queue_consumer import mark_reviewed, record_and_retire from application.tests.reviewer._envelopes import RUN, linked, review AT = datetime(2026, 9, 17, 12, 0, 0, tzinfo=timezone.utc) @@ -78,5 +80,48 @@ def test_empty_and_falsy_ids_are_a_no_op(self) -> None: 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.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.row.id, 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.row.id, at=AT) + 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 index a95919144..2ae7e91e2 100644 --- a/application/tests/reviewer/review_source_test.py +++ b/application/tests/reviewer/review_source_test.py @@ -12,10 +12,13 @@ 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 @@ -121,6 +124,51 @@ def test_unreadable_envelope_is_skipped_reported_and_never_consumed(self) -> Non 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 index bfee22ae1..c0da1431a 100644 --- a/application/utils/reviewer/README.md +++ b/application/utils/reviewer/README.md @@ -18,19 +18,20 @@ A (harvester) -> B (filter) -> C (librarian) -> decision_queue -> D (reviewer) `status='review_required'`, so a human queue is structurally unable to retire an auto-link, even handed the right id. -**The verdict is persisted before the row is retired.** `JsonlCorrectionsLog` -records the human's decision (fsync'd) first; `mark_reviewed` stamps -`consumed_at` second. A crash between the two leaves the row unconsumed and the -verdict re-recordable, never the reverse. This is the same rule Module C follows -against Module B's queue, one handoff downstream. +**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 `review_id`; readers take the last one. 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 | -| Verdict log | `corrections_log.py` | append-only JSONL, one `ReviewVerdict` per line, the RFC's "no db bloat" decision | +| 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 @@ -41,12 +42,18 @@ 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 realistically shaped -items with `y`/`n`, and it measures whether the flow stays under the RFC's -3-second-per-item bar. The D1 UI is not allowed to be slower than this page. +pre-code experiment: open it in a browser, review ten fake items with `y`, `n` +and `c`, 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 diff --git a/application/utils/reviewer/__init__.py b/application/utils/reviewer/__init__.py index 70b3deb29..14cc98d40 100644 --- a/application/utils/reviewer/__init__.py +++ b/application/utils/reviewer/__init__.py @@ -11,8 +11,8 @@ - **``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 verdict is persisted before the row is retired.** Stamping +- **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 verdict is durably recorded would destroy the one thing the review + 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 index bba29ae66..2fa006392 100644 --- a/application/utils/reviewer/corrections_log.py +++ b/application/utils/reviewer/corrections_log.py @@ -1,12 +1,16 @@ -"""The corrections log: every human verdict, append-only, one JSON per line. +"""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" — verdicts append to a -JSONL log rather than growing new tables. The RFC's bonus tier, loss -warehousing, falls out for free: each record carries the ids needed to join -back to the full envelope, and ``decision_queue`` rows are never deleted, so -(input, prediction, human label) is always reconstructable without duplicating -the envelope into every line. +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. ``review_id`` joins a line back to its envelope, and ``decision_queue`` +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 ``review_id``. The log +stays append-only, so readers take the **last** line per ``review_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 @@ -24,8 +28,9 @@ import json import os +import uuid from datetime import datetime, timezone -from typing import Literal, Optional +from typing import Any, Dict, List, Literal, Optional def _fsync_dir_chain(path: str) -> None: @@ -58,37 +63,71 @@ def _fsync_dir_chain(path: str) -> None: current = parent -from pydantic import BaseModel, ConfigDict +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") -#: Bumped when a record gains/loses fields; readers filter on it. -CORRECTIONS_SCHEMA_VERSION = "0.1.0" + review_duration_ms: Optional[int] = Field(default=None, ge=0) + keybind_used: Optional[str] = None -class ReviewVerdict(BaseModel): - """One human decision about one ``review_required`` row.""" +class HumanDecision(BaseModel): + """One human decision about one review item (``human-decision.json``).""" model_config = ConfigDict(extra="forbid") - schema_version: str = CORRECTIONS_SCHEMA_VERSION - #: decision_queue primary key — the join back to the full envelope. - decision_row_id: str - review_id: Optional[str] + schema_version: str = SCHEMA_VERSION + decision_id: str = Field(default_factory=lambda: str(uuid.uuid4())) + review_id: str chunk_id: str pipeline_run_id: str - #: why C routed this to a human, carried so the log is analysable alone. - reason_code: Optional[str] - #: B's label on the source chunk, for the same reason. - source_label: Optional[str] - verdict: Literal["approved", "rejected"] - #: On approval: the CRE id the human confirmed (from suggested_links or - #: typed). On rejection: None. - approved_cre_id: Optional[str] = None - reviewer: str - reviewed_at: datetime + 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 verdicts to a JSONL file, creating parent directories as needed.""" + """Appends decisions to a JSONL file, creating parent directories as needed.""" def __init__(self, path: str) -> None: self._path = path @@ -97,28 +136,33 @@ def __init__(self, path: str) -> None: def path(self) -> str: return self._path - def record(self, verdict: ReviewVerdict) -> None: - """Append one verdict durably (flush + fsync before returning). + 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 — same rule as C's - "consumption is gated on persistence", one queue downstream. + 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) - payload = verdict.model_dump(mode="json") - # One canonical timestamp format: naive UTC ISO, exactly what - # decision_queue itself stores, so joining the log to the queue never - # fights offsets. Aware input is converted; naive input is taken as UTC, + # 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. - reviewed_at = verdict.reviewed_at - if reviewed_at.tzinfo: - reviewed_at = reviewed_at.astimezone(timezone.utc).replace(tzinfo=None) - payload["reviewed_at"] = reviewed_at.isoformat() - line = json.dumps(payload, sort_keys=True) - with open(self._path, "a", encoding="utf-8") as fh: - fh.write(line + "\n") + 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 @@ -126,4 +170,10 @@ def record(self, verdict: ReviewVerdict) -> None: _fsync_dir_chain(os.path.dirname(self._path) or ".") -__all__ = ["CORRECTIONS_SCHEMA_VERSION", "JsonlCorrectionsLog", "ReviewVerdict"] +__all__ = [ + "HumanDecision", + "JsonlCorrectionsLog", + "LossEvent", + "RejectionDetail", + "UiMeta", +] diff --git a/application/utils/reviewer/queue_consumer.py b/application/utils/reviewer/queue_consumer.py index 04665f127..573f04553 100644 --- a/application/utils/reviewer/queue_consumer.py +++ b/application/utils/reviewer/queue_consumer.py @@ -11,10 +11,12 @@ 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 verdict in the corrections -log first, stamp here second. A crash between the two leaves the row unconsumed -and the verdict re-recordable — never the other way round, where the row is -retired and the verdict lost. +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 +``review_id``, and readers take the last one. """ from cre_logging import get_logger @@ -24,6 +26,10 @@ 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 # Same bind-parameter ceiling reasoning as Module C's write-back. @@ -76,4 +82,20 @@ def mark_reviewed(session: Any, row_ids: Iterable[str], *, at: datetime) -> int: return stamped -__all__ = ["mark_reviewed"] +def record_and_retire( + session: Any, + log: JsonlCorrectionsLog, + decision: HumanDecision, + row_id: str, + *, + at: datetime, +) -> int: + """Persist ``decision``, then stamp its row; return rows stamped. + + The caller owns the session and commits, as with ``mark_reviewed``. + """ + log.record(decision) + return mark_reviewed(session, [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 index 4591f495b..daaf30a5d 100644 --- a/application/utils/reviewer/review_source.py +++ b/application/utils/reviewer/review_source.py @@ -32,7 +32,7 @@ from abc import ABC, abstractmethod from datetime import datetime -from typing import Iterator, List, Optional +from typing import Any, Iterator, List, Optional from pydantic import BaseModel, ConfigDict, ValidationError @@ -40,6 +40,8 @@ #: 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): @@ -80,8 +82,9 @@ class DbReviewSource(ReviewSource): ``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 one page. Rows are ordered by ``created_at`` then ``id`` so a - limited page is stable and reproducible. + ``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__( @@ -98,7 +101,14 @@ def __init__( #: module docstring for why this deliberately differs from C. self.unreadable_row_ids: List[str] = [] - def _query(self) -> object: + 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 @@ -110,14 +120,20 @@ def _query(self) -> object: if self._run_id: query = query.filter(DecisionQueueItem.pipeline_run_id == self._run_id) query = query.order_by(DecisionQueueItem.created_at, DecisionQueueItem.id) - if self._limit is not None: - query = query.limit(self._limit) - return query + + 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]: - for row in self._query(): # type: ignore[attr-defined] + self.unreadable_row_ids = [] + if self._limit is not None and self._limit <= 0: + return + yielded = 0 + for row in self._rows(): try: - yield PendingReview( + item = PendingReview( row_id=row.id, chunk_id=row.chunk_id, artifact_id=row.artifact_id, @@ -141,3 +157,7 @@ def items(self) -> Iterator[PendingReview]: 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 index d095ced0b..643253e77 100644 --- a/docs/gsoc_2026_module_d/click_speed_prototype.html +++ b/docs/gsoc_2026_module_d/click_speed_prototype.html @@ -28,68 +28,103 @@
-

y approve   n reject   - u undo last   (target: under 3s per item)

+

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 51c999169..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: 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 `review_id`; readers take the last one. +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