diff --git a/.gitignore b/.gitignore index 33614223c..887ae9d29 100644 --- a/.gitignore +++ b/.gitignore @@ -73,6 +73,7 @@ standards_cache.sqlite !docs/Final_eval_blog_gsoc2026/module_B_final_blog.md !application/utils/librarian/README.md !docs/gsoc_2026_module_c/*.md +!docs/gsoc_2026_module_a/*.md !docs/gsoc_2026_module_b/module_a_contract.md !docs/gsoc_2026_module_b/module_c_contract.md !docs/gsoc_2026_module_b/module_b_runbook.md diff --git a/Makefile b/Makefile index 2f7c3c54b..1eece7b79 100644 --- a/Makefile +++ b/Makefile @@ -181,6 +181,14 @@ alembic-guardrail: [ -d "./venv" ] && . ./venv/bin/activate &&\ python scripts/check_alembic_revision_guardrail.py +oie-pipeline: + [ -d "./venv" ] && . ./venv/bin/activate &&\ + PYTHONPATH=. python scripts/run_oie_pipeline.py --cache_file "$(or $(CACHE_FILE),sqlite:///$(CURDIR)/standards_cache.sqlite)" $(OIE_ARGS) + +oie-e2e-smoke: + [ -d "./venv" ] && . ./venv/bin/activate &&\ + PYTHONPATH=. python scripts/run_oie_e2e_smoke.py + openapi-generate: [ -d "./venv" ] && . ./venv/bin/activate &&\ python scripts/generate_openapi.py diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index 11224e48e..530b67ca0 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -892,6 +892,25 @@ def run(args: argparse.Namespace) -> None: # pragma: no cover logger.info("Exported %s rows to %s", rows, csv_out) return + if getattr(args, "run_harvester", False): + import sys + + from application import sqla + from application.utils.harvester.pipeline import run_harvester + + db_connect(args.cache_file) + repos_yaml = getattr(args, "harvester_repos_yaml", "") or None + summary = run_harvester( + sqla.session, + args.run_id.strip(), + repos_yaml=repos_yaml, + dry_run=getattr(args, "harvester_dry_run", False), + ) + print(summary.to_json()) + if summary.status == "degraded": + sys.exit(1) + return + if getattr(args, "run_noise_filter", False): import sys diff --git a/application/tests/harvester_test/artifact_registry_test.py b/application/tests/harvester_test/artifact_registry_test.py new file mode 100644 index 000000000..39d0c54f7 --- /dev/null +++ b/application/tests/harvester_test/artifact_registry_test.py @@ -0,0 +1,99 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.models import ArtifactRegistryRecord + + +class ArtifactRegistryTests(unittest.TestCase): + def test_insert_record(self): + registry = ArtifactRegistry() + + record = ArtifactRegistryRecord( + artifact_id="art:test:file.md", + repository="OWASP/ASVS", + locator_path="file.md", + content_hash="abc", + last_commit_sha="123", + last_pipeline_run="run1", + last_processed_at=datetime.now(), + status="new", + ) + + registry.upsert(record) + + self.assertTrue(registry.exists(record.artifact_id)) + + def test_get_record(self): + registry = ArtifactRegistry() + + record = ArtifactRegistryRecord( + artifact_id="art:test:file.md", + repository="OWASP/ASVS", + locator_path="file.md", + content_hash="abc", + last_commit_sha="123", + last_pipeline_run="run1", + last_processed_at=datetime.now(), + status="new", + ) + + registry.upsert(record) + + stored = registry.get(record.artifact_id) + assert stored is not None + + self.assertEqual(stored.content_hash, "abc") + + def test_update_record(self): + registry = ArtifactRegistry() + + record = ArtifactRegistryRecord( + artifact_id="art:test:file.md", + repository="OWASP/ASVS", + locator_path="file.md", + content_hash="abc", + last_commit_sha="123", + last_pipeline_run="run1", + last_processed_at=datetime.now(), + status="new", + ) + + registry.upsert(record) + + record.content_hash = "xyz" + record.status = "updated" + + registry.upsert(record) + + stored = registry.get(record.artifact_id) + assert stored is not None + + self.assertEqual(stored.content_hash, "xyz") + self.assertEqual(stored.status, "updated") + + def test_all_records(self): + registry = ArtifactRegistry() + + for i in range(3): + registry.upsert( + ArtifactRegistryRecord( + artifact_id=f"art:{i}", + repository="repo", + locator_path=f"{i}.md", + content_hash=str(i), + last_commit_sha="sha", + last_pipeline_run="run", + last_processed_at=datetime.now(), + status="new", + ) + ) + + self.assertEqual( + len(registry.all()), + 3, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/checkpoint_manager_test.py b/application/tests/harvester_test/checkpoint_manager_test.py new file mode 100644 index 000000000..a3ddb7cea --- /dev/null +++ b/application/tests/harvester_test/checkpoint_manager_test.py @@ -0,0 +1,68 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.checkpoint_manager import CheckpointManager +from application.utils.harvester.models import CheckpointRecord + + +class CheckpointManagerTests(unittest.TestCase): + def test_save_checkpoint(self): + manager = CheckpointManager() + + checkpoint = CheckpointRecord( + repository="OWASP/ASVS", + pipeline_run_id="run1", + last_processed_commit="abc123", + status="running", + updated_at=datetime.now(), + ) + + manager.save(checkpoint) + + self.assertIsNotNone(manager.get("OWASP/ASVS")) + + def test_update_commit(self): + manager = CheckpointManager() + + checkpoint = CheckpointRecord( + repository="OWASP/ASVS", + pipeline_run_id="run1", + last_processed_commit="abc123", + status="running", + updated_at=datetime.now(), + ) + + manager.save(checkpoint) + + manager.update_commit( + "OWASP/ASVS", + "deadbeef", + ) + + stored = manager.get("OWASP/ASVS") + assert stored is not None + + self.assertEqual(stored.last_processed_commit, "deadbeef") + + def test_mark_completed(self): + manager = CheckpointManager() + + checkpoint = CheckpointRecord( + repository="OWASP/ASVS", + pipeline_run_id="run1", + last_processed_commit="abc123", + status="running", + updated_at=datetime.now(), + ) + + manager.save(checkpoint) + manager.mark_completed("OWASP/ASVS") + + stored = manager.get("OWASP/ASVS") + assert stored is not None + + self.assertEqual(stored.status, "completed") + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/chunk_pipeline_test.py b/application/tests/harvester_test/chunk_pipeline_test.py new file mode 100644 index 000000000..858a17709 --- /dev/null +++ b/application/tests/harvester_test/chunk_pipeline_test.py @@ -0,0 +1,36 @@ +import unittest +from unittest.mock import Mock + +from application.utils.harvester.chunk_pipeline import DocumentChunkPipeline +from application.utils.harvester.models import Document, IngestChunkRecord + + +class DocumentChunkPipelineTests(unittest.TestCase): + def test_invalid_record_is_rejected_before_return(self): + document = Mock(spec=Document) + document.text = "Some document text." + + chunker = Mock() + chunker.chunk.return_value = ["chunk"] + + record_builder = Mock() + invalid_record = Mock(spec=IngestChunkRecord) + record_builder.build.return_value = [invalid_record] + + validator = Mock() + validator.validate.side_effect = ValueError("invalid chunk record") + + pipeline = DocumentChunkPipeline( + chunker=chunker, + record_builder=record_builder, + validator=validator, + ) + + with self.assertRaisesRegex(ValueError, "invalid chunk record"): + pipeline.chunk(document) + + validator.validate.assert_called_once_with(invalid_record) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/chunk_record_builder_test.py b/application/tests/harvester_test/chunk_record_builder_test.py new file mode 100644 index 000000000..55f2c1ee4 --- /dev/null +++ b/application/tests/harvester_test/chunk_record_builder_test.py @@ -0,0 +1,75 @@ +import unittest +from datetime import datetime, timezone + +from application.utils.harvester.chunk_record_builder import ChunkRecordBuilder +from application.utils.harvester.chunk_record_validator import ( + ChunkRecordValidator, + ingest_record_to_payload, +) +from application.utils.harvester.chunker import ChunkInfo +from application.utils.harvester.models import ( + Document, + HeadingNode, + Locator, + SourceInfo, +) +from application.utils.noise_filter.schemas import ChangeRecord + + +class ChunkRecordBuilderTests(unittest.TestCase): + def _document(self, text: str, headings: list[HeadingNode]) -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:OWASP/ASVS:README.md", + pipeline_run_id="run-1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc1234deadbeef", + committed_at=datetime(2026, 2, 1, 1, 0, 0, tzinfo=timezone.utc), + ), + locator=Locator( + kind="repo_path", + id="README.md", + path="README.md", + ), + heading_structure=headings, + ) + + def test_builds_change_record_shaped_payload(self) -> None: + text = "# Root\n\nFirst paragraph." + document = self._document( + text, + [HeadingNode(level=1, text="Root", start_line=1, end_line=3)], + ) + chunk = ChunkInfo(text=text, start_char_idx=0, end_char_idx=len(text)) + records = ChunkRecordBuilder().build(document, [chunk]) + self.assertEqual(len(records), 1) + record = records[0] + self.assertEqual(record.chunk_id, "chk:art:OWASP/ASVS:README.md:0") + self.assertEqual(record.pipeline_run_id, "run-1") + self.assertEqual(record.source_repo, "OWASP/ASVS") + self.assertEqual(record.locator_path, "README.md") + self.assertEqual(record.span.heading_path, ["Root"]) + + ChunkRecordValidator().validate(record) + payload = ingest_record_to_payload(record) + ChangeRecord.model_validate(payload) + + def test_indexes_multiple_chunks(self) -> None: + text = "AAAA\n\nBBBB" + document = self._document(text, []) + chunks = [ + ChunkInfo(text="AAAA\n\n", start_char_idx=0, end_char_idx=6), + ChunkInfo(text="BBBB", start_char_idx=6, end_char_idx=10), + ] + records = ChunkRecordBuilder().build(document, chunks) + self.assertEqual(records[0].span.index, 0) + self.assertEqual(records[0].span.total, 2) + self.assertEqual(records[1].span.index, 1) + self.assertEqual(records[1].chunk_id, "chk:art:OWASP/ASVS:README.md:1") + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/chunk_record_validator_test.py b/application/tests/harvester_test/chunk_record_validator_test.py new file mode 100644 index 000000000..c71664cfe --- /dev/null +++ b/application/tests/harvester_test/chunk_record_validator_test.py @@ -0,0 +1,89 @@ +import unittest + +from application.utils.harvester.chunk_record_validator import ( + ChunkRecordValidator, +) +from application.utils.harvester.models import ( + IngestChunkRecord, + SpanInfo, +) + + +def valid_record() -> IngestChunkRecord: + return IngestChunkRecord( + schema_version="0.2.0", + chunk_id="chk:art:OWASP/OpenCRE:README.md:0", + artifact_id="art:OWASP/OpenCRE:README.md", + pipeline_run_id="run-1", + text="Some valid chunk content.", + span=SpanInfo( + heading_path=["Introduction"], + start_line=1, + end_line=2, + index=0, + total=1, + start_char_idx=0, + end_char_idx=25, + ), + source_type="github", + source_repo="OWASP/OpenCRE", + source_commit_sha="abc1234deadbeef", + source_committed_at="2026-02-01T01:00:00Z", + locator_kind="repo_path", + locator_id="README.md", + locator_path="README.md", + ) + + +class ChunkRecordValidatorTests(unittest.TestCase): + def test_valid_record(self) -> None: + ChunkRecordValidator().validate(valid_record()) + + def test_empty_text_is_rejected(self) -> None: + record = valid_record() + record.text = " " + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_invalid_chunk_id_is_rejected(self) -> None: + record = valid_record() + record.chunk_id = "invalid-id" + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_missing_span_index_is_rejected(self) -> None: + record = valid_record() + record.span.index = None + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_index_outside_total_is_rejected(self) -> None: + record = valid_record() + record.span.index = 1 + record.span.total = 1 + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_invalid_character_range_is_rejected(self) -> None: + record = valid_record() + record.span.start_char_idx = 25 + record.span.end_char_idx = 10 + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_negative_character_offsets_are_rejected(self) -> None: + record = valid_record() + record.span.start_char_idx = -1 + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_invalid_line_range_is_rejected(self) -> None: + record = valid_record() + record.span.start_line = 5 + record.span.end_line = 3 + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/chunker_test.py b/application/tests/harvester_test/chunker_test.py new file mode 100644 index 000000000..22180531a --- /dev/null +++ b/application/tests/harvester_test/chunker_test.py @@ -0,0 +1,81 @@ +import unittest +from datetime import datetime, timezone + +from application.utils.harvester.chunker import ChunkInfo, DocumentChunker +from application.utils.harvester.models import ( + Document, + HeadingNode, + Locator, + SourceInfo, +) +from application.utils.harvester.schemas import ChunkingConfig + + +def _doc(text: str, headings: list[HeadingNode] | None = None) -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:OWASP/ASVS:README.md", + pipeline_run_id="run-1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc1234", + committed_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ), + locator=Locator(kind="repo_path", id="README.md", path="README.md"), + heading_structure=headings or [], + ) + + +class DocumentChunkerTests(unittest.TestCase): + def test_empty_document_returns_no_chunks(self) -> None: + chunker = DocumentChunker( + ChunkingConfig(strategy="fixed_size", max_tokens=100, overlap_tokens=10) + ) + self.assertEqual(chunker.chunk(""), []) + + def test_whitespace_document_returns_no_chunks(self) -> None: + chunker = DocumentChunker( + ChunkingConfig(strategy="fixed_size", max_tokens=100, overlap_tokens=10) + ) + self.assertEqual(chunker.chunk(" \n\n "), []) + + def test_markdown_heading_keeps_sections_separate(self) -> None: + text = "# Auth\n\nAAA\n\n# Storage\n\nBBB\n" + document = _doc( + text, + [ + HeadingNode(level=1, text="Auth", start_line=1, end_line=3), + HeadingNode(level=1, text="Storage", start_line=5, end_line=7), + ], + ) + chunker = DocumentChunker( + ChunkingConfig( + strategy="markdown_heading", max_tokens=500, overlap_tokens=10 + ) + ) + chunks = chunker.chunk(text, document=document) + self.assertGreaterEqual(len(chunks), 2) + joined = "".join(c.text for c in chunks) + self.assertIn("AAA", joined) + self.assertIn("BBB", joined) + # Storage content must not start before its heading offset. + storage_start = text.index("# Storage") + storage_chunks = [c for c in chunks if c.start_char_idx >= storage_start] + self.assertTrue(any("BBB" in c.text for c in storage_chunks)) + + def test_fixed_size_respects_budget(self) -> None: + text = ("word " * 200).strip() + chunker = DocumentChunker( + ChunkingConfig(strategy="fixed_size", max_tokens=20, overlap_tokens=5) + ) + chunks = chunker.chunk(text) + self.assertGreater(len(chunks), 1) + for chunk in chunks: + self.assertIsInstance(chunk, ChunkInfo) + self.assertLessEqual(len(chunk.text), 20 * 4 + 5) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/chunking_benchmark_test.py b/application/tests/harvester_test/chunking_benchmark_test.py new file mode 100644 index 000000000..711316f2a --- /dev/null +++ b/application/tests/harvester_test/chunking_benchmark_test.py @@ -0,0 +1,49 @@ +import os +import time +import unittest + +from application.utils.harvester.chunker import DocumentChunker + + +@unittest.skipUnless( + os.getenv("RUN_CHUNKING_BENCHMARK") == "1", + "Chunking benchmark requires RUN_CHUNKING_BENCHMARK=1", +) +class ChunkingBenchmarkTests(unittest.TestCase): + def test_chunking_benchmark(self): + text = ( + "# Introduction\n\n" + "Python functions define reusable behavior. " + "Variables store values and expressions compute results. " * 20 + + "\n\n## Architecture\n\n" + + "The architecture separates ingestion from retrieval. " + "Each component has a clearly defined responsibility. " * 20 + + "\n\n## Storage\n\n" + + "Persistent state is protected by transactional operations. " + "Commit and rollback provide atomicity and consistency. " * 20 + ) + + start = time.perf_counter() + + chunks = DocumentChunker().chunk(text) + + elapsed = time.perf_counter() - start + + self.assertGreater(len(chunks), 0) + self.assertTrue(all(chunk.text.strip() for chunk in chunks)) + self.assertTrue( + all( + 0 <= chunk.start_char_idx < chunk.end_char_idx <= len(text) + for chunk in chunks + ) + ) + + print( + f"\nChunking benchmark: " + f"{len(chunks)} chunks, " + f"{elapsed:.3f}s, " + f"input={len(text)} chars" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/content_hash_test.py b/application/tests/harvester_test/content_hash_test.py new file mode 100644 index 000000000..60d84108f --- /dev/null +++ b/application/tests/harvester_test/content_hash_test.py @@ -0,0 +1,35 @@ +import unittest + +from application.utils.harvester.content_hash import ( + generate_content_hash, +) + + +class ContentHashTests(unittest.TestCase): + def test_same_text_same_hash(self): + text = "Hello World" + + self.assertEqual( + generate_content_hash(text), + generate_content_hash(text), + ) + + def test_different_text_different_hash(self): + self.assertNotEqual( + generate_content_hash("Hello"), + generate_content_hash("World"), + ) + + def test_empty_string(self): + digest = generate_content_hash("") + + self.assertEqual(len(digest), 64) + + def test_hash_is_hex(self): + digest = generate_content_hash("OpenCRE") + + int(digest, 16) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/deduplication_metrics_test.py b/application/tests/harvester_test/deduplication_metrics_test.py new file mode 100644 index 000000000..13e4e1152 --- /dev/null +++ b/application/tests/harvester_test/deduplication_metrics_test.py @@ -0,0 +1,35 @@ +import unittest + +from application.utils.harvester.deduplication_metrics import DeduplicationMetrics +from application.utils.harvester.models import DeduplicationStatus + + +class DeduplicationMetricsTests(unittest.TestCase): + def test_records_new_document(self): + metrics = DeduplicationMetrics() + + metrics.record(DeduplicationStatus.NEW) + + self.assertEqual(metrics.total_artifacts_scanned, 1) + self.assertEqual(metrics.artifacts_new, 1) + self.assertEqual(metrics.artifacts_emitted, 1) + + def test_records_updated_document(self): + metrics = DeduplicationMetrics() + + metrics.record(DeduplicationStatus.UPDATED) + + self.assertEqual(metrics.artifacts_updated, 1) + self.assertEqual(metrics.artifacts_emitted, 1) + + def test_records_unchanged_document(self): + metrics = DeduplicationMetrics() + + metrics.record(DeduplicationStatus.UNCHANGED) + + self.assertEqual(metrics.artifacts_unchanged, 1) + self.assertEqual(metrics.artifacts_skipped, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/diff_normalizer_test.py b/application/tests/harvester_test/diff_normalizer_test.py index 04eb1ce3f..075b8fae2 100644 --- a/application/tests/harvester_test/diff_normalizer_test.py +++ b/application/tests/harvester_test/diff_normalizer_test.py @@ -9,7 +9,6 @@ DiffBlock, ) - DIFF_METADATA = { "repository": "OWASP/ASVS", "commit_sha": "abc123", diff --git a/application/tests/harvester_test/diff_parser_test.py b/application/tests/harvester_test/diff_parser_test.py index a4444e8d9..877d70a09 100644 --- a/application/tests/harvester_test/diff_parser_test.py +++ b/application/tests/harvester_test/diff_parser_test.py @@ -10,6 +10,11 @@ TEST_COMMITTED_AT = datetime.now(UTC) +TEST_REPOSITORY = "OWASP/ASVS" +TEST_COMMIT_SHA = "abc123" +TEST_COMMITTED_AT = datetime.now(UTC) + + class DiffParserTests(unittest.TestCase): def test_single_file_diff(self): parser = DiffParser() diff --git a/application/tests/harvester_test/diff_pipeline_test.py b/application/tests/harvester_test/diff_pipeline_test.py index e5420a4ed..1f29048fd 100644 --- a/application/tests/harvester_test/diff_pipeline_test.py +++ b/application/tests/harvester_test/diff_pipeline_test.py @@ -1,8 +1,8 @@ -from datetime import UTC, datetime import os import subprocess import time import unittest +from datetime import UTC, datetime from application.utils.harvester.diff_normalizer import DiffNormalizer from application.utils.harvester.diff_parser import DiffParser @@ -10,10 +10,13 @@ from application.utils.harvester.git_repository_client import GitRepositoryClient +@unittest.skipUnless( + os.getenv("RUN_DIFF_PIPELINE_BENCHMARK") == "1", + "Diff pipeline benchmark requires RUN_DIFF_PIPELINE_BENCHMARK=1", +) class DiffPipelineBenchmark(unittest.TestCase): """ Simple benchmark to ensure the complete diff pipeline remains fast. - This is not intended as a strict performance benchmark, only as a regression guard against accidental slowdowns. """ @@ -63,3 +66,7 @@ def test_pipeline_benchmark(self): print(f"\nPipeline took {elapsed:.3f}s") self.assertLess(elapsed, 5) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/diff_retriever_test.py b/application/tests/harvester_test/diff_retriever_test.py index b587bbe33..1a23e0b8e 100644 --- a/application/tests/harvester_test/diff_retriever_test.py +++ b/application/tests/harvester_test/diff_retriever_test.py @@ -1,7 +1,7 @@ import unittest from unittest.mock import MagicMock -from unittest.mock import patch from unittest.mock import call +from unittest.mock import patch from application.utils.harvester.diff_retriever import ( DiffRetriever, @@ -24,7 +24,6 @@ def test_get_diff(self, mock_run): client.get_local_path.return_value = "/tmp/repo" retriever = DiffRetriever(client) - diff = retriever.get_diff( "abc123", "def456", diff --git a/application/tests/harvester_test/document_deduplicator_test.py b/application/tests/harvester_test/document_deduplicator_test.py new file mode 100644 index 000000000..64601713f --- /dev/null +++ b/application/tests/harvester_test/document_deduplicator_test.py @@ -0,0 +1,77 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.document_deduplicator import ( + DocumentDeduplicator, +) +from application.utils.harvester.models import ( + DeduplicationStatus, + Document, + Locator, + SourceInfo, +) + + +class DocumentDeduplicatorTests(unittest.TestCase): + def create_document(self, text: str) -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:test:file.md", + pipeline_run_id="run1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + ), + locator=Locator( + kind="repo_path", + id="file.md", + path="file.md", + ), + heading_structure=[], + span=None, + ) + + def test_new_document(self): + registry = ArtifactRegistry() + + deduplicator = DocumentDeduplicator(registry) + result = deduplicator.process(self.create_document("hello")) + self.assertEqual(result, DeduplicationStatus.NEW) + + def test_unchanged_document_refreshes_commit_metadata(self) -> None: + registry = ArtifactRegistry() + deduplicator = DocumentDeduplicator(registry) + first = self.create_document("hello") + first.source.commit_sha = "aaa111" + deduplicator.process(first) + + second = self.create_document("hello") + second.source.commit_sha = "bbb222" + second.pipeline_run_id = "run2" + result = deduplicator.process(second) + self.assertEqual(result, DeduplicationStatus.UNCHANGED) + stored = registry.get(second.artifact_id) + assert stored is not None + self.assertEqual(stored.last_commit_sha, "bbb222") + self.assertEqual(stored.last_pipeline_run, "run2") + + def test_updated_document(self): + registry = ArtifactRegistry() + + deduplicator = DocumentDeduplicator(registry) + + deduplicator.process(self.create_document("hello")) + + result = deduplicator.process( + self.create_document("changed"), + ) + + self.assertEqual(result, DeduplicationStatus.UPDATED) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/document_validator_test.py b/application/tests/harvester_test/document_validator_test.py index c445d7245..7de91263b 100644 --- a/application/tests/harvester_test/document_validator_test.py +++ b/application/tests/harvester_test/document_validator_test.py @@ -71,14 +71,17 @@ def test_invalid_source_type(self): self.assertFalse(validator.validate(document)) - def test_non_markdown_document_is_valid(self): + def test_reject_bare_art_prefix(self) -> None: validator = DocumentValidator() - document = make_document() - document.heading_structure = [] - document.text = '{"hello": "world"}' + document.artifact_id = "art:" + self.assertFalse(validator.validate(document)) - self.assertTrue(validator.validate(document)) + def test_reject_locator_id_mismatch(self) -> None: + validator = DocumentValidator() + document = make_document() + document.locator.id = "other.md" + self.assertFalse(validator.validate(document)) if __name__ == "__main__": diff --git a/application/tests/harvester_test/git_repository_client_integration_test.py b/application/tests/harvester_test/git_repository_client_integration_test.py index f2f585098..c12d923bb 100644 --- a/application/tests/harvester_test/git_repository_client_integration_test.py +++ b/application/tests/harvester_test/git_repository_client_integration_test.py @@ -21,8 +21,9 @@ def repository_url(self) -> str: def git(*args, cwd=None): + # Disable hooks so CI/sandbox environments that block hook writes still work. subprocess.run( - ["git", *args], + ["git", "-c", "core.hooksPath=/dev/null", *args], cwd=cwd, check=True, capture_output=True, @@ -32,7 +33,7 @@ def git(*args, cwd=None): def git_output(*args, cwd=None): return subprocess.run( - ["git", *args], + ["git", "-c", "core.hooksPath=/dev/null", *args], cwd=cwd, check=True, capture_output=True, @@ -42,16 +43,27 @@ def git_output(*args, cwd=None): class GitRepositoryClientIntegrationTests(unittest.TestCase): def setUp(self): - self.tempdir = tempfile.TemporaryDirectory() + # Keep temp dirs inside the repo so sandboxed runners can write hooks/objects. + self._tmpdir_root = Path(__file__).resolve().parents[3] / "tmp" / "harvester-it" + self._tmpdir_root.mkdir(parents=True, exist_ok=True) + self.tempdir = tempfile.TemporaryDirectory(dir=self._tmpdir_root) self.root = Path(self.tempdir.name) self.remote = self.root / "remote.git" self.work = self.root / "work" self.cache = self.root / "cache" - git("init", "--bare", self.remote) - - git("clone", self.remote, self.work) + git("init", "--bare", str(self.remote)) + + try: + git("clone", str(self.remote), str(self.work)) + except subprocess.CalledProcessError as exc: + err = (exc.stderr or "") + (exc.stdout or "") + if "Operation not permitted" in err: + raise unittest.SkipTest( + "environment blocks git clone/config writes (sandbox)" + ) from exc + raise git("config", "user.name", "Test User", cwd=self.work) git("config", "user.email", "test@example.com", cwd=self.work) diff --git a/application/tests/harvester_test/git_repository_client_test.py b/application/tests/harvester_test/git_repository_client_test.py index d012594c9..98fc07fcb 100644 --- a/application/tests/harvester_test/git_repository_client_test.py +++ b/application/tests/harvester_test/git_repository_client_test.py @@ -116,7 +116,6 @@ def test_checkout_runs_git_command(self, mock_run): "-C", str(client.get_local_path()), "checkout", - "--", "main", ], check=True, @@ -156,8 +155,10 @@ def test_clone_runs_git_command(self, mock_run): @patch("application.utils.harvester.git_repository_client.subprocess.run") def test_get_file_at_commit(self, mock_run): - - mock_run.return_value = MagicMock(stdout="# Hello\nWorld\n") + mock_run.side_effect = [ + MagicMock(stdout="42\n"), + MagicMock(stdout="# Hello\nWorld\n"), + ] client = GitRepositoryClient("OWASP", "ASVS", "master") @@ -166,20 +167,8 @@ def test_get_file_at_commit(self, mock_run): content = client.get_file_at_commit("abc123", "README.md") self.assertEqual(content, "# Hello\nWorld\n") - - mock_run.assert_called_once_with( - [ - "git", - "-C", - "/tmp/repo", - "show", - "abc123:README.md", - ], - capture_output=True, - text=True, - check=True, - timeout=30, - ) + self.assertEqual(mock_run.call_count, 2) + self.assertIn("--end-of-options", mock_run.call_args_list[1].args[0]) if __name__ == "__main__": diff --git a/application/tests/harvester_test/harvest_pipeline_test.py b/application/tests/harvester_test/harvest_pipeline_test.py new file mode 100644 index 000000000..50e5769ee --- /dev/null +++ b/application/tests/harvester_test/harvest_pipeline_test.py @@ -0,0 +1,187 @@ +import unittest +from datetime import datetime, timezone +from unittest.mock import Mock + +from application import create_app, sqla +from application.database.db import HarvestInput +from application.utils.harvester.chunk_pipeline import DocumentChunkPipeline +from application.utils.harvester.harvest_writer import write_harvest_input +from application.utils.harvester.models import ( + Document, + HeadingNode, + IngestChunkRecord, + Locator, + SourceInfo, + SpanInfo, +) +from application.utils.harvester.schemas import ChunkingConfig +from application.utils.noise_filter.schemas import ChangeRecord +from application.utils.oie_orchestrator import run_oie_pipeline + + +class HarvestWriterTests(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.app_context = self.app.app_context() + self.app_context.push() + sqla.create_all() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.app_context.pop() + + def test_writes_pending_change_records(self) -> None: + record = IngestChunkRecord( + schema_version="0.2.0", + chunk_id="chk:art:OWASP/ASVS:a.md:0", + artifact_id="art:OWASP/ASVS:a.md", + pipeline_run_id="run-xyz", + text="Authentication should use MFA", + span=SpanInfo( + heading_path=["Auth"], + start_line=1, + end_line=1, + index=0, + total=1, + start_char_idx=0, + end_char_idx=30, + ), + source_type="github", + source_repo="OWASP/ASVS", + source_commit_sha="abc1234deadbeef", + source_committed_at="2026-02-01T01:00:00Z", + locator_kind="repo_path", + locator_id="a.md", + locator_path="a.md", + ) + written = write_harvest_input(sqla.session, "run-xyz", [record]) + self.assertEqual(written, 1) + row = sqla.session.query(HarvestInput).one() + self.assertEqual(row.pipeline_run_id, "run-xyz") + self.assertEqual(row.status, "pending") + ChangeRecord.model_validate(row.payload) + self.assertEqual(row.payload["pipeline_run_id"], "run-xyz") + + +class DocumentChunkPipelineIntegrationTests(unittest.TestCase): + def test_emits_valid_change_records(self) -> None: + text = "# Auth\n\nUse MFA everywhere.\n" + document = Document( + schema_version="0.2.0", + artifact_id="art:OWASP/ASVS:auth.md", + pipeline_run_id="run-1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc1234deadbeef", + committed_at=datetime(2026, 2, 1, tzinfo=timezone.utc), + ), + locator=Locator(kind="repo_path", id="auth.md", path="auth.md"), + heading_structure=[ + HeadingNode(level=1, text="Auth", start_line=1, end_line=3) + ], + ) + pipeline = DocumentChunkPipeline( + chunking=ChunkingConfig( + strategy="markdown_heading", max_tokens=200, overlap_tokens=10 + ) + ) + records = pipeline.chunk(document) + self.assertGreaterEqual(len(records), 1) + for record in records: + ChangeRecord.model_validate( + { + "schema_version": record.schema_version, + "chunk_id": record.chunk_id, + "artifact_id": record.artifact_id, + "pipeline_run_id": record.pipeline_run_id, + "text": record.text, + "span": { + "index": record.span.index, + "total": record.span.total, + "heading_path": record.span.heading_path, + "start_char_idx": record.span.start_char_idx, + "end_char_idx": record.span.end_char_idx, + "start_line": record.span.start_line, + "end_line": record.span.end_line, + }, + "source": { + "type": record.source_type, + "repo": record.source_repo, + "commit_sha": record.source_commit_sha, + "committed_at": record.source_committed_at, + }, + "locator": { + "kind": record.locator_kind, + "id": record.locator_id, + "path": record.locator_path, + }, + } + ) + + +class OieOrchestratorTests(unittest.TestCase): + def test_sequences_a_b_c_and_stops_on_a_error(self) -> None: + a_calls = [] + + def run_a(session, run_id, **kwargs): + a_calls.append(run_id) + summary = Mock() + summary.status = "degraded" + summary.to_json.return_value = '{"status":"degraded"}' + return summary + + b_calls = [] + + def run_b(session, run_id, **kwargs): + b_calls.append(run_id) + summary = Mock() + summary.status = "ok" + summary.to_json.return_value = '{"status":"ok"}' + return summary + + result = run_oie_pipeline( + cache_file="sqlite://", + pipeline_run_id="run-1", + dry_run=True, + sync_repos=False, + run_harvester_fn=run_a, + run_noise_filter_fn=run_b, + run_librarian_queue_fn=lambda *a, **k: {"ok": True}, + ) + self.assertEqual(a_calls, ["run-1"]) + self.assertEqual(b_calls, []) + self.assertFalse(result.to_dict()["ok"]) + self.assertEqual(result.stages[0].status, "error") + + def test_runs_all_stages_when_ok(self) -> None: + def ok_summary(*args, **kwargs): + summary = Mock() + summary.status = "ok" + summary.to_json.return_value = '{"status":"ok"}' + return summary + + result = run_oie_pipeline( + cache_file="sqlite://", + pipeline_run_id="run-2", + dry_run=True, + sync_repos=False, + run_harvester_fn=ok_summary, + run_noise_filter_fn=ok_summary, + run_librarian_queue_fn=lambda *a, **k: {"status": "ok"}, + ) + self.assertTrue(result.to_dict()["ok"]) + self.assertEqual( + [s.name for s in result.stages], + [ + "module_a_harvester", + "module_b_noise_filter", + "module_c_librarian", + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/heading_extractor_test.py b/application/tests/harvester_test/heading_extractor_test.py index e1411b9cd..94413edd3 100644 --- a/application/tests/harvester_test/heading_extractor_test.py +++ b/application/tests/harvester_test/heading_extractor_test.py @@ -102,6 +102,28 @@ def test_heading_stops_at_same_level(self): self.assertEqual(headings[1].end_line, 7) self.assertEqual(headings[2].end_line, 7) + def test_ignores_headings_inside_fenced_code(self) -> None: + text = """# Real + +``` +# Not A Heading +``` + +## Also Real +""" + headings = HeadingExtractor().extract(text) + self.assertEqual([h.text for h in headings], ["Real", "Also Real"]) + + def test_ignores_indented_code_headings(self) -> None: + text = """# Real + + # Indented Fake + +## Also Real +""" + headings = HeadingExtractor().extract(text) + self.assertEqual([h.text for h in headings], ["Real", "Also Real"]) + if __name__ == "__main__": unittest.main() diff --git a/application/tests/harvester_test/incremental_pipeline_test.py b/application/tests/harvester_test/incremental_pipeline_test.py new file mode 100644 index 000000000..9728a8c47 --- /dev/null +++ b/application/tests/harvester_test/incremental_pipeline_test.py @@ -0,0 +1,80 @@ +import unittest +from datetime import datetime +from unittest.mock import Mock + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.document_deduplicator import DocumentDeduplicator +from application.utils.harvester.incremental_pipeline import IncrementalPipeline +from application.utils.harvester.models import Document, Locator, SourceInfo + + +class IncrementalPipelineTests(unittest.TestCase): + def make_document(self, text: str, commit_sha: str = "abc1234") -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:OWASP/ASVS:file.md", + pipeline_run_id="run1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha=commit_sha, + committed_at=datetime.now(), + ), + locator=Locator( + kind="repo_path", + id="file.md", + path="file.md", + ), + heading_structure=[], + span=None, + ) + + def test_only_new_and_updated_are_emitted(self) -> None: + registry = ArtifactRegistry() + dedup = DocumentDeduplicator(registry) + store = Mock() + pipeline = IncrementalPipeline( + deduplicator=dedup, + checkpoint_store=store, + provider="github", + owner="OWASP", + repository_name="ASVS", + branch="master", + repository_id="owasp-asvs", + ) + + docs = [ + self.make_document("hello"), + self.make_document("hello"), + self.make_document("changed"), + ] + emitted = pipeline.process( + "OWASP/ASVS", + "run1", + docs, + last_processed_commit="abc1234", + ) + self.assertEqual(len(emitted), 2) + store.save.assert_called_once() + saved = store.save.call_args[0][0] + self.assertEqual(saved.last_processed_commit, "abc1234") + + def test_rejects_empty_checkpoint_commit(self) -> None: + pipeline = IncrementalPipeline( + checkpoint_store=Mock(), + owner="OWASP", + repository_name="ASVS", + repository_id="owasp-asvs", + ) + with self.assertRaises(ValueError): + pipeline.process( + "OWASP/ASVS", + "run1", + [self.make_document("hello")], + last_processed_commit=" ", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index af2d96ded..e89b95fe4 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -26,34 +26,54 @@ FilteringBenchmarkResult, ) -from .heading_extractor import ( - HeadingExtractor, - HeadingNode, -) +from .heading_extractor import HeadingExtractor +from .models import HeadingNode from .document_builder import DocumentBuilder from .document_validator import DocumentValidator +from .content_hash import generate_content_hash +from .artifact_registry import ArtifactRegistry +from .document_deduplicator import DocumentDeduplicator +from .checkpoint_manager import CheckpointManager +from .checkpoint_store import CheckpointStore +from .incremental_pipeline import IncrementalPipeline +from .deduplication_metrics import DeduplicationMetrics +from .chunker import ChunkInfo, DocumentChunker +from .chunk_pipeline import DocumentChunkPipeline +from .pipeline import RunSummary, run_harvester __all__ = [ + "ArtifactRegistry", "build_repository_cache_path", + "CheckpointManager", + "CheckpointStore", + "ChunkInfo", "ChunkingConfig", "ConfigLoaderError", + "DeduplicationMetrics", "DiffRetriever", "DocumentBuilder", + "DocumentChunker", + "DocumentChunkPipeline", + "DocumentDeduplicator", "DocumentValidator", - "GitRepositoryClient", "FileFilter", - "FilteringMetricsCollector", "FilteringBenchmark", "FilteringBenchmarkResult", + "FilteringMetricsCollector", + "generate_content_hash", + "GitRepositoryClient", "HeadingExtractor", "HeadingNode", + "IncrementalPipeline", "PathRules", "PollingConfig", "RepositoryClient", "RepositoryConfig", "RepositoryValidationError", "ReposFile", + "RunSummary", "load_repo_config", + "run_harvester", "validate_repositories", ] diff --git a/application/utils/harvester/artifact_registry.py b/application/utils/harvester/artifact_registry.py new file mode 100644 index 000000000..621f8e567 --- /dev/null +++ b/application/utils/harvester/artifact_registry.py @@ -0,0 +1,24 @@ +from datetime import datetime +from .models import ArtifactRegistryRecord + + +class ArtifactRegistry: + """ + In-memory registry for artifact deduplication. + """ + + def __init__(self): + self._records: dict[str, ArtifactRegistryRecord] = {} + + def get(self, artifact_id: str) -> ArtifactRegistryRecord | None: + return self._records.get(artifact_id) + + def exists(self, artifact_id: str) -> bool: + return artifact_id in self._records + + def upsert(self, record: ArtifactRegistryRecord) -> None: + record.last_processed_at = datetime.now() + self._records[record.artifact_id] = record + + def all(self) -> list[ArtifactRegistryRecord]: + return list(self._records.values()) diff --git a/application/utils/harvester/checkpoint_manager.py b/application/utils/harvester/checkpoint_manager.py new file mode 100644 index 000000000..c9f0fdd3c --- /dev/null +++ b/application/utils/harvester/checkpoint_manager.py @@ -0,0 +1,36 @@ +from datetime import datetime + +from .models import CheckpointRecord + + +class CheckpointManager: + """ + Stores pipeline checkpoints for incremental processing. + """ + + def __init__(self): + self._checkpoints: dict[str, CheckpointRecord] = {} + + def save(self, checkpoint: CheckpointRecord) -> None: + self._checkpoints[checkpoint.repository] = checkpoint + + def get(self, repository: str) -> CheckpointRecord | None: + return self._checkpoints.get(repository) + + def update_commit(self, repository: str, commit_sha: str) -> None: + checkpoint = self._checkpoints.get(repository) + + if checkpoint is None: + return + + checkpoint.last_processed_commit = commit_sha + checkpoint.updated_at = datetime.now() + + def mark_completed(self, repository: str) -> None: + checkpoint = self._checkpoints.get(repository) + + if checkpoint is None: + return + + checkpoint.status = "completed" + checkpoint.updated_at = datetime.now() diff --git a/application/utils/harvester/chunk_pipeline.py b/application/utils/harvester/chunk_pipeline.py new file mode 100644 index 000000000..7beb30857 --- /dev/null +++ b/application/utils/harvester/chunk_pipeline.py @@ -0,0 +1,30 @@ +from .chunk_record_builder import ChunkRecordBuilder +from .chunk_record_validator import ChunkRecordValidator +from .chunker import DocumentChunker +from .models import Document, IngestChunkRecord +from .schemas import ChunkingConfig + + +class DocumentChunkPipeline: + """ + Runs config-driven chunking followed by structure-aware RFC + chunk-record construction and validation. + """ + + def __init__( + self, + chunking: ChunkingConfig | None = None, + chunker: DocumentChunker | None = None, + record_builder: ChunkRecordBuilder | None = None, + validator: ChunkRecordValidator | None = None, + ) -> None: + self._chunker = chunker or DocumentChunker(chunking) + self._record_builder = record_builder or ChunkRecordBuilder() + self._validator = validator or ChunkRecordValidator() + + def chunk(self, document: Document) -> list[IngestChunkRecord]: + chunks = self._chunker.chunk(document.text, document=document) + records = self._record_builder.build(document, chunks) + for record in records: + self._validator.validate(record) + return records diff --git a/application/utils/harvester/chunk_record_builder.py b/application/utils/harvester/chunk_record_builder.py new file mode 100644 index 000000000..8cc292536 --- /dev/null +++ b/application/utils/harvester/chunk_record_builder.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Optional + +from .chunker import DocumentChunker +from .models import ChunkInfo, Document, IngestChunkRecord, SpanInfo + +if TYPE_CHECKING: + from .schemas import ChunkingConfig + + +@dataclass(slots=True) +class ChunkRecordBuilder: + """ + Converts ChunkInfo objects into Module-B-facing ingest records. + """ + + SCHEMA_VERSION = "0.2.0" + + def build( + self, + document: Document, + chunks: list[ChunkInfo], + ) -> list[IngestChunkRecord]: + total = len(chunks) + committed_at = document.source.committed_at + if isinstance(committed_at, datetime): + if committed_at.tzinfo is None: + committed_at = committed_at.replace(tzinfo=timezone.utc) + committed_at_str = committed_at.isoformat().replace("+00:00", "Z") + else: + committed_at_str = str(committed_at) + + records: list[IngestChunkRecord] = [] + for index, chunk in enumerate(chunks): + heading_path = self._heading_path_for_chunk(document, chunk) + start_line, end_line = self._line_range( + document.text, + chunk.start_char_idx, + chunk.end_char_idx, + ) + records.append( + IngestChunkRecord( + schema_version=self.SCHEMA_VERSION, + chunk_id=f"chk:{document.artifact_id}:{index}", + artifact_id=document.artifact_id, + pipeline_run_id=document.pipeline_run_id, + text=chunk.text, + span=SpanInfo( + heading_path=heading_path, + start_line=start_line, + end_line=end_line, + index=index, + total=total, + start_char_idx=chunk.start_char_idx, + end_char_idx=chunk.end_char_idx, + ), + source_type=document.source.type, + source_repo=document.source.repository, + source_commit_sha=document.source.commit_sha, + source_committed_at=committed_at_str, + locator_kind=document.locator.kind, + locator_id=document.locator.id, + locator_path=document.locator.path, + ) + ) + return records + + @staticmethod + def _heading_path_for_chunk( + document: Document, + chunk: ChunkInfo, + ) -> list[str]: + start_line, _ = ChunkRecordBuilder._line_range( + document.text, + chunk.start_char_idx, + chunk.end_char_idx, + ) + active = [ + heading + for heading in document.heading_structure + if heading.start_line <= start_line <= heading.end_line + ] + active.sort(key=lambda heading: heading.start_line) + path: list[str] = [] + for heading in active: + while len(path) >= heading.level: + path.pop() + path.append(heading.text) + return path + + @staticmethod + def _line_range( + text: str, + start_char_idx: int, + end_char_idx: int, + ) -> tuple[int, int]: + if not 0 <= start_char_idx < end_char_idx <= len(text): + raise ValueError("Chunk character offsets are outside the source document") + start_line = text.count("\n", 0, start_char_idx) + 1 + end_position = end_char_idx - 1 + end_line = text.count("\n", 0, end_position) + 1 + return start_line, end_line + + +def chunk_document( + document: Document, + config: Optional["ChunkingConfig"] = None, +) -> list[IngestChunkRecord]: + chunker = DocumentChunker(config) + chunks = chunker.chunk(document.text, document=document) + return ChunkRecordBuilder().build(document, chunks) diff --git a/application/utils/harvester/chunk_record_validator.py b/application/utils/harvester/chunk_record_validator.py new file mode 100644 index 000000000..0645e69fa --- /dev/null +++ b/application/utils/harvester/chunk_record_validator.py @@ -0,0 +1,89 @@ +from application.utils.noise_filter.schemas import ChangeRecord + +from .models import IngestChunkRecord + + +class ChunkRecordValidator: + """ + Validates RFC-facing ingestion chunk records against Module B's contract. + """ + + def validate(self, record: IngestChunkRecord) -> None: + if not record.schema_version.strip(): + raise ValueError("Chunk record schema_version must not be empty") + + if not record.chunk_id.startswith("chk:"): + raise ValueError("Chunk record chunk_id must start with 'chk:'") + + if not record.artifact_id.strip(): + raise ValueError("Chunk record artifact_id must not be empty") + + if not record.pipeline_run_id.strip(): + raise ValueError("Chunk record pipeline_run_id must not be empty") + + if not record.text.strip(): + raise ValueError("Chunk record text must not be empty") + + span = record.span + if span.index is None or span.total is None: + raise ValueError("Chunk record span must contain index and total") + + if span.index < 0: + raise ValueError("Chunk record span.index must be non-negative") + + if span.total <= 0: + raise ValueError("Chunk record span.total must be positive") + + if span.index >= span.total: + raise ValueError("Chunk record span.index must be less than total") + + if span.start_char_idx is None or span.end_char_idx is None: + raise ValueError("Chunk record span must contain character offsets") + + if span.start_char_idx < 0 or span.end_char_idx < 0: + raise ValueError("Chunk record character offsets must be non-negative") + + if span.start_char_idx >= span.end_char_idx: + raise ValueError( + "Chunk record start_char_idx must be less than end_char_idx" + ) + + if span.start_line <= 0: + raise ValueError("Chunk record start_line must be positive") + + if span.end_line < span.start_line: + raise ValueError("Chunk record end_line must not precede start_line") + + # Canonical gate: must round-trip Module B ChangeRecord. + ChangeRecord.model_validate(ingest_record_to_payload(record)) + + +def ingest_record_to_payload(record: IngestChunkRecord) -> dict: + """Serialize an ingest record to the Module A → B JSON payload shape.""" + return { + "schema_version": record.schema_version, + "chunk_id": record.chunk_id, + "artifact_id": record.artifact_id, + "pipeline_run_id": record.pipeline_run_id, + "text": record.text, + "span": { + "index": record.span.index, + "total": record.span.total, + "heading_path": list(record.span.heading_path), + "start_char_idx": record.span.start_char_idx, + "end_char_idx": record.span.end_char_idx, + "start_line": record.span.start_line, + "end_line": record.span.end_line, + }, + "source": { + "type": record.source_type, + "repo": record.source_repo, + "commit_sha": record.source_commit_sha, + "committed_at": record.source_committed_at, + }, + "locator": { + "kind": record.locator_kind, + "id": record.locator_id, + "path": record.locator_path, + }, + } diff --git a/application/utils/harvester/chunker.py b/application/utils/harvester/chunker.py new file mode 100644 index 000000000..3eec8a369 --- /dev/null +++ b/application/utils/harvester/chunker.py @@ -0,0 +1,115 @@ +from .models import ChunkInfo, Document +from .schemas import ChunkingConfig + +__all__ = ["ChunkInfo", "DocumentChunker"] + + +class DocumentChunker: + """ + Splits documents using the repository ``ChunkingConfig``. + + Strategies: + - ``markdown_heading``: one chunk per heading section (plus preamble), + then size-split oversized sections. + - ``fixed_size``: sliding windows by approximate token budget. + - ``html_readability``: treated as fixed_size until a dedicated parser exists. + """ + + # Rough chars-per-token for budget checks without a tokenizer dependency. + CHARS_PER_TOKEN = 4 + + def __init__(self, config: ChunkingConfig | None = None) -> None: + self._config = config + + def chunk(self, text: str, *, document: Document | None = None) -> list[ChunkInfo]: + if not text.strip(): + return [] + + config = self._config + if config is None: + return self._fixed_size(text, max_tokens=1200, overlap_tokens=100) + + strategy = config.strategy + if strategy == "markdown_heading" and document is not None: + return self._markdown_heading(text, document, config) + return self._fixed_size( + text, + max_tokens=config.max_tokens, + overlap_tokens=config.overlap_tokens, + ) + + def _markdown_heading( + self, text: str, document: Document, config: ChunkingConfig + ) -> list[ChunkInfo]: + headings = document.heading_structure + if not headings: + return self._fixed_size( + text, + max_tokens=config.max_tokens, + overlap_tokens=config.overlap_tokens, + ) + + lines = text.splitlines(keepends=True) + # Map 1-based line -> char offset of line start. + line_starts = [0] + for line in lines: + line_starts.append(line_starts[-1] + len(line)) + + sections: list[tuple[int, int]] = [] + first_heading_start = headings[0].start_line + if first_heading_start > 1: + sections.append((1, first_heading_start - 1)) + + for heading in headings: + sections.append((heading.start_line, heading.end_line)) + + chunks: list[ChunkInfo] = [] + for start_line, end_line in sections: + start_char = line_starts[start_line - 1] + end_char = line_starts[min(end_line, len(lines))] + section = text[start_char:end_char] + if not section.strip(): + continue + sized = self._fixed_size( + section, + max_tokens=config.max_tokens, + overlap_tokens=config.overlap_tokens, + ) + for piece in sized: + chunks.append( + ChunkInfo( + text=piece.text, + start_char_idx=start_char + piece.start_char_idx, + end_char_idx=start_char + piece.end_char_idx, + ) + ) + return chunks + + def _fixed_size( + self, text: str, *, max_tokens: int, overlap_tokens: int + ) -> list[ChunkInfo]: + window = max(1, max_tokens * self.CHARS_PER_TOKEN) + overlap = min(max(0, overlap_tokens * self.CHARS_PER_TOKEN), window - 1) + step = max(1, window - overlap) + + chunks: list[ChunkInfo] = [] + start = 0 + length = len(text) + while start < length: + end = min(start + window, length) + # Prefer breaking on a newline when not at EOF. + if end < length: + nl = text.rfind("\n", start + 1, end) + if nl > start: + end = nl + 1 + piece = text[start:end] + if piece.strip(): + chunks.append( + ChunkInfo(text=piece, start_char_idx=start, end_char_idx=end) + ) + if end >= length: + break + start = start + step if step > 0 else end + if start >= end: + start = end + return chunks diff --git a/application/utils/harvester/content_hash.py b/application/utils/harvester/content_hash.py new file mode 100644 index 000000000..ed7bd8bda --- /dev/null +++ b/application/utils/harvester/content_hash.py @@ -0,0 +1,13 @@ +import hashlib + + +def generate_content_hash(text: str) -> str: + """ + Generate a deterministic SHA-256 hash for document content. + + Used for artifact-level deduplication. + """ + + return hashlib.sha256( + text.encode("utf-8"), + ).hexdigest() diff --git a/application/utils/harvester/deduplication_metrics.py b/application/utils/harvester/deduplication_metrics.py new file mode 100644 index 000000000..a3843a7fc --- /dev/null +++ b/application/utils/harvester/deduplication_metrics.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass +from .models import DeduplicationStatus + + +@dataclass(slots=True) +class DeduplicationMetrics: + total_artifacts_scanned: int = 0 + + artifacts_new: int = 0 + artifacts_updated: int = 0 + artifacts_unchanged: int = 0 + + artifacts_emitted: int = 0 + artifacts_skipped: int = 0 + + def record(self, status: DeduplicationStatus) -> None: + self.total_artifacts_scanned += 1 + + if status is DeduplicationStatus.NEW: + self.artifacts_new += 1 + self.artifacts_emitted += 1 + + elif status is DeduplicationStatus.UPDATED: + self.artifacts_updated += 1 + self.artifacts_emitted += 1 + + elif status is DeduplicationStatus.UNCHANGED: + self.artifacts_unchanged += 1 + self.artifacts_skipped += 1 diff --git a/application/utils/harvester/document_deduplicator.py b/application/utils/harvester/document_deduplicator.py new file mode 100644 index 000000000..b1e953f26 --- /dev/null +++ b/application/utils/harvester/document_deduplicator.py @@ -0,0 +1,54 @@ +from datetime import datetime + +from .artifact_registry import ArtifactRegistry +from .content_hash import generate_content_hash +from .models import ( + ArtifactRegistryRecord, + DeduplicationStatus, + Document, +) + + +class DocumentDeduplicator: + """ + Performs artifact-level deduplication within a process. + + Documents are classified as NEW, UPDATED, or UNCHANGED. + """ + + def __init__(self, registry: ArtifactRegistry): + self._registry = registry + + def process(self, document: Document) -> DeduplicationStatus: + content_hash = generate_content_hash(document.text) + existing = self._registry.get(document.artifact_id) + now = datetime.now() + + if existing is None: + self._registry.upsert( + ArtifactRegistryRecord( + artifact_id=document.artifact_id, + repository=document.source.repository, + locator_path=document.locator.path, + content_hash=content_hash, + last_commit_sha=document.source.commit_sha, + last_pipeline_run=document.pipeline_run_id, + last_processed_at=now, + status=DeduplicationStatus.NEW.value, + ) + ) + return DeduplicationStatus.NEW + + existing.last_commit_sha = document.source.commit_sha + existing.last_pipeline_run = document.pipeline_run_id + existing.last_processed_at = now + + if existing.content_hash == content_hash: + existing.status = DeduplicationStatus.UNCHANGED.value + self._registry.upsert(existing) + return DeduplicationStatus.UNCHANGED + + existing.content_hash = content_hash + existing.status = DeduplicationStatus.UPDATED.value + self._registry.upsert(existing) + return DeduplicationStatus.UPDATED diff --git a/application/utils/harvester/document_validator.py b/application/utils/harvester/document_validator.py index 9e099a4aa..497e90b15 100644 --- a/application/utils/harvester/document_validator.py +++ b/application/utils/harvester/document_validator.py @@ -9,13 +9,13 @@ class DocumentValidator: """ def validate(self, document: Document) -> bool: - if not document.schema_version: + if not document.schema_version.strip(): return False - if not document.artifact_id.startswith("art:"): + if not self._valid_artifact_id(document.artifact_id): return False - if not document.pipeline_run_id: + if not document.pipeline_run_id.strip(): return False if not document.text: @@ -24,10 +24,13 @@ def validate(self, document: Document) -> bool: if document.source.type != "github": return False - if not document.source.repository: + if not document.source.repository.strip(): return False - if not document.source.commit_sha: + if "/" not in document.source.repository: + return False + + if not document.source.commit_sha.strip(): return False if document.source.committed_at is None: @@ -36,7 +39,28 @@ def validate(self, document: Document) -> bool: if document.locator.kind != "repo_path": return False - if not document.locator.path: + if not document.locator.id.strip(): + return False + + if not document.locator.path.strip(): + return False + + if document.locator.id != document.locator.path: return False return True + + @staticmethod + def _valid_artifact_id(artifact_id: str) -> bool: + # Expected: art:/: with nonempty path. + if not artifact_id.startswith("art:"): + return False + rest = artifact_id[len("art:") :] + if ":" not in rest: + return False + repo, path = rest.split(":", 1) + if not repo.strip() or "/" not in repo: + return False + if not path.strip(): + return False + return True diff --git a/application/utils/harvester/git_repository_client.py b/application/utils/harvester/git_repository_client.py index b650b74f7..06e27d286 100644 --- a/application/utils/harvester/git_repository_client.py +++ b/application/utils/harvester/git_repository_client.py @@ -156,13 +156,14 @@ def checkout(self, reference: str) -> None: ) try: + # Do not insert "--" before the revision: that would treat it as a + # pathspec and leave HEAD unchanged. subprocess.run( [ "git", "-C", str(self.local_path), "checkout", - "--", reference, ], check=True, @@ -285,6 +286,8 @@ def is_valid_repository(self, repository_path: Path) -> bool: def verify_repository_integrity(self) -> bool: return self.is_valid_repository(self.local_path) + MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024 + def get_file_at_commit(self, commit_sha: str, file_path: str) -> str: """ Retrieve the contents of a file at a specific commit. @@ -299,6 +302,34 @@ def get_file_at_commit(self, commit_sha: str, file_path: str) -> str: Returns: File contents as a string. """ + if not commit_sha or commit_sha.startswith("-"): + raise ValueError("Invalid commit SHA") + if not file_path or file_path.startswith("-"): + raise ValueError("Invalid file path") + if "\x00" in file_path: + raise ValueError("Invalid file path") + + # Resolve blob size before loading contents into memory. + size_result = subprocess.run( + [ + "git", + "-C", + str(self.get_local_path()), + "cat-file", + "-s", + f"{commit_sha}:{file_path}", + ], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + size = int(size_result.stdout.strip()) + if size > self.MAX_FILE_SIZE_BYTES: + raise ValueError( + f"File size ({size} bytes) exceeds " + f"maximum supported size ({self.MAX_FILE_SIZE_BYTES} bytes)." + ) result = subprocess.run( [ @@ -306,6 +337,7 @@ def get_file_at_commit(self, commit_sha: str, file_path: str) -> str: "-C", str(self.get_local_path()), "show", + "--end-of-options", f"{commit_sha}:{file_path}", ], capture_output=True, diff --git a/application/utils/harvester/harvest_writer.py b/application/utils/harvester/harvest_writer.py new file mode 100644 index 000000000..b24816dfd --- /dev/null +++ b/application/utils/harvester/harvest_writer.py @@ -0,0 +1,50 @@ +"""Persist Module A ChangeRecord payloads into ``harvest_input``.""" + +from __future__ import annotations + +from typing import Any, Iterable + +from application.database.db import HarvestInput +from application.utils.harvester.chunk_record_validator import ingest_record_to_payload +from application.utils.harvester.models import IngestChunkRecord + + +def write_harvest_input( + session: Any, + pipeline_run_id: str, + records: Iterable[IngestChunkRecord], + *, + dry_run: bool = False, +) -> int: + """ + Insert pending ``harvest_input`` rows for one pipeline run. + + Top-level ``pipeline_run_id`` matches the payload field (Module A contract). + Returns the number of rows that would be / were written. + """ + if not pipeline_run_id or not pipeline_run_id.strip(): + raise ValueError("pipeline_run_id must be non-empty") + + written = 0 + for record in records: + if record.pipeline_run_id != pipeline_run_id: + raise ValueError( + f"record pipeline_run_id {record.pipeline_run_id!r} " + f"!= harvest run {pipeline_run_id!r}" + ) + payload = ingest_record_to_payload(record) + if dry_run: + written += 1 + continue + session.add( + HarvestInput( + pipeline_run_id=pipeline_run_id, + status="pending", + payload=payload, + ) + ) + written += 1 + + if not dry_run and written: + session.commit() + return written diff --git a/application/utils/harvester/heading_extractor.py b/application/utils/harvester/heading_extractor.py index 7941a805b..9b3eef32b 100644 --- a/application/utils/harvester/heading_extractor.py +++ b/application/utils/harvester/heading_extractor.py @@ -1,32 +1,43 @@ from dataclasses import dataclass + from .models import HeadingNode +@dataclass(slots=True) +class _FenceState: + in_fence: bool = False + + class HeadingExtractor: """ - Extracts Markdown headings and their line ranges. + Extracts Markdown ATX headings and their line ranges. Heading ranges extend until the next heading of the same - or higher level, or the end of the document. + or higher level, or the end of the document. Lines inside + fenced code blocks and indented code blocks are ignored. """ def extract(self, text: str) -> list[HeadingNode]: lines = text.splitlines() - headings: list[HeadingNode] = [] + fence = _FenceState() for line_number, line in enumerate(lines, start=1): - stripped = line.lstrip() + if self._toggle_fence(line, fence): + continue + if fence.in_fence: + continue + if self._is_indented_code(line): + continue + stripped = line.lstrip() if not stripped.startswith("#"): continue hashes = len(stripped) - len(stripped.lstrip("#")) - - if hashes == 0: + if hashes == 0 or hashes > 6: continue - - if len(stripped) > hashes and stripped[hashes] != " ": + if len(stripped) <= hashes or stripped[hashes] != " ": continue headings.append( @@ -45,3 +56,17 @@ def extract(self, text: str) -> list[HeadingNode]: break return headings + + @staticmethod + def _toggle_fence(line: str, fence: _FenceState) -> bool: + stripped = line.lstrip() + if stripped.startswith("```") or stripped.startswith("~~~"): + fence.in_fence = not fence.in_fence + return True + return False + + @staticmethod + def _is_indented_code(line: str) -> bool: + if not line.strip(): + return False + return line.startswith(" ") or line.startswith("\t") diff --git a/application/utils/harvester/incremental_pipeline.py b/application/utils/harvester/incremental_pipeline.py new file mode 100644 index 000000000..088e4fe94 --- /dev/null +++ b/application/utils/harvester/incremental_pipeline.py @@ -0,0 +1,99 @@ +from datetime import datetime, timezone +from typing import Any + +from .artifact_registry import ArtifactRegistry +from .checkpoint_store import CheckpointStore +from .deduplication_metrics import DeduplicationMetrics +from .document_deduplicator import DocumentDeduplicator +from .document_validator import DocumentValidator +from .models import ( + DeduplicationStatus, + Document, + RepositoryCheckpoint, +) + + +class IncrementalPipeline: + """ + Coordinates document validation, deduplication, and durable checkpoints. + + Only NEW or UPDATED validated documents are emitted downstream. + Checkpoints are written via ``CheckpointStore`` (Postgres/SQLite). + """ + + def __init__( + self, + deduplicator: DocumentDeduplicator | None = None, + checkpoint_store: CheckpointStore | None = None, + validator: DocumentValidator | None = None, + *, + provider: str = "github", + owner: str = "", + repository_name: str = "", + branch: str = "main", + repository_id: str = "", + ) -> None: + self._deduplicator = deduplicator or DocumentDeduplicator(ArtifactRegistry()) + self._checkpoint_store = checkpoint_store or CheckpointStore() + self._validator = validator or DocumentValidator() + self.metrics = DeduplicationMetrics() + self._provider = provider + self._owner = owner + self._repository_name = repository_name + self._branch = branch + self._repository_id = repository_id + + def process( + self, + repository: str, + pipeline_run_id: str, + documents: list[Document], + *, + last_processed_commit: str | None = None, + ) -> list[Document]: + emitted: list[Document] = [] + metrics = DeduplicationMetrics() + + for document in documents: + if document.pipeline_run_id != pipeline_run_id: + raise ValueError( + f"document pipeline_run_id {document.pipeline_run_id!r} " + f"does not match process run {pipeline_run_id!r}" + ) + if document.source.repository != repository: + raise ValueError( + f"document source.repository {document.source.repository!r} " + f"does not match process repository {repository!r}" + ) + if not self._validator.validate(document): + raise ValueError(f"document failed validation: {document.artifact_id}") + + status = self._deduplicator.process(document) + metrics.record(status) + if status != DeduplicationStatus.UNCHANGED: + emitted.append(document) + + commit_sha = last_processed_commit + if commit_sha is None and documents: + commit_sha = documents[-1].source.commit_sha + if commit_sha: + self._persist_checkpoint(pipeline_run_id, commit_sha) + + self.metrics = metrics + return emitted + + def _persist_checkpoint(self, pipeline_run_id: str, commit_sha: str) -> None: + if not commit_sha.strip(): + raise ValueError("refusing to persist empty last_processed_commit") + repository_id = self._repository_id or f"{self._owner}/{self._repository_name}" + self._checkpoint_store.save( + RepositoryCheckpoint( + repository_id=repository_id, + last_processed_commit=commit_sha, + updated_at=datetime.now(timezone.utc), + provider=self._provider, + owner=self._owner or repository_id.split("/")[0], + repository=self._repository_name or repository_id.split("/", 1)[-1], + branch=self._branch, + ) + ) diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index b30985a0d..b8db2083f 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -1,5 +1,7 @@ from dataclasses import dataclass from datetime import datetime +from enum import Enum + from pydantic import BaseModel @@ -85,3 +87,66 @@ class Document: locator: Locator heading_structure: list[HeadingNode] span: SpanInfo | None = None + + +@dataclass(slots=True) +class ArtifactRegistryRecord: + """ + Tracks the processing state of an artifact. + Used for deduplication within a harvester process. + """ + + artifact_id: str + repository: str + locator_path: str + content_hash: str + last_commit_sha: str + last_pipeline_run: str + last_processed_at: datetime + status: str + + +class DeduplicationStatus(str, Enum): + NEW = "new" + UPDATED = "updated" + UNCHANGED = "unchanged" + + +@dataclass(slots=True) +class CheckpointRecord: + repository: str + pipeline_run_id: str + last_processed_commit: str + status: str + updated_at: datetime + + +@dataclass(slots=True) +class ChunkInfo: + text: str + start_char_idx: int + end_char_idx: int + + +@dataclass(slots=True) +class IngestChunkRecord: + """ + RFC-facing chunk ready to validate as Module B ChangeRecord. + + ``source_repo`` is the ``owner/repo`` string written to ``source.repo``. + ``committed_at`` is an ISO-8601 string (or None only before validation). + """ + + schema_version: str + chunk_id: str + artifact_id: str + pipeline_run_id: str + text: str + span: SpanInfo + source_type: str + source_repo: str + source_commit_sha: str + source_committed_at: str + locator_kind: str + locator_id: str + locator_path: str diff --git a/application/utils/harvester/pipeline.py b/application/utils/harvester/pipeline.py new file mode 100644 index 000000000..61df4c0d1 --- /dev/null +++ b/application/utils/harvester/pipeline.py @@ -0,0 +1,217 @@ +"""Module A entry point: harvest OWASP repos → ``harvest_input``. + +Shape mirrors Module B's ``run_noise_filter``: +``(session, pipeline_run_id, ..., dry_run) -> RunSummary`` with ``to_json()``. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from application.utils.harvester.change_detector import ChangeDetector +from application.utils.harvester.checkpoint_store import CheckpointStore +from application.utils.harvester.chunk_pipeline import DocumentChunkPipeline +from application.utils.harvester.config_loader import load_repo_config +from application.utils.harvester.document_builder import DocumentBuilder +from application.utils.harvester.file_filter import FileFilter +from application.utils.harvester.git_repository_client import GitRepositoryClient +from application.utils.harvester.harvest_writer import write_harvest_input +from application.utils.harvester.incremental_pipeline import IncrementalPipeline +from application.utils.harvester.models import DiffBlock, Document +from application.utils.harvester.repos_validator import validate_repositories +from application.utils.harvester.schemas import RepositoryConfig + +logger = logging.getLogger(__name__) + +DEFAULT_REPOS_YAML = Path(__file__).with_name("repos.yaml") + + +@dataclass +class RunSummary: + """Outcome of one Module A harvest run; the CLI emits this as JSON.""" + + run_id: str + repositories: int = 0 + files_seen: int = 0 + files_retained: int = 0 + documents_emitted: int = 0 + chunks_written: int = 0 + errors: int = 0 + dry_run: bool = False + status: str = "ok" + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + +def run_harvester( + session: Any, + pipeline_run_id: str, + *, + repos_yaml: str | Path | None = None, + dry_run: bool = False, + sync_repos: bool = True, +) -> RunSummary: + """ + Harvest configured repositories and stage chunks in ``harvest_input``. + + For each enabled repo: optionally sync, detect files changed since the + durable checkpoint, build documents, dedupe, chunk, validate as + ChangeRecords, and insert pending ``harvest_input`` rows. + """ + if not pipeline_run_id or not pipeline_run_id.strip(): + raise ValueError("run_harvester needs a non-empty pipeline_run_id") + + run_id = pipeline_run_id.strip() + summary = RunSummary(run_id=run_id, dry_run=dry_run) + + repos_path = Path(repos_yaml) if repos_yaml else DEFAULT_REPOS_YAML + repos_file = load_repo_config(repos_path) + validate_repositories(repos_file.repositories) + + checkpoint_store = CheckpointStore(session=session) + builder = DocumentBuilder() + + for repo_cfg in repos_file.repositories: + if not repo_cfg.enabled: + continue + summary.repositories += 1 + try: + written = _harvest_repository( + session=session, + repo_cfg=repo_cfg, + pipeline_run_id=run_id, + checkpoint_store=checkpoint_store, + builder=builder, + dry_run=dry_run, + sync_repos=sync_repos, + summary=summary, + ) + summary.chunks_written += written + except Exception: + summary.errors += 1 + logger.exception( + "harvester failed for repository %s/%s", + repo_cfg.owner, + repo_cfg.repo, + ) + + if summary.errors and summary.chunks_written == 0: + summary.status = "degraded" + elif summary.errors: + summary.status = "degraded" + return summary + + +def _harvest_repository( + *, + session: Any, + repo_cfg: RepositoryConfig, + pipeline_run_id: str, + checkpoint_store: CheckpointStore, + builder: DocumentBuilder, + dry_run: bool, + sync_repos: bool, + summary: RunSummary, +) -> int: + client = GitRepositoryClient( + owner=repo_cfg.owner, + repository=repo_cfg.repo, + branch=repo_cfg.branch, + ) + if sync_repos: + client.sync() + + head = client.get_current_commit_sha() + checkpoint = checkpoint_store.load(repo_cfg.id) + base = checkpoint.last_processed_commit if checkpoint else None + + detector = ChangeDetector(client) + if base: + modified = detector.get_modified_files_since(base, head) + else: + # First run: treat all tracked files under include paths as candidates + # via an empty-tree diff against HEAD. + modified = detector.get_modified_files_since( + "4b825dc642cb6eb9a060e54bf8d6927bf442cfb4", # git empty tree + head, + ) + + file_filter = FileFilter(exclude_patterns=list(repo_cfg.paths.exclude)) + # Path include globs: keep files matching any include pattern. + from pathspec import PathSpec + + include_spec = PathSpec.from_lines("gitignore", repo_cfg.paths.include) + candidates = [ + path + for path in modified + if include_spec.match_file(path) and path in file_filter.filter_files([path]) + ] + summary.files_seen += len(modified) + summary.files_retained += len(candidates) + + committed_at = _commit_timestamp(client, head) + documents: list[Document] = [] + for path in candidates: + text = client.get_file_at_commit(head, path) + block = DiffBlock( + file_path=path, + added_lines=[], + repository=f"{repo_cfg.owner}/{repo_cfg.repo}", + commit_sha=head, + committed_at=committed_at, + ) + documents.append(builder.build(block, text, pipeline_run_id)) + + incremental = IncrementalPipeline( + checkpoint_store=checkpoint_store, + provider="github", + owner=repo_cfg.owner, + repository_name=repo_cfg.repo, + branch=repo_cfg.branch, + repository_id=repo_cfg.id, + ) + emitted = incremental.process( + repository=f"{repo_cfg.owner}/{repo_cfg.repo}", + pipeline_run_id=pipeline_run_id, + documents=documents, + last_processed_commit=head, + ) + summary.documents_emitted += len(emitted) + + chunk_pipeline = DocumentChunkPipeline(chunking=repo_cfg.chunking) + records = [] + for document in emitted: + records.extend(chunk_pipeline.chunk(document)) + + return write_harvest_input(session, pipeline_run_id, records, dry_run=dry_run) + + +def _commit_timestamp(client: GitRepositoryClient, commit_sha: str) -> datetime: + import subprocess + + result = subprocess.run( + [ + "git", + "-C", + str(client.get_local_path()), + "show", + "-s", + "--format=%cI", + commit_sha, + ], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + raw = result.stdout.strip() + # fromisoformat handles offsets; normalize Z. + if raw.endswith("Z"): + raw = raw[:-1] + "+00:00" + return datetime.fromisoformat(raw).astimezone(timezone.utc) diff --git a/application/utils/oie_orchestrator/__init__.py b/application/utils/oie_orchestrator/__init__.py new file mode 100644 index 000000000..e3dd203b0 --- /dev/null +++ b/application/utils/oie_orchestrator/__init__.py @@ -0,0 +1,15 @@ +"""OIE A→B→C orchestrator package.""" + +from .pipeline import ( + OrchestratorResult, + StageResult, + run_oie_demo_pipeline, + run_oie_pipeline, +) + +__all__ = [ + "OrchestratorResult", + "StageResult", + "run_oie_demo_pipeline", + "run_oie_pipeline", +] diff --git a/application/utils/oie_orchestrator/pipeline.py b/application/utils/oie_orchestrator/pipeline.py new file mode 100644 index 000000000..df2213825 --- /dev/null +++ b/application/utils/oie_orchestrator/pipeline.py @@ -0,0 +1,298 @@ +"""OIE orchestrator — A → B → C for one ``pipeline_run_id``. + +Production sequencing: run each stage, wait for process/library return, +then start the next. Modules communicate only through DB tables: + + A writes ``harvest_input`` → B writes ``knowledge_queue`` → C writes + ``decision_queue`` / stamps ``consumed_at``. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any, Callable, Dict, List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class StageResult: + """Outcome of one orchestrator stage (module A, B, or C).""" + + name: str + status: str # ok | skipped | error + detail: str + summary: Optional[Dict[str, Any]] = None + + +@dataclass +class OrchestratorResult: + """Full A→B→C run summary (JSON-serializable).""" + + run_id: str + dry_run: bool + stages: List[StageResult] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "run_id": self.run_id, + "dry_run": self.dry_run, + "stages": [asdict(s) for s in self.stages], + "ok": all(s.status in ("ok", "skipped", "degraded") for s in self.stages), + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), indent=2) + + +def _summary_dict(summary: Any) -> Dict[str, Any]: + if hasattr(summary, "to_json"): + return json.loads(summary.to_json()) + if hasattr(summary, "__dict__"): + return dict(summary.__dict__) + return {"raw": str(summary)} + + +def _stage_status_from_summary(summary: Any) -> str: + """Map module RunSummary.status to orchestrator stage status. + + Module C currently always reports ``degraded: N decided without the safety + path`` behind ``NullSafetyGuard`` — that is declared, not a hard failure, so + the stage is ``degraded`` (pipeline may continue; Module D must refuse while + unevaluated > 0). Other ``degraded`` values (A/B partial runs, C row errors) + map to ``error`` so ``stop_on_error`` can halt. + """ + raw = getattr(summary, "status", None) + if isinstance(summary, dict): + raw = summary.get("status", raw) + text = str(raw or "ok") + if text == "ok": + return "ok" + if text.startswith("degraded") and "without the safety path" in text: + # Pure safety-path gap, no errored rows mixed in. + if "errored" not in text: + return "degraded" + return "error" + + +def _connect(cache_file: str) -> Any: + from application import sqla + from application.cmd.cre_main import db_connect + + db_connect(cache_file) + return sqla.session + + +def _stage_module_a( + run_id: str, + cache_file: str, + *, + skip: bool, + dry_run: bool, + sync_repos: bool, + run_harvester_fn: Optional[Callable[..., Any]] = None, +) -> StageResult: + if skip: + return StageResult( + name="module_a_harvester", + status="skipped", + detail="skip_a=True; harvester not invoked", + ) + + fn = run_harvester_fn + if fn is None: + from application.utils.harvester.pipeline import run_harvester + + fn = run_harvester + + try: + session = _connect(cache_file) + summary = fn( + session, + run_id, + dry_run=dry_run, + sync_repos=sync_repos, + ) + return StageResult( + name="module_a_harvester", + status=_stage_status_from_summary(summary), + detail=f"run_harvester completed for run_id={run_id!r}", + summary=_summary_dict(summary), + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Module A stage failed") + return StageResult( + name="module_a_harvester", + status="error", + detail=f"run_harvester failed: {exc}", + ) + + +def _stage_module_b( + run_id: str, + cache_file: str, + *, + skip: bool, + dry_run: bool, + run_noise_filter_fn: Optional[Callable[..., Any]] = None, +) -> StageResult: + if skip: + return StageResult( + name="module_b_noise_filter", + status="skipped", + detail="skip_b=True; noise filter not invoked", + ) + + fn = run_noise_filter_fn + if fn is None: + from application.utils.noise_filter.pipeline import run_noise_filter + + fn = run_noise_filter + + try: + session = _connect(cache_file) + summary = fn(session, run_id, dry_run=dry_run) + return StageResult( + name="module_b_noise_filter", + status=_stage_status_from_summary(summary), + detail=f"run_noise_filter completed for run_id={run_id!r}", + summary=_summary_dict(summary), + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Module B stage failed") + return StageResult( + name="module_b_noise_filter", + status="error", + detail=f"run_noise_filter failed: {exc}", + ) + + +def _stage_module_c( + run_id: str, + cache_file: str, + *, + skip: bool, + dry_run: bool, + run_librarian_queue_fn: Optional[Callable[..., Any]] = None, +) -> StageResult: + if skip: + return StageResult( + name="module_c_librarian", + status="skipped", + detail="skip_c=True; librarian not invoked", + ) + + try: + if run_librarian_queue_fn is not None: + # Injected path (tests / hermetic smoke): caller owns session + sink. + summary = run_librarian_queue_fn(run_id, dry_run=dry_run) + else: + from application.cmd.cre_main import db_connect + from application.utils.librarian.config_loader import load_config + from application.utils.librarian.envelope_sink import ( + DbEnvelopeSink, + NullEnvelopeSink, + ) + from application.utils.librarian.factory import build_components + from application.utils.librarian.queue_runner import run_librarian_queue + + cfg = load_config() + database = db_connect(path=cache_file) + components = build_components(database, config=cfg) + sink = ( + NullEnvelopeSink() + if dry_run + else DbEnvelopeSink(database.session, run_id) + ) + summary = run_librarian_queue( + database.session, + run_id, + components, + cfg, + at=datetime.now(timezone.utc), + sink=sink, + dry_run=dry_run, + ) + + return StageResult( + name="module_c_librarian", + status=_stage_status_from_summary(summary), + detail=f"run_librarian_queue completed for run_id={run_id!r}", + summary=( + _summary_dict(summary) if not isinstance(summary, dict) else summary + ), + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Module C stage failed") + return StageResult( + name="module_c_librarian", + status="error", + detail=f"run_librarian_queue failed: {exc}", + ) + + +def run_oie_pipeline( + *, + cache_file: str, + pipeline_run_id: Optional[str] = None, + skip_a: bool = False, + skip_b: bool = False, + skip_c: bool = False, + dry_run: bool = False, + sync_repos: bool = True, + stop_on_error: bool = True, + run_harvester_fn: Optional[Callable[..., Any]] = None, + run_noise_filter_fn: Optional[Callable[..., Any]] = None, + run_librarian_queue_fn: Optional[Callable[..., Any]] = None, +) -> OrchestratorResult: + """ + Run A→B→C for one ``pipeline_run_id``. + + Defaults run all stages for real (not dry-run). Inject callables in tests. + When ``stop_on_error`` is True (default), later stages are skipped after + an earlier stage returns ``error``. + """ + run_id = (pipeline_run_id or "").strip() or ( + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ) + result = OrchestratorResult(run_id=run_id, dry_run=dry_run) + + a = _stage_module_a( + run_id, + cache_file, + skip=skip_a, + dry_run=dry_run, + sync_repos=sync_repos, + run_harvester_fn=run_harvester_fn, + ) + result.stages.append(a) + if stop_on_error and a.status == "error": + return result + + b = _stage_module_b( + run_id, + cache_file, + skip=skip_b, + dry_run=dry_run, + run_noise_filter_fn=run_noise_filter_fn, + ) + result.stages.append(b) + if stop_on_error and b.status == "error": + return result + + c = _stage_module_c( + run_id, + cache_file, + skip=skip_c, + dry_run=dry_run, + run_librarian_queue_fn=run_librarian_queue_fn, + ) + result.stages.append(c) + return result + + +# Back-compat alias used by the draft PoC script name. +run_oie_demo_pipeline = run_oie_pipeline diff --git a/cre.py b/cre.py index 4dff20fc4..d34a44540 100644 --- a/cre.py +++ b/cre.py @@ -314,23 +314,41 @@ def main() -> None: action="store_true", help="run Module B noise/relevance filter over a harvest run's chunks", ) + parser.add_argument( + "--run_harvester", + action="store_true", + help="run Module A harvester; writes ChangeRecords into harvest_input", + ) parser.add_argument( "--run_id", default="", - help="pipeline_run_id to process (required with --run_noise_filter; " - "with --run_librarian, selects the live knowledge_queue path)", + help="pipeline_run_id to process (required with --run_harvester / " + "--run_noise_filter; with --run_librarian, selects the live " + "knowledge_queue path)", ) parser.add_argument( "--noise_filter_dry_run", action="store_true", help="classify without writing to knowledge_queue or marking rows processed", ) + parser.add_argument( + "--harvester_dry_run", + action="store_true", + help="run Module A without writing harvest_input rows", + ) + parser.add_argument( + "--harvester_repos_yaml", + default="", + help="optional path to repos.yaml for --run_harvester", + ) args = parser.parse_args() if args.export and not args.csv: parser.error("--export requires --csv ") if args.run_noise_filter and not args.run_id.strip(): parser.error("--run_noise_filter requires --run_id ") + if args.run_harvester and not args.run_id.strip(): + parser.error("--run_harvester requires --run_id ") # The live queue path takes its rows from the DB, so a fixture path would be # silently ignored rather than doing what the caller plainly asked for. if args.librarian_source and args.run_id.strip(): diff --git a/docs/gsoc_2026_module_a/blockers.md b/docs/gsoc_2026_module_a/blockers.md new file mode 100644 index 000000000..fb4d32cfb --- /dev/null +++ b/docs/gsoc_2026_module_a/blockers.md @@ -0,0 +1,77 @@ +# Module A stack blockers (#1029 → #1038 → #1044) + +**Audience:** maintainers finishing Module A after merging the GSoC week 6–8 stack. +**Status:** open as of 2026-08-29 review. +**Goal after fixes:** orchestrator can `run A → wait exit → B reads harvest_input → C`. + +None of these PRs are merge-ready as an A *stage*. CI is green; the gaps are product/contract, not test failures. + +--- + +## Shared (blocks e2e for all three) + +| # | Blocker | Why it matters | +|---|---------|----------------| +| S1 | Nothing writes `harvest_input` | B has nowhere to read; A→B handoff missing | +| S2 | No `run_harvester` / `cre.py --run_harvester --run_id` | Orchestrator cannot start A or wait on exit | +| S3 | Chunk emit is not a Module B `ChangeRecord` | B validates with Pydantic and will mark rows `error` | + +Until S1–S3 land, seed `harvest_input` manually (as draft orchestrator #996 already does). + +--- + +## #1029 — week 6 (documents / artifacts) + +| # | Blocker | Location | +|---|---------|----------| +| B1 | `git checkout -- ` treats ref as **pathspec**, does not switch commits | `git_repository_client.py` `checkout()` | +| B2 | `git show {commit}:{path}` lacks `--` / `--end-of-options`; leading-dash paths are argv injection | `get_file_at_commit()` | +| B3 | Heading extractor treats indented / fenced `#` lines as headings | `heading_extractor.py` | +| B4 | Validator accepts `artifact_id="art:"`; never checks `locator.id` | `document_validator.py` | +| B5 | Internal `source.repository` vs contract `source.repo` | Must rename at persist boundary | + +Week 6 is an acceptable *library* once B1–B4 are fixed. It is not an A stage alone. + +--- + +## #1038 — week 7 (dedup / checkpoints) + +| # | Blocker | Location | +|---|---------|----------| +| B6 | `ArtifactRegistry` and `CheckpointManager` are **in-memory dicts** | Die on process exit; ignore `CheckpointStore` / `harvester_checkpoint` already on `main` | +| B7 | `IncrementalPipeline` saves `last_processed_commit=""` then updates per doc | Crash mid-run leaves useless / dangerous checkpoint | +| B8 | UNCHANGED path does not refresh `last_commit_sha` / `last_pipeline_run` | Stale registry metadata | +| B9 | `DocumentValidator` never called from `process()` | Invalid docs can emit | + +--- + +## #1044 — week 8 (chunking tip) + +| # | Blocker | Location | +|---|---------|----------| +| B10 | `IngestChunkRecord` drops `pipeline_run_id`, `source`, `locator` | `chunk_record_builder.py` / `models.py` — B will reject | +| B11 | `llama-index-*` + `textacy` added to **prod** `requirements.txt` | Violates Heroku slug guard (torch / sentence-transformers off prod) | +| B12 | `repos.yaml` chunking config ignored | Config says `markdown_heading` + tokens; code always uses LlamaIndex semantic splitter | +| B13 | Semantic chunks can cross heading boundaries | Wrong `heading_path` for B’s LLM prompt | +| B14 | Negative `start_char_idx` not rejected | `chunk_record_validator.py` | + +--- + +## Minimum finish checklist (post-merge “mod a nits”) + +1. Fix B1–B4 (git argv + headings + validator). +2. Persist checkpoints via existing `CheckpointStore` / `harvester_checkpoint` (B6–B7). +3. Emit full `ChangeRecord`; validate with `application.utils.noise_filter.schemas.ChangeRecord` (B5, B10, B14). +4. Prefer `markdown_heading` / fixed-size chunking from `repos.yaml`; keep LlamaIndex **off** prod requirements (B11–B13). +5. Add `run_harvester(session, pipeline_run_id, …) -> RunSummary` and `cre.py --run_harvester --run_id`. +6. INSERT `harvest_input` rows: top-level `pipeline_run_id` == payload `pipeline_run_id`, `status=pending`. +7. Wire orchestrator to call A for real (not `todo` / skip-by-default). + +--- + +## Out of scope for the nits PR (optional later) + +- RSS / `feed_item` emission +- WSTG / SAMM / Top10 in `repos.yaml` (expand sources) +- Nightly GitHub Action (orchestrator or cron can own schedule) +- PDF / HTML extractors diff --git a/docs/gsoc_2026_module_a/line-by-line.md b/docs/gsoc_2026_module_a/line-by-line.md new file mode 100644 index 000000000..de6a07555 --- /dev/null +++ b/docs/gsoc_2026_module_a/line-by-line.md @@ -0,0 +1,170 @@ +# Module A stack — line-by-line review notes + +**PRs:** [#1029](https://github.com/OWASP/OpenCRE/pull/1029) → [#1038](https://github.com/OWASP/OpenCRE/pull/1038) → [#1044](https://github.com/OWASP/OpenCRE/pull/1044) +**Tip commit reviewed:** `week_8-chunking-retrieval` @ `6affe36` +**Companion:** `blockers.md` (severity only). This file is file-level ask / keep / change. + +--- + +## `application/utils/harvester/git_repository_client.py` + +| Lines / area | Finding | Action | +|--------------|---------|--------| +| `checkout()` uses `git checkout -- ` | `--` makes Git treat the argument as a **pathspec**, not a branch/commit. Checkout does not switch HEADs. | **Fix:** `["git", "-C", path, "checkout", reference]` after leading-dash reject. Add test that branch/commit switches. | +| `get_file_at_commit()` `git show f"{commit}:{path}"` | No `--end-of-options`. Path `--output=…` becomes a Git option (injection). | **Fix:** `git show --end-of-options f"{sha}:{path}"` or pass `--` and validate path. Reject leading `-` on path. Size-limit before loading huge blobs. | +| `clone` / `fetch` / `sync` | Solid atomic clone + lock pattern already on main. | **Keep.** | +| `get_file_at_commit` timeout=30 | Inconsistent with 300 elsewhere; OK for file read. | **Keep** (nit: document). | + +--- + +## `application/utils/harvester/heading_extractor.py` + +| Lines / area | Finding | Action | +|--------------|---------|--------| +| `stripped = line.lstrip()` then `startswith("#")` | Indented code and fenced blocks can look like headings. | **Fix:** track fence state; skip lines with ≥4 leading spaces; ignore `#` inside fences. | +| Range extend until same/higher level | Correct for ATX headings outside code. | **Keep** once fence/indent fixed. | +| No setext (`===` / `---`) support | Acceptable for OWASP md corpus. | **Out of scope.** | + +--- + +## `application/utils/harvester/artifact_id.py` + +| Finding | Action | +|---------|--------| +| `art:{repository}:{file_path}` matches contract mock shape | **Keep.** | +| Empty repo/path → `art:` or `art::` | Reject in validator (see below). | + +--- + +## `application/utils/harvester/document_builder.py` + +| Finding | Action | +|---------|--------| +| Builds from **full file text** at commit, not only added lines | **Keep** — correct for B. | +| `SourceInfo.repository` not `repo` | Rename at ChangeRecord emit (`source.repo`). | +| `SCHEMA_VERSION = "0.2.0"` | Align with contract pin; B does not version-gate. **Keep.** | +| `committed_at` as `datetime` on Document | Serialize ISO-8601 string for B. | + +--- + +## `application/utils/harvester/document_validator.py` + +| Finding | Action | +|---------|--------| +| `artifact_id.startswith("art:")` accepts `"art:"` | Require nonempty repo + path after prefix (split on `:`, ≥3 parts with nonempty). | +| No `locator.id` check | Require `locator.id` and `id == path` for `repo_path`. | +| Returns `bool` instead of raising | Prefer raise (like chunk validator) or keep bool but call from pipeline. | + +--- + +## `application/utils/harvester/document_deduplicator.py` + +| Finding | Action | +|---------|--------| +| NEW / UPDATED / UNCHANGED on content hash | **Keep** idea. | +| UNCHANGED: only flips status, skips commit/run metadata | **Fix:** always update `last_commit_sha`, `last_pipeline_run`, `last_processed_at`. | +| Registry is caller-injected | Wire to durable store or accept ephemeral only with DB checkpoint as source of truth. | + +--- + +## `application/utils/harvester/artifact_registry.py` / `checkpoint_manager.py` + +| Finding | Action | +|---------|--------| +| Pure in-memory `dict` | **Replace** checkpoint path with `CheckpointStore` → `harvester_checkpoint`. Artifact registry may stay in-process for a single run; must not be the only incremental memory. | +| `IncrementalPipeline` saves `last_processed_commit=""` | **Fix:** never persist empty SHA; write real SHA only after successful processing; use existing store. | + +--- + +## `application/utils/harvester/incremental_pipeline.py` + +| Finding | Action | +|---------|--------| +| Emits NEW/UPDATED only | **Keep.** | +| No document validation before emit | Call `DocumentValidator`. | +| No mismatch check: `document.source.repository` / `pipeline_run_id` vs args | Reject mismatched docs before checkpoint write. | +| Does not write `harvest_input` | Not this file’s job alone — need top-level runner. | + +--- + +## `application/utils/harvester/chunker.py` + +| Finding | Action | +|---------|--------| +| LlamaIndex `SemanticSplitterNodeParser` + HF embed | **Do not ship on prod `requirements.txt`.** Prefer `markdown_heading` / fixed-size from `repos.yaml`. Optional: keep semantic behind `requirements-dev` + env flag, off by default. | +| Empty text → `[]` | **Keep.** | +| Ignores `ChunkingConfig.max_tokens` / `overlap_tokens` / `strategy` | Honor YAML. | + +--- + +## `application/utils/harvester/chunk_record_builder.py` + +| Finding | Action | +|---------|--------| +| Builds span index/total/heading_path/offsets | **Keep** structure. | +| `IngestChunkRecord` omits `pipeline_run_id`, `source`, `locator` | **Fix:** emit full ChangeRecord-shaped payload (copy from Document). | +| `chunk_id` embeds heading + hash | Contract mock uses `chk:art:…:idx`. Either is fine if stable; prefer index-based `chk:{artifact_id}:{index}` for B fixture parity, or keep hash form but document. | +| Heading path from start line only | After heading-aware chunking, OK; with semantic cross-heading splits, wrong — fix chunker first. | + +--- + +## `application/utils/harvester/chunk_record_validator.py` + +| Finding | Action | +|---------|--------| +| Checks index/total/order | **Keep.** | +| Missing negative char offset reject | **Fix:** `start_char_idx >= 0`, `end_char_idx >= 0`. | +| Does not validate source/locator (because absent) | After B10 fix, validate or delegate to `ChangeRecord.model_validate`. | + +--- + +## `application/utils/harvester/chunk_pipeline.py` + +| Finding | Action | +|---------|--------| +| chunker → builder → validate each | **Keep** shape. | +| No provenance on records | Fixed in builder. | +| No DB write | Top-level `run_harvester` owns that. | + +--- + +## `application/utils/harvester/models.py` (week 8 tip) + +| Finding | Action | +|---------|--------| +| `Document` has provenance; `IngestChunkRecord` does not | Align chunk model with B `ChangeRecord` fields. | +| `SourceInfo.repository` | Emit as `repo` in JSON. | +| `DeduplicationStatus` / registry records | Fine as internal. | + +--- + +## `requirements.txt` / `requirements-dev.txt` + +| Finding | Action | +|---------|--------| +| Tip adds `llama-index-core`, `llama-index-embeddings-huggingface`, `textacy` to **prod** | **Revert from prod.** Dev-only if kept at all. Never pull sentence-transformers onto Heroku slug via HF embed path. | + +--- + +## Tests + +| Finding | Action | +|---------|--------| +| Broad unit coverage for builders/chunkers | **Keep** and extend. | +| Missing: ChangeRecord round-trip, `harvest_input` insert, CLI, git checkout switch, fence headings, oversized `git show` | **Add** in nits PR. | +| Diff pipeline benchmark env-gated | **Keep.** | + +--- + +## What “done” looks like for Module A (acceptance) + +```bash +python cre.py --run_harvester --run_id --cache_file +# exit 0, JSON summary on stdout +# harvest_input has pending rows for R whose payload passes ChangeRecord + +python cre.py --run_noise_filter --run_id --cache_file +python cre.py --run_librarian --run_id --cache_file +``` + +Orchestrator sequences those three and waits on each exit code. diff --git a/docs/gsoc_2026_module_a/runbook.md b/docs/gsoc_2026_module_a/runbook.md new file mode 100644 index 000000000..e648c67c5 --- /dev/null +++ b/docs/gsoc_2026_module_a/runbook.md @@ -0,0 +1,54 @@ +# Module A — Harvester runbook + +**Audience:** operators / OIE orchestrator. **Status:** v1 (2026-08-29). + +Module A is a **stateless batch step**: the orchestrator invokes it once per +`pipeline_run_id`; it syncs configured OWASP repos, chunks changed markdown, +validates each chunk as a Module B `ChangeRecord`, inserts `harvest_input` +rows (`status=pending`), updates durable `harvester_checkpoint`s, and exits +with a JSON summary. + +--- + +## Invoke + +```bash +python cre.py --run_harvester --run_id --cache_file +``` + +Optional: + +- `--harvester_dry_run` — classify path without inserting rows +- `--harvester_repos_yaml PATH` — override `application/utils/harvester/repos.yaml` + +Orchestrated (A → B → C): + +```bash +make oie-pipeline OIE_ARGS='--run_id 20260829T020000Z' +# or +PYTHONPATH=. python scripts/run_oie_pipeline.py --cache_file --run_id +``` + +Hermetic A→B→C smoke (no git sync, no LLM / embedding API): + +```bash +make oie-e2e-smoke +``` + +Live notes: + +- Module B needs an LLM classifier (Vertex / configured provider) unless you inject one. +- Module C needs embeddings + cross-encoder (`requirements-dev.txt`) unless you inject stubs. +- Use `--skip-c` / `--dry-run` / `--no-sync-repos` as needed while bootstrapping. + +--- + +## Guarantees + +- Payload `pipeline_run_id` equals the top-level `harvest_input.pipeline_run_id`. +- Every written payload validates as `application.utils.noise_filter.schemas.ChangeRecord`. +- Chunking follows `repos.yaml` (`markdown_heading` / `fixed_size`); no LlamaIndex on the prod slug. +- Repo-level errors are isolated; a partial run returns `status=degraded` (exit 1). + +See also: `blockers.md`, `line-by-line.md`, and +`docs/gsoc_2026_module_b/module_a_contract.md`. diff --git a/scripts/run_oie_e2e_smoke.py b/scripts/run_oie_e2e_smoke.py new file mode 100644 index 000000000..09ee451e0 --- /dev/null +++ b/scripts/run_oie_e2e_smoke.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Hermetic OIE A→B→C smoke (no network, no LLM / embedding API). + +Seeds a ChangeRecord into harvest_input the way Module A would, runs the +orchestrator with an injected always-KNOWLEDGE Module B classifier and stub +Module C retriever/reranker/scaler, and asserts knowledge_queue was drained +into decision_queue with consumed_at stamped. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from datetime import datetime, timezone +from pathlib import Path + + +def main() -> int: + os.environ.setdefault("FLASK_CONFIG", "development") + os.environ.setdefault("NO_LOAD_GRAPH_DB", "1") + + from application import sqla + from application.cmd.cre_main import db_connect + from application.database.db import ( + DecisionQueueItem, + HarvestInput, + KnowledgeQueueItem, + ) + from application.utils.harvester.harvest_writer import write_harvest_input + from application.utils.harvester.models import IngestChunkRecord, SpanInfo + from application.utils.harvester.pipeline import RunSummary + from application.utils.librarian.config_loader import LibrarianConfig + from application.utils.librarian.envelope_sink import DbEnvelopeSink + from application.utils.librarian.factory import LibrarianComponents + from application.utils.librarian.queue_runner import run_librarian_queue + from application.utils.librarian.schemas import CreCandidate, RetrievalAudit + from application.utils.noise_filter.pipeline import run_noise_filter + from application.utils.noise_filter.schemas import ClassifyResult + from application.utils.oie_orchestrator import run_oie_pipeline + + run_id = "smoke-20260829T000000Z" + at = datetime(2026, 8, 29, tzinfo=timezone.utc) + + class _FakeClassifier: + def classify_batch(self, records): + return [ + ClassifyResult(label="KNOWLEDGE", confidence=0.95, reasoning="smoke") + for _ in records + ] + + class _Retriever: + def retrieve(self, text: str) -> RetrievalAudit: + return RetrievalAudit( + retriever="stub/1.0.0", + candidates=[CreCandidate(cre_id="616-305", score_vector=0.9)], + reranked=[], + threshold=0.0, + ) + + class _Reranker: + def rerank(self, text: str, audit: RetrievalAudit) -> RetrievalAudit: + return audit.model_copy( + update={ + "reranked": [ + CreCandidate(cre_id="616-305", score_rerank=20.0), + ] + } + ) + + class _Scaler: + def confidence(self, logits) -> float: + return 0.95 + + with tempfile.TemporaryDirectory() as tmp: + cache_db = f"sqlite:///{Path(tmp) / 'smoke.sqlite'}" + db_connect(cache_db) + sqla.create_all() + + record = IngestChunkRecord( + schema_version="0.2.0", + chunk_id="chk:art:OWASP/ASVS:4.0/en/auth.md:0", + artifact_id="art:OWASP/ASVS:4.0/en/auth.md", + pipeline_run_id=run_id, + text="Use MFA for all admin accounts.", + span=SpanInfo( + heading_path=["Authentication"], + start_line=3, + end_line=3, + index=0, + total=1, + start_char_idx=0, + end_char_idx=31, + ), + source_type="github", + source_repo="OWASP/ASVS", + source_commit_sha="abc1234deadbeef", + source_committed_at="2026-08-29T00:00:00Z", + locator_kind="repo_path", + locator_id="4.0/en/auth.md", + locator_path="4.0/en/auth.md", + ) + write_harvest_input(sqla.session, run_id, [record]) + + def run_a(session, pipeline_run_id, **kwargs): + return RunSummary( + run_id=pipeline_run_id, + repositories=1, + chunks_written=1, + status="ok", + ) + + def run_b(session, pipeline_run_id, **kwargs): + return run_noise_filter( + session, + pipeline_run_id, + classifier=_FakeClassifier(), + dry_run=False, + ) + + def run_c(pipeline_run_id, **kwargs): + cfg = LibrarianConfig( + crossencoder_model="stub", + retriever_backend="in_memory", + top_k_retrieval=20, + top_k_rerank=5, + link_threshold=0.80, + temperature=1.0, + batch_size=32, + ece_target=0.10, + conformal_alpha=0.10, + ) + components = LibrarianComponents( + retriever=_Retriever(), + reranker=_Reranker(), + scaler=_Scaler(), + known_cre_ids=frozenset({"616-305"}), + ) + return run_librarian_queue( + sqla.session, + pipeline_run_id, + components, + cfg, + at=at, + sink=DbEnvelopeSink(sqla.session, pipeline_run_id), + dry_run=False, + ) + + result = run_oie_pipeline( + cache_file=cache_db, + pipeline_run_id=run_id, + dry_run=False, + sync_repos=False, + run_harvester_fn=run_a, + run_noise_filter_fn=run_b, + run_librarian_queue_fn=run_c, + ) + pending = ( + sqla.session.query(HarvestInput) + .filter_by(pipeline_run_id=run_id, status="pending") + .count() + ) + processed = ( + sqla.session.query(HarvestInput) + .filter_by(pipeline_run_id=run_id, status="processed") + .count() + ) + queued = ( + sqla.session.query(KnowledgeQueueItem) + .filter_by(pipeline_run_id=run_id) + .count() + ) + consumed = ( + sqla.session.query(KnowledgeQueueItem) + .filter( + KnowledgeQueueItem.pipeline_run_id == run_id, + KnowledgeQueueItem.consumed_at.isnot(None), + ) + .count() + ) + decisions = ( + sqla.session.query(DecisionQueueItem) + .filter_by(pipeline_run_id=run_id) + .count() + ) + + out = { + "orchestrator_ok": result.to_dict()["ok"], + "harvest_pending": pending, + "harvest_processed": processed, + "knowledge_queue_rows": queued, + "knowledge_consumed": consumed, + "decision_queue_rows": decisions, + "stages": result.to_dict()["stages"], + } + print(json.dumps(out, indent=2)) + ok = ( + out["orchestrator_ok"] + and processed >= 1 + and queued >= 1 + and consumed >= 1 + and decisions >= 1 + ) + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/run_oie_pipeline.py b/scripts/run_oie_pipeline.py new file mode 100644 index 000000000..e58a566fb --- /dev/null +++ b/scripts/run_oie_pipeline.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +"""Run the OIE A→B→C pipeline for one pipeline_run_id.""" + +from __future__ import annotations + +import argparse +import os +import sys + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Orchestrate Module A → B → C for one pipeline_run_id" + ) + parser.add_argument( + "--cache_file", + default=os.environ.get("DATABASE_URL") + or os.environ.get("DEV_DATABASE_URL") + or "sqlite:///", + help="SQLAlchemy DB URL (default: DATABASE_URL / DEV_DATABASE_URL / memory)", + ) + parser.add_argument( + "--run_id", + default="", + help="pipeline_run_id (default: UTC timestamp)", + ) + parser.add_argument("--skip-a", action="store_true") + parser.add_argument("--skip-b", action="store_true") + parser.add_argument("--skip-c", action="store_true") + parser.add_argument( + "--dry-run", + action="store_true", + help="run stages without persisting queue writes where supported", + ) + parser.add_argument( + "--no-sync-repos", + action="store_true", + help="skip git clone/fetch during Module A (use local cache as-is)", + ) + parser.add_argument( + "--continue-on-error", + action="store_true", + help="run later stages even if an earlier stage errors", + ) + args = parser.parse_args() + + # db_connect (inside each stage) creates + pushes the Flask app context. + # Do not nest an extra app_context here — that pops the wrong stack frame. + os.environ.setdefault("FLASK_CONFIG", "development") + os.environ.setdefault("NO_LOAD_GRAPH_DB", "1") + + from application.utils.oie_orchestrator import run_oie_pipeline + + result = run_oie_pipeline( + cache_file=args.cache_file, + pipeline_run_id=args.run_id or None, + skip_a=args.skip_a, + skip_b=args.skip_b, + skip_c=args.skip_c, + dry_run=args.dry_run, + sync_repos=not args.no_sync_repos, + stop_on_error=not args.continue_on_error, + ) + print(result.to_json()) + return 0 if result.to_dict()["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main())