module_D week_1: The Reviewer - read path, verdict log, and the click-speed gate

.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

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),
+    )

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()

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()