Add harvest_input and knowledge_queue models with pipeline support

application/cmd/cre_main.py

@@ -893,6 +893,22 @@ def run(args: argparse.Namespace) -> None:  # pragma: no cover
         logger.info("Exported %s rows to %s", rows, csv_out)
         return
 
+    if getattr(args, "run_noise_filter", False):
+        from application import sqla
+        from application.utils.noise_filter.pipeline import run_noise_filter
+
+        run_id = getattr(args, "run_id", "").strip()
+        if not run_id:
+            raise ValueError("--run_noise_filter requires --run_id <pipeline_run_id>")
+        db_connect(args.cache_file)
+        summary = run_noise_filter(
+            sqla.session,
+            run_id,
+            dry_run=getattr(args, "noise_filter_dry_run", False),
+        )
+        print(summary.to_json())
+        return
+
     if args.add and getattr(args, "from_ai_exchange_csv", None):
         add_from_ai_exchange_csv(
             csv_path=args.from_ai_exchange_csv,

application/database/db.py

@@ -257,6 +257,77 @@ class StagedChangeSet(BaseModel):  # type: ignore
     created_at = sqla.Column(sqla.DateTime, nullable=False)
 
 
+# --- Module B: Noise/Relevance Filter (harvest in -> knowledge queue out) ---
+
+
+class HarvestInput(BaseModel):  # type: ignore
+    """Module A's harvested chunks, staged for Module B to classify.
+
+    Module A writes one row per chunk (the full Module A v0.3 ChangeRecord in
+    `payload`). The orchestrator triggers Module B, which reads the rows for a
+    given `pipeline_run_id`, classifies them, and marks each `processed`.
+    Contract: docs/gsoc_2026_module_b/orchestrator_integration_design.md.
+    """
+
+    __tablename__ = "harvest_input"
+    id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
+    pipeline_run_id = sqla.Column(sqla.String, nullable=False, index=True)
+    status = sqla.Column(
+        sqla.String, nullable=False, default="pending"
+    )  # pending | processed
+    # A's ChangeRecord (contract v0.3). JSONB on Postgres, JSON on SQLite
+    # (dev/CI/tests); Module B parses it with schemas.ChangeRecord directly.
+    payload = sqla.Column(sqla.JSON().with_variant(JSONB, "postgresql"), nullable=False)
+    created_at = sqla.Column(
+        sqla.DateTime, nullable=False, server_default=sqla.func.now()
+    )
+    __table_args__ = (
+        sqla.Index("ix_harvest_input_run_status", "pipeline_run_id", "status"),
+    )
+
+
+class KnowledgeQueueItem(BaseModel):  # type: ignore
+    """Module B's output queue: security-knowledge chunks for Module C.
+
+    Module B inserts KNOWLEDGE and UNCERTAIN verdicts (NOISE is dropped),
+    deduped on `content_hash`. Module C reads unconsumed rows and sets
+    `consumed_at`. Contract: docs/gsoc_2026_module_b/module_c_contract.md (v0.2).
+    """
+
+    __tablename__ = "knowledge_queue"
+    id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
+    content_hash = sqla.Column(sqla.String, nullable=False)  # B-computed dedup key
+    # provenance / traceability (Module A v0.3 record)
+    chunk_id = sqla.Column(sqla.String, nullable=False)
+    artifact_id = sqla.Column(sqla.String, nullable=False)
+    pipeline_run_id = sqla.Column(sqla.String, nullable=False)
+    schema_version = sqla.Column(sqla.String, nullable=False)
+    source_type = sqla.Column(sqla.String, nullable=False)  # github | rss
+    source_repo = sqla.Column(sqla.String, nullable=True)
+    source_commit_sha = sqla.Column(sqla.String, nullable=True)
+    source_committed_at = sqla.Column(sqla.String, nullable=True)  # ISO-8601, unparsed
+    feed_url = sqla.Column(sqla.String, nullable=True)
+    post_guid = sqla.Column(sqla.String, nullable=True)
+    locator_kind = sqla.Column(sqla.String, nullable=False)
+    locator_path = sqla.Column(sqla.String, nullable=False)
+    span_index = sqla.Column(sqla.Integer, nullable=False)
+    span_total = sqla.Column(sqla.Integer, nullable=False)
+    span_heading_path = sqla.Column(sqla.Text, nullable=True)  # JSON-encoded list[str]
+    # payload + B's verdict
+    text = sqla.Column(sqla.Text, nullable=False)
+    llm_label = sqla.Column(sqla.String, nullable=False)  # KNOWLEDGE | UNCERTAIN
+    confidence = sqla.Column(sqla.Float, nullable=False)
+    llm_reasoning = sqla.Column(sqla.Text, nullable=True)
+    created_at = sqla.Column(
+        sqla.DateTime, nullable=False, server_default=sqla.func.now()
+    )
+    consumed_at = sqla.Column(sqla.DateTime, nullable=True)
+    __table_args__ = (
+        sqla.Index("ix_knowledge_queue_unconsumed", "consumed_at"),
+        sqla.UniqueConstraint("content_hash", name="uq_content_hash"),
+    )
+
+
 def create_import_run(source: str, version: Optional[str] = None) -> ImportRun:
     """Create and persist an import run record. Returns the new ImportRun."""
     from datetime import datetime, timezone

application/tests/noise_filter/pipeline_test.py

@@ -0,0 +1,131 @@
+"""Tests for application.utils.noise_filter.pipeline.
+
+In-memory SQLite (create_app(mode="test") + create_all). The LLM is a fake
+classifier injected into run_noise_filter, so no real API calls.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from application import create_app, sqla
+from application.database.db import HarvestInput, KnowledgeQueueItem
+from application.utils.noise_filter.pipeline import run_noise_filter
+from application.utils.noise_filter.schemas import ClassifyResult
+
+
+def _payload(path="document/auth.md", text="security testing content"):
+    return {
+        "schema_version": "0.2.0",
+        "chunk_id": f"chk:{path}",
+        "artifact_id": f"art:{path}",
+        "pipeline_run_id": "run1",
+        "text": text,
+        "span": {"index": 0, "total": 1, "heading_path": []},
+        "source": {
+            "type": "github",
+            "repo": "OWASP/test",
+            "commit_sha": "abc123",
+            "committed_at": "2026-07-17T00:00:00Z",
+        },
+        "locator": {"kind": "repo_path", "id": path, "path": path},
+    }
+
+
+class _FakeClassifier:
+    """Returns preset verdicts; asserts they align with the survivor count."""
+
+    def __init__(self, verdicts):
+        self.verdicts = verdicts
+
+    def classify_batch(self, records):
+        assert len(records) == len(self.verdicts), (len(records), len(self.verdicts))
+        return list(self.verdicts)
+
+
+def _v(label, conf=0.9):
+    return ClassifyResult(label=label, confidence=conf, reasoning="r")
+
+
+class PipelineTests(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 _add(self, payload, status="pending", run_id="run1"):
+        sqla.session.add(
+            HarvestInput(pipeline_run_id=run_id, status=status, payload=payload)
+        )
+        sqla.session.commit()
+
+    def test_happy_path(self) -> None:
+        self._add(_payload("document/auth.md"))  # survives -> KNOWLEDGE
+        self._add(_payload("frontend/app.css"))  # regex-dropped (NOISE)
+        self._add(_payload("document/xss.md"))  # survives -> NOISE
+        clf = _FakeClassifier([_v("KNOWLEDGE"), _v("NOISE")])
+
+        s = run_noise_filter(sqla.session, "run1", classifier=clf)
+
+        self.assertEqual(s.read, 3)
+        self.assertEqual(s.dropped_noise, 2)  # 1 regex + 1 llm
+        self.assertEqual(s.kept_knowledge, 1)
+        self.assertEqual(s.inserted, 1)
+        self.assertEqual(KnowledgeQueueItem.query.count(), 1)
+        # all input rows marked processed
+        self.assertEqual(HarvestInput.query.filter_by(status="pending").count(), 0)
+
+    def test_parse_error_marks_row_error(self) -> None:
+        bad = _payload()
+        del bad["text"]  # violates ChangeRecord (text required)
+        self._add(bad)
+        clf = _FakeClassifier([])  # no survivors reach the LLM
+
+        s = run_noise_filter(sqla.session, "run1", classifier=clf)
+
+        self.assertEqual((s.read, s.parse_errors), (1, 1))
+        self.assertEqual(HarvestInput.query.filter_by(status="error").count(), 1)
+        self.assertEqual(KnowledgeQueueItem.query.count(), 0)
+
+    def test_dry_run_does_not_persist(self) -> None:
+        self._add(_payload("document/auth.md"))
+        clf = _FakeClassifier([_v("KNOWLEDGE")])
+
+        s = run_noise_filter(sqla.session, "run1", classifier=clf, dry_run=True)
+
+        self.assertEqual(s.kept_knowledge, 1)
+        self.assertEqual(s.inserted, 0)
+        self.assertEqual(KnowledgeQueueItem.query.count(), 0)
+        # row stays pending (dry run mutates nothing)
+        self.assertEqual(HarvestInput.query.filter_by(status="pending").count(), 1)
+
+    def test_only_pending_rows_read(self) -> None:
+        self._add(_payload("document/auth.md"), status="processed")
+        self._add(_payload("document/xss.md"), status="pending")
+        clf = _FakeClassifier([_v("KNOWLEDGE")])
+
+        s = run_noise_filter(sqla.session, "run1", classifier=clf)
+
+        self.assertEqual(s.read, 1)
+        self.assertEqual(s.inserted, 1)
+
+    def test_run_scoped_by_pipeline_run_id(self) -> None:
+        self._add(_payload("document/auth.md"), run_id="run1")
+        self._add(_payload("document/xss.md"), run_id="run2")
+        clf = _FakeClassifier([_v("KNOWLEDGE")])
+
+        s = run_noise_filter(sqla.session, "run1", classifier=clf)
+
+        self.assertEqual(s.read, 1)
+        self.assertEqual(HarvestInput.query.filter_by(status="pending").count(), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

application/tests/noise_filter/queue_writer_test.py

@@ -0,0 +1,108 @@
+"""Tests for application.utils.noise_filter.queue_writer.
+
+Uses an in-memory SQLite DB (create_app(mode="test") + create_all), matching
+the project's db_test.py pattern. No migration needed.
+"""
+
+from __future__ import annotations
+
+import json
+import unittest
+
+from application import create_app, sqla
+from application.database.db import KnowledgeQueueItem
+from application.utils.noise_filter.queue_writer import write_verdicts
+from application.utils.noise_filter.schemas import ChangeRecord, ClassifyResult
+
+
+def _record(chunk_id="chk", source=None, heading_path=None) -> ChangeRecord:
+    return ChangeRecord.model_validate(
+        {
+            "schema_version": "0.2.0",
+            "chunk_id": chunk_id,
+            "artifact_id": "art:test",
+            "pipeline_run_id": "run1",
+            "text": "some security text",
+            "span": {"index": 0, "total": 1, "heading_path": heading_path or []},
+            "source": source
+            or {
+                "type": "github",
+                "repo": "OWASP/test",
+                "commit_sha": "abc123",
+                "committed_at": "2026-07-17T00:00:00Z",
+            },
+            "locator": {"kind": "repo_path", "id": "p.md", "path": "p.md"},
+        }
+    )
+
+
+def _verdict(label="KNOWLEDGE", conf=0.9) -> ClassifyResult:
+    return ClassifyResult(label=label, confidence=conf, reasoning="because")
+
+
+class QueueWriterTests(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_noise_dropped_keepers_written(self) -> None:
+        triples = [
+            (_record("a"), _verdict("KNOWLEDGE"), "h1"),
+            (_record("b"), _verdict("NOISE"), "h2"),
+            (_record("c"), _verdict("UNCERTAIN", 0.0), "h3"),
+        ]
+        stats = write_verdicts(sqla.session, triples)
+        self.assertEqual((stats.inserted, stats.dropped_noise), (2, 1))
+        labels = sorted(r.llm_label for r in KnowledgeQueueItem.query.all())
+        self.assertEqual(labels, ["KNOWLEDGE", "UNCERTAIN"])
+
+    def test_dedup_within_batch(self) -> None:
+        triples = [
+            (_record("a"), _verdict(), "same"),
+            (_record("b"), _verdict(), "same"),
+        ]
+        stats = write_verdicts(sqla.session, triples)
+        self.assertEqual((stats.inserted, stats.deduped), (1, 1))
+        self.assertEqual(KnowledgeQueueItem.query.count(), 1)
+
+    def test_dedup_against_existing_rows(self) -> None:
+        write_verdicts(sqla.session, [(_record("a"), _verdict(), "h1")])
+        stats = write_verdicts(sqla.session, [(_record("b"), _verdict(), "h1")])
+        self.assertEqual((stats.inserted, stats.deduped), (0, 1))
+        self.assertEqual(KnowledgeQueueItem.query.count(), 1)
+
+    def test_github_source_columns(self) -> None:
+        write_verdicts(
+            sqla.session, [(_record("a", heading_path=["Auth"]), _verdict(), "h1")]
+        )
+        row = KnowledgeQueueItem.query.first()
+        self.assertEqual(row.source_type, "github")
+        self.assertEqual(row.source_repo, "OWASP/test")
+        self.assertEqual(row.source_commit_sha, "abc123")
+        self.assertIsNone(row.feed_url)
+        self.assertEqual(json.loads(row.span_heading_path), ["Auth"])
+
+    def test_rss_source_columns(self) -> None:
+        rss = {
+            "type": "rss",
+            "feed_url": "https://example.org/feed.xml",
+            "post_guid": "guid-123",
+        }
+        write_verdicts(sqla.session, [(_record("a", source=rss), _verdict(), "h1")])
+        row = KnowledgeQueueItem.query.first()
+        self.assertEqual(row.source_type, "rss")
+        self.assertEqual(row.feed_url, "https://example.org/feed.xml")
+        self.assertEqual(row.post_guid, "guid-123")
+        self.assertIsNone(row.source_repo)
+
+
+if __name__ == "__main__":
+    unittest.main()