From 4b1780596363240c484e437967f971ea915bfe23 Mon Sep 17 00:00:00 2001 From: manshusainishab Date: Sun, 19 Jul 2026 06:13:57 +0530 Subject: [PATCH 1/6] feat(module-b): add harvest_input + knowledge_queue models --- application/database/db.py | 71 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/application/database/db.py b/application/database/db.py index 4ca0dc040..444cc491d 100644 --- a/application/database/db.py +++ b/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 From 33de19cc13e3966ff9c0950c704efeac0b636fd1 Mon Sep 17 00:00:00 2001 From: manshusainishab Date: Sun, 19 Jul 2026 06:13:57 +0530 Subject: [PATCH 2/6] feat(module-b): add queue_writer with content_hash dedup --- .../tests/noise_filter/queue_writer_test.py | 108 +++++++++++++++++ .../utils/noise_filter/queue_writer.py | 111 ++++++++++++++++++ 2 files changed, 219 insertions(+) create mode 100644 application/tests/noise_filter/queue_writer_test.py create mode 100644 application/utils/noise_filter/queue_writer.py diff --git a/application/tests/noise_filter/queue_writer_test.py b/application/tests/noise_filter/queue_writer_test.py new file mode 100644 index 000000000..337a77cf0 --- /dev/null +++ b/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() diff --git a/application/utils/noise_filter/queue_writer.py b/application/utils/noise_filter/queue_writer.py new file mode 100644 index 000000000..a83bc1902 --- /dev/null +++ b/application/utils/noise_filter/queue_writer.py @@ -0,0 +1,111 @@ +"""Module B output stage: write classified chunks to `knowledge_queue`. + +This is Module B's DB boundary. It maps a (ChangeRecord, ClassifyResult, +content_hash) triple into a `KnowledgeQueueItem` row and inserts the keepers, +deduped on `content_hash`. NOISE verdicts are dropped (they never reach the +queue); KNOWLEDGE and UNCERTAIN are written (UNCERTAIN is for Module D's HITL +review). Module C reads from `knowledge_queue`; see module_c_contract.md (v0.2). + +The classifier and schemas stay DB-free by design; this module is the only +part of Module B that imports the SQLAlchemy layer. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Iterable + +from application.database.db import KnowledgeQueueItem +from application.utils.noise_filter.schemas import ChangeRecord, ClassifyResult + + +@dataclass(frozen=True) +class WriteStats: + """Outcome of writing a batch of verdicts to the queue.""" + + inserted: int = 0 + deduped: int = 0 + dropped_noise: int = 0 + + +def to_queue_item( + record: ChangeRecord, verdict: ClassifyResult, content_hash: str +) -> KnowledgeQueueItem: + """Map one classified record to a `knowledge_queue` row (unsaved).""" + source = record.source + item = KnowledgeQueueItem( + content_hash=content_hash, + chunk_id=record.chunk_id, + artifact_id=record.artifact_id, + pipeline_run_id=record.pipeline_run_id, + schema_version=record.schema_version, + source_type=source.type, + locator_kind=record.locator.kind, + locator_path=record.locator.path, + span_index=record.span.index, + span_total=record.span.total, + span_heading_path=json.dumps(record.span.heading_path), + text=record.text, + llm_label=verdict.label, + confidence=verdict.confidence, + llm_reasoning=verdict.reasoning, + ) + if source.type == "github": + item.source_repo = source.repo + item.source_commit_sha = source.commit_sha + item.source_committed_at = source.committed_at + elif source.type == "rss": + item.feed_url = source.feed_url + item.post_guid = source.post_guid + return item + + +def write_verdicts( + session, + triples: Iterable[tuple[ChangeRecord, ClassifyResult, str]], +) -> WriteStats: + """Insert keeper verdicts into `knowledge_queue`, deduped on content_hash. + + Args: + session: the SQLAlchemy session (caller owns connect/teardown). + triples: (record, verdict, content_hash) per classified chunk. + + NOISE is dropped. Duplicates (same content_hash already queued, or repeated + within this batch) are skipped. Commits once at the end. + """ + triples = list(triples) + keepers = [(r, v, h) for r, v, h in triples if v.label != "NOISE"] + dropped_noise = len(triples) - len(keepers) + + # Content hashes already present in the queue (single query, not per-row). + candidate_hashes = {h for _, _, h in keepers} + existing: set[str] = set() + if candidate_hashes: + rows = ( + session.query(KnowledgeQueueItem.content_hash) + .filter(KnowledgeQueueItem.content_hash.in_(candidate_hashes)) + .all() + ) + existing = {row[0] for row in rows} + + seen: set[str] = set() + inserted = 0 + deduped = 0 + for record, verdict, content_hash in keepers: + if content_hash in existing or content_hash in seen: + deduped += 1 + continue + seen.add(content_hash) + session.add(to_queue_item(record, verdict, content_hash)) + inserted += 1 + + session.commit() + return WriteStats(inserted=inserted, deduped=deduped, dropped_noise=dropped_noise) + + +__all__ = [ + "WriteStats", + "to_queue_item", + "write_verdicts", +] From 14753f3c8742af2a360e6dfa50edcd69b1ecda8a Mon Sep 17 00:00:00 2001 From: manshusainishab Date: Sun, 19 Jul 2026 06:13:57 +0530 Subject: [PATCH 3/6] feat(module-b): add pipeline (harvest_input -> classify -> knowledge_queue) --- .../tests/noise_filter/pipeline_test.py | 131 ++++++++++++++++ application/utils/noise_filter/pipeline.py | 140 ++++++++++++++++++ 2 files changed, 271 insertions(+) create mode 100644 application/tests/noise_filter/pipeline_test.py create mode 100644 application/utils/noise_filter/pipeline.py diff --git a/application/tests/noise_filter/pipeline_test.py b/application/tests/noise_filter/pipeline_test.py new file mode 100644 index 000000000..b67077857 --- /dev/null +++ b/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() diff --git a/application/utils/noise_filter/pipeline.py b/application/utils/noise_filter/pipeline.py new file mode 100644 index 000000000..1b6c47948 --- /dev/null +++ b/application/utils/noise_filter/pipeline.py @@ -0,0 +1,140 @@ +"""Module B pipeline orchestrator: harvest_input -> classify -> knowledge_queue. + +This is the entry point the orchestrator invokes (via the CLI in cre.py). For a +given `pipeline_run_id` it reads Module A's pending rows from `harvest_input`, +runs the three-stage gate (regex -> sanitize -> LLM classifier), writes the +keepers to `knowledge_queue` (deduped), marks the input rows processed, and +returns a RunSummary the CLI prints as JSON for the orchestrator to consume. + +Recall-first is preserved end to end: only NOISE is dropped; KNOWLEDGE and +UNCERTAIN always reach the queue. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import asdict, dataclass +from typing import Optional + +from pydantic import ValidationError + +from application.database.db import HarvestInput +from application.utils.noise_filter.config_loader import NoiseFilterConfig, load_config +from application.utils.noise_filter.hashing import compute_content_hash +from application.utils.noise_filter.llm_classifier import LLMClassifier +from application.utils.noise_filter.queue_writer import write_verdicts +from application.utils.noise_filter.regex_filter import RegexFilter +from application.utils.noise_filter.sanitize import sanitize_text +from application.utils.noise_filter.schemas import ChangeRecord + +logger = logging.getLogger(__name__) + + +@dataclass +class RunSummary: + """Outcome of one Module B run; the CLI emits this as JSON.""" + + run_id: str + read: int = 0 + parse_errors: int = 0 + dropped_noise: int = 0 # regex-dropped + LLM NOISE + kept_knowledge: int = 0 + kept_uncertain: int = 0 + inserted: int = 0 + deduped: int = 0 + dry_run: bool = False + status: str = "ok" + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + +def _sanitized(record: ChangeRecord) -> ChangeRecord: + """Stage 1.5: copy with sanitized text (no-op on clean input).""" + try: + clean = sanitize_text(record.text) + except ValueError: + return record # sanitization emptied the text; keep original for the LLM + return record.model_copy(update={"text": clean}) + + +def run_noise_filter( + session, + pipeline_run_id: str, + config: Optional[NoiseFilterConfig] = None, + classifier: Optional[LLMClassifier] = None, + *, + dry_run: bool = False, +) -> RunSummary: + """Classify one harvest run's chunks and enqueue the keepers. + + Args: + session: SQLAlchemy session (caller owns connect/teardown). + pipeline_run_id: the run to process (scopes the harvest_input rows). + config: Module B settings; defaults to load_config(). + classifier: injectable LLMClassifier (tests pass a fake); default builds + one from config. + dry_run: classify but do not write to the queue or mark rows processed. + """ + config = config or load_config() + summary = RunSummary(run_id=pipeline_run_id, dry_run=dry_run) + + rows = ( + session.query(HarvestInput) + .filter_by(pipeline_run_id=pipeline_run_id, status="pending") + .all() + ) + summary.read = len(rows) + if not rows: + return summary + + # Parse payloads -> ChangeRecord; invalid rows are counted and flagged. + parsed: list[tuple[HarvestInput, ChangeRecord]] = [] + failed: list[HarvestInput] = [] + for row in rows: + try: + parsed.append((row, ChangeRecord.model_validate(row.payload))) + except ValidationError as e: + summary.parse_errors += 1 + failed.append(row) + logger.warning("harvest_input row %s failed validation: %s", row.id, e) + + # Stage 1: regex path filter (dropped = NOISE). Survivors get Stage 1.5 sanitize. + regex = RegexFilter() + survivors: list[ChangeRecord] = [] + for _row, record in parsed: + is_noise, _reason = regex.is_noise_record(record) + if is_noise: + summary.dropped_noise += 1 + else: + survivors.append(_sanitized(record)) + + # Stage 2: LLM classify. + classifier = classifier or LLMClassifier(config) + verdicts = classifier.classify_batch(survivors) + + triples = [ + (rec, v, compute_content_hash(rec.text)) for rec, v in zip(survivors, verdicts) + ] + summary.kept_knowledge = sum(1 for _, v, _ in triples if v.label == "KNOWLEDGE") + summary.kept_uncertain = sum(1 for _, v, _ in triples if v.label == "UNCERTAIN") + summary.dropped_noise += sum(1 for _, v, _ in triples if v.label == "NOISE") + + if dry_run: + return summary + + write = write_verdicts(session, triples) + summary.inserted = write.inserted + summary.deduped = write.deduped + + # Mark input rows so a re-run doesn't reprocess them. + for row, _ in parsed: + row.status = "processed" + for row in failed: + row.status = "error" + session.commit() + return summary + + +__all__ = ["RunSummary", "run_noise_filter"] From 220aaf43ecc0199096e14b88d8d35f9f7b5ec4fd Mon Sep 17 00:00:00 2001 From: manshusainishab Date: Sun, 19 Jul 2026 06:13:57 +0530 Subject: [PATCH 4/6] feat(module-b): add --run_noise_filter CLI entry point --- application/cmd/cre_main.py | 16 ++++++++++++++++ cre.py | 15 +++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index fbacecdb0..3b6fe8cb9 100644 --- a/application/cmd/cre_main.py +++ b/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 ") + 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, diff --git a/cre.py b/cre.py index b46467281..0d04d5e2b 100644 --- a/cre.py +++ b/cre.py @@ -301,6 +301,21 @@ def main() -> None: default="", help="output CSV path for --export", ) + parser.add_argument( + "--run_noise_filter", + action="store_true", + help="run Module B noise/relevance filter over a harvest run's chunks", + ) + parser.add_argument( + "--run_id", + default="", + help="pipeline_run_id to process (required with --run_noise_filter)", + ) + parser.add_argument( + "--noise_filter_dry_run", + action="store_true", + help="classify without writing to knowledge_queue or marking rows processed", + ) args = parser.parse_args() if args.export and not args.csv: From f56cf1fc00a54d839e5a95e23ee43127f664210f Mon Sep 17 00:00:00 2001 From: manshusainishab Date: Sun, 19 Jul 2026 06:21:29 +0530 Subject: [PATCH 5/6] feat(module-b): add Alembic migration for harvest_input + knowledge_queue --- .../d4e5f6a7b8c9_add_module_b_tables.py | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py diff --git a/migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py b/migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py new file mode 100644 index 000000000..b669083bc --- /dev/null +++ b/migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py @@ -0,0 +1,81 @@ +"""add Module B tables: harvest_input + knowledge_queue + +Revision ID: d4e5f6a7b8c9 +Revises: c7d8e9f0a1b2 +Create Date: 2026-07-19 + +harvest_input -- Module A writes harvested chunks here (JSONB payload); B reads. +knowledge_queue -- Module B writes classified keepers here; Module C reads. +See docs/gsoc_2026_module_b/orchestrator_integration_design.md and +module_c_contract.md (v0.2). +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision = "d4e5f6a7b8c9" +down_revision = "c7d8e9f0a1b2" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "harvest_input", + sa.Column("id", sa.String, primary_key=True), + sa.Column("pipeline_run_id", sa.String, nullable=False), + sa.Column("status", sa.String, nullable=False), + # A's ChangeRecord: JSONB on Postgres, JSON elsewhere. + sa.Column( + "payload", + sa.JSON().with_variant(postgresql.JSONB, "postgresql"), + nullable=False, + ), + sa.Column( + "created_at", sa.DateTime, nullable=False, server_default=sa.func.now() + ), + ) + op.create_index( + "ix_harvest_input_pipeline_run_id", "harvest_input", ["pipeline_run_id"] + ) + op.create_index( + "ix_harvest_input_run_status", "harvest_input", ["pipeline_run_id", "status"] + ) + + op.create_table( + "knowledge_queue", + sa.Column("id", sa.String, primary_key=True), + sa.Column("content_hash", sa.String, nullable=False), + sa.Column("chunk_id", sa.String, nullable=False), + sa.Column("artifact_id", sa.String, nullable=False), + sa.Column("pipeline_run_id", sa.String, nullable=False), + sa.Column("schema_version", sa.String, nullable=False), + sa.Column("source_type", sa.String, nullable=False), + sa.Column("source_repo", sa.String, nullable=True), + sa.Column("source_commit_sha", sa.String, nullable=True), + sa.Column("source_committed_at", sa.String, nullable=True), + sa.Column("feed_url", sa.String, nullable=True), + sa.Column("post_guid", sa.String, nullable=True), + sa.Column("locator_kind", sa.String, nullable=False), + sa.Column("locator_path", sa.String, nullable=False), + sa.Column("span_index", sa.Integer, nullable=False), + sa.Column("span_total", sa.Integer, nullable=False), + sa.Column("span_heading_path", sa.Text, nullable=True), + sa.Column("text", sa.Text, nullable=False), + sa.Column("llm_label", sa.String, nullable=False), + sa.Column("confidence", sa.Float, nullable=False), + sa.Column("llm_reasoning", sa.Text, nullable=True), + sa.Column( + "created_at", sa.DateTime, nullable=False, server_default=sa.func.now() + ), + sa.Column("consumed_at", sa.DateTime, nullable=True), + sa.UniqueConstraint("content_hash", name="uq_content_hash"), + ) + op.create_index("ix_knowledge_queue_unconsumed", "knowledge_queue", ["consumed_at"]) + + +def downgrade(): + op.drop_table("knowledge_queue") + op.drop_table("harvest_input") From 51aa2b6207dd8e3aabf96d3f6024fb00112ea6d1 Mon Sep 17 00:00:00 2001 From: manshusainishab Date: Thu, 23 Jul 2026 09:24:04 +0530 Subject: [PATCH 6/6] fix(module-b): use ON CONFLICT DO NOTHING for idempotent knowledge_queue writes --- .../utils/noise_filter/queue_writer.py | 131 ++++++++++-------- 1 file changed, 77 insertions(+), 54 deletions(-) diff --git a/application/utils/noise_filter/queue_writer.py b/application/utils/noise_filter/queue_writer.py index a83bc1902..fbeaf86eb 100644 --- a/application/utils/noise_filter/queue_writer.py +++ b/application/utils/noise_filter/queue_writer.py @@ -1,10 +1,12 @@ """Module B output stage: write classified chunks to `knowledge_queue`. This is Module B's DB boundary. It maps a (ChangeRecord, ClassifyResult, -content_hash) triple into a `KnowledgeQueueItem` row and inserts the keepers, -deduped on `content_hash`. NOISE verdicts are dropped (they never reach the -queue); KNOWLEDGE and UNCERTAIN are written (UNCERTAIN is for Module D's HITL -review). Module C reads from `knowledge_queue`; see module_c_contract.md (v0.2). +content_hash) triple into a `knowledge_queue` row and inserts the keepers with +DB-level idempotence (`INSERT ... ON CONFLICT (content_hash) DO NOTHING`), so a +concurrent or replayed run that produces the same content never aborts the +batch. NOISE verdicts are dropped (they never reach the queue); KNOWLEDGE and +UNCERTAIN are written (UNCERTAIN is for Module D's HITL review). Module C reads +from `knowledge_queue`; see module_c_contract.md (v0.2). The classifier and schemas stay DB-free by design; this module is the only part of Module B that imports the SQLAlchemy layer. @@ -16,7 +18,7 @@ from dataclasses import dataclass from typing import Iterable -from application.database.db import KnowledgeQueueItem +from application.database.db import KnowledgeQueueItem, generate_uuid from application.utils.noise_filter.schemas import ChangeRecord, ClassifyResult @@ -29,36 +31,56 @@ class WriteStats: dropped_noise: int = 0 +def _row_values( + record: ChangeRecord, verdict: ClassifyResult, content_hash: str +) -> dict: + """Map one classified record to a `knowledge_queue` column dict. + + All provenance columns are present (None where not applicable) so a batch + insert has uniform keys. `created_at` is left to the server default. + """ + source = record.source + is_github = source.type == "github" + is_rss = source.type == "rss" + return { + "id": generate_uuid(), + "content_hash": content_hash, + "chunk_id": record.chunk_id, + "artifact_id": record.artifact_id, + "pipeline_run_id": record.pipeline_run_id, + "schema_version": record.schema_version, + "source_type": source.type, + "source_repo": source.repo if is_github else None, + "source_commit_sha": source.commit_sha if is_github else None, + "source_committed_at": source.committed_at if is_github else None, + "feed_url": source.feed_url if is_rss else None, + "post_guid": source.post_guid if is_rss else None, + "locator_kind": record.locator.kind, + "locator_path": record.locator.path, + "span_index": record.span.index, + "span_total": record.span.total, + "span_heading_path": json.dumps(record.span.heading_path), + "text": record.text, + "llm_label": verdict.label, + "confidence": verdict.confidence, + "llm_reasoning": verdict.reasoning, + } + + def to_queue_item( record: ChangeRecord, verdict: ClassifyResult, content_hash: str ) -> KnowledgeQueueItem: - """Map one classified record to a `knowledge_queue` row (unsaved).""" - source = record.source - item = KnowledgeQueueItem( - content_hash=content_hash, - chunk_id=record.chunk_id, - artifact_id=record.artifact_id, - pipeline_run_id=record.pipeline_run_id, - schema_version=record.schema_version, - source_type=source.type, - locator_kind=record.locator.kind, - locator_path=record.locator.path, - span_index=record.span.index, - span_total=record.span.total, - span_heading_path=json.dumps(record.span.heading_path), - text=record.text, - llm_label=verdict.label, - confidence=verdict.confidence, - llm_reasoning=verdict.reasoning, - ) - if source.type == "github": - item.source_repo = source.repo - item.source_commit_sha = source.commit_sha - item.source_committed_at = source.committed_at - elif source.type == "rss": - item.feed_url = source.feed_url - item.post_guid = source.post_guid - return item + """Map one classified record to a `knowledge_queue` row object (unsaved).""" + return KnowledgeQueueItem(**_row_values(record, verdict, content_hash)) + + +def _dialect_insert(session): + """Return an INSERT construct that supports ON CONFLICT for this backend.""" + if session.get_bind().dialect.name == "postgresql": + from sqlalchemy.dialects.postgresql import insert + else: + from sqlalchemy.dialects.sqlite import insert + return insert(KnowledgeQueueItem) def write_verdicts( @@ -71,37 +93,38 @@ def write_verdicts( session: the SQLAlchemy session (caller owns connect/teardown). triples: (record, verdict, content_hash) per classified chunk. - NOISE is dropped. Duplicates (same content_hash already queued, or repeated - within this batch) are skipped. Commits once at the end. + NOISE is dropped. Duplicates are skipped idempotently: identical content + already queued (or written concurrently by another run) is dropped by + `ON CONFLICT (content_hash) DO NOTHING`, and duplicates repeated within this + batch are collapsed first. Commits once at the end. """ triples = list(triples) keepers = [(r, v, h) for r, v, h in triples if v.label != "NOISE"] dropped_noise = len(triples) - len(keepers) - # Content hashes already present in the queue (single query, not per-row). - candidate_hashes = {h for _, _, h in keepers} - existing: set[str] = set() - if candidate_hashes: - rows = ( - session.query(KnowledgeQueueItem.content_hash) - .filter(KnowledgeQueueItem.content_hash.in_(candidate_hashes)) - .all() - ) - existing = {row[0] for row in rows} + # Collapse duplicate content_hash within this batch -- ON CONFLICT only + # guards against already-committed rows, not duplicates inside one INSERT. + unique: dict[str, tuple[ChangeRecord, ClassifyResult, str]] = {} + for record, verdict, content_hash in keepers: + unique.setdefault(content_hash, (record, verdict, content_hash)) - seen: set[str] = set() inserted = 0 - deduped = 0 - for record, verdict, content_hash in keepers: - if content_hash in existing or content_hash in seen: - deduped += 1 - continue - seen.add(content_hash) - session.add(to_queue_item(record, verdict, content_hash)) - inserted += 1 + if unique: + rows = [_row_values(r, v, h) for (r, v, h) in unique.values()] + stmt = ( + _dialect_insert(session) + .values(rows) + .on_conflict_do_nothing(index_elements=["content_hash"]) + .returning(KnowledgeQueueItem.id) + ) + inserted = len(session.execute(stmt).fetchall()) session.commit() - return WriteStats(inserted=inserted, deduped=deduped, dropped_noise=dropped_noise) + return WriteStats( + inserted=inserted, + deduped=len(keepers) - inserted, + dropped_noise=dropped_noise, + ) __all__ = [