From cf96fa1d25a03f22912dcbde69d4ae9252898965 Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Sun, 27 Sep 2026 22:50:17 +0530 Subject: [PATCH 1/2] 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/2] 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();