From aa79243a3b39eb17a60a92f242ec560ba210e0cf Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 13:47:17 +0530 Subject: [PATCH 01/27] feat(harvester): add git diff retrieval pipeline --- application/tests/harvester_test/diff_retriever_test.py | 2 +- application/utils/harvester/diff_retriever.py | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/application/tests/harvester_test/diff_retriever_test.py b/application/tests/harvester_test/diff_retriever_test.py index 502ac2d39..3c0c703b2 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, diff --git a/application/utils/harvester/diff_retriever.py b/application/utils/harvester/diff_retriever.py index 7efd45560..29814ea2a 100644 --- a/application/utils/harvester/diff_retriever.py +++ b/application/utils/harvester/diff_retriever.py @@ -8,7 +8,6 @@ class DiffRetriever: """ - Retrieves unified git diffs between two commits. This class is responsible only for retrieving raw diff text. From 15a0ba3cc1b31faaeac062105e8bfd7636aff432 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 14:00:26 +0530 Subject: [PATCH 02/27] feat(harvester): parse unified git diffs --- application/tests/harvester_test/diff_parser_test.py | 1 + 1 file changed, 1 insertion(+) diff --git a/application/tests/harvester_test/diff_parser_test.py b/application/tests/harvester_test/diff_parser_test.py index a4444e8d9..48f9d954d 100644 --- a/application/tests/harvester_test/diff_parser_test.py +++ b/application/tests/harvester_test/diff_parser_test.py @@ -5,6 +5,7 @@ DiffParser, ) + TEST_REPOSITORY = "OWASP/ASVS" TEST_COMMIT_SHA = "abc123" TEST_COMMITTED_AT = datetime.now(UTC) From 20f05158bf7c95632422a0e8a0e588758366ee15 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 14:45:58 +0530 Subject: [PATCH 03/27] feat(harvester): normalize extracted diff content --- application/utils/harvester/diff_normalizer.py | 17 +++-------------- 1 file changed, 3 insertions(+), 14 deletions(-) diff --git a/application/utils/harvester/diff_normalizer.py b/application/utils/harvester/diff_normalizer.py index 756d89147..a227a7044 100644 --- a/application/utils/harvester/diff_normalizer.py +++ b/application/utils/harvester/diff_normalizer.py @@ -1,26 +1,15 @@ -import re -import unicodedata +import textacy.preprocessing as prep from .models import DiffBlock class DiffNormalizer: - """ - Normalizes extracted diff content. - - Whitespace is collapsed, Unicode normalized, - and empty lines removed. - """ - def normalize_line(self, line: str) -> str: - line = unicodedata.normalize("NFKC", line) - line = re.sub(r"\s+", " ", line) + line = prep.normalize.unicode(line) + line = prep.normalize.whitespace(line) return line.strip() def normalize(self, blocks: list[DiffBlock]) -> list[DiffBlock]: - """ - Normalize every added line in each DiffBlock. - """ normalized: list[DiffBlock] = [] for block in blocks: From f3bcb063b865f05b9344f4c6924dd173216a4afa Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 17:14:07 +0530 Subject: [PATCH 04/27] Enhance diff pipeline with metadata and normalization --- application/tests/harvester_test/diff_parser_test.py | 4 ++++ .../tests/harvester_test/diff_pipeline_test.py | 8 -------- .../tests/harvester_test/diff_retriever_test.py | 4 ---- application/utils/harvester/diff_normalizer.py | 11 +++++++++++ application/utils/harvester/diff_retriever.py | 6 ------ 5 files changed, 15 insertions(+), 18 deletions(-) diff --git a/application/tests/harvester_test/diff_parser_test.py b/application/tests/harvester_test/diff_parser_test.py index 48f9d954d..877d70a09 100644 --- a/application/tests/harvester_test/diff_parser_test.py +++ b/application/tests/harvester_test/diff_parser_test.py @@ -5,6 +5,10 @@ DiffParser, ) +TEST_REPOSITORY = "OWASP/ASVS" +TEST_COMMIT_SHA = "abc123" +TEST_COMMITTED_AT = datetime.now(UTC) + TEST_REPOSITORY = "OWASP/ASVS" TEST_COMMIT_SHA = "abc123" diff --git a/application/tests/harvester_test/diff_pipeline_test.py b/application/tests/harvester_test/diff_pipeline_test.py index 07160f170..37c7f25a9 100644 --- a/application/tests/harvester_test/diff_pipeline_test.py +++ b/application/tests/harvester_test/diff_pipeline_test.py @@ -19,7 +19,6 @@ class DiffPipelineBenchmark(unittest.TestCase): """ def test_pipeline_benchmark(self): - if os.getenv("OPENCRE_RUN_NETWORK_TESTS") != "1": self.skipTest("Network benchmark disabled") @@ -29,9 +28,7 @@ def test_pipeline_benchmark(self): "master", ) client.sync() - head_commit = client.get_current_commit_sha() - previous_commit = subprocess.run( [ "git", @@ -51,23 +48,18 @@ def test_pipeline_benchmark(self): normalizer = DiffNormalizer() start = time.perf_counter() - diff = retriever.get_diff( previous_commit, head_commit, ) - blocks = parser.parse( diff, repository="OWASP/ASVS", commit_sha=head_commit, committed_at=datetime.now(UTC), ) - normalizer.normalize(blocks) - elapsed = time.perf_counter() - start print(f"\nPipeline took {elapsed:.3f}s") - self.assertLess(elapsed, 5) diff --git a/application/tests/harvester_test/diff_retriever_test.py b/application/tests/harvester_test/diff_retriever_test.py index 3c0c703b2..f8c3ea142 100644 --- a/application/tests/harvester_test/diff_retriever_test.py +++ b/application/tests/harvester_test/diff_retriever_test.py @@ -16,10 +16,8 @@ def test_get_diff(self, mock_run): MagicMock(stdout="def456\n"), MagicMock(stdout=b"diff --git a/README.md b/README.md\n"), ] - client = MagicMock() client.get_local_path.return_value = "/tmp/repo" - retriever = DiffRetriever(client) diff = retriever.get_diff( @@ -31,7 +29,6 @@ def test_get_diff(self, mock_run): diff, "diff --git a/README.md b/README.md\n", ) - mock_run.assert_has_calls( [ call( @@ -88,7 +85,6 @@ def test_large_diff_raises(self, mock_run): client = MagicMock() client.get_local_path.return_value = "/tmp/repo" - retriever = DiffRetriever(client) with self.assertRaises(ValueError): diff --git a/application/utils/harvester/diff_normalizer.py b/application/utils/harvester/diff_normalizer.py index a227a7044..fe8773349 100644 --- a/application/utils/harvester/diff_normalizer.py +++ b/application/utils/harvester/diff_normalizer.py @@ -1,15 +1,26 @@ import textacy.preprocessing as prep +from application.utils.harvester import repository_client from .models import DiffBlock class DiffNormalizer: + """ + Normalizes extracted diff content. + + Whitespace is collapsed, Unicode normalized, + and empty lines removed. + """ + def normalize_line(self, line: str) -> str: line = prep.normalize.unicode(line) line = prep.normalize.whitespace(line) return line.strip() def normalize(self, blocks: list[DiffBlock]) -> list[DiffBlock]: + """ + Normalize every added line in each DiffBlock. + """ normalized: list[DiffBlock] = [] for block in blocks: diff --git a/application/utils/harvester/diff_retriever.py b/application/utils/harvester/diff_retriever.py index 29814ea2a..665d10d19 100644 --- a/application/utils/harvester/diff_retriever.py +++ b/application/utils/harvester/diff_retriever.py @@ -11,9 +11,7 @@ class DiffRetriever: Retrieves unified git diffs between two commits. This class is responsible only for retrieving raw diff text. - Parsing and normalization are handled by downstream components. - """ MAX_DIFF_SIZE_BYTES = 50 * 1024 * 1024 @@ -34,7 +32,6 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: Raises: subprocess.CalledProcessError: If git diff fails. - ValueError: If the diff exceeds the configured size limit. """ @@ -43,7 +40,6 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: base_commit, target_commit, ) - base_commit = self._resolve_commit(base_commit) target_commit = self._resolve_commit(target_commit) @@ -69,9 +65,7 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: raise diff_bytes = result.stdout - diff_size = len(diff_bytes) - if diff_size > self.MAX_DIFF_SIZE_BYTES: raise ValueError( f"Diff size ({diff_size} bytes) exceeds " From a6299e2d3a455679fd1a3095065d1c6da68ddf27 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Tue, 14 Jul 2026 18:34:02 +0530 Subject: [PATCH 05/27] feat(harvester): add RFC document data models and artifact.py --- application/utils/harvester/artifact_id.py | 11 ++++++ application/utils/harvester/models.py | 46 ++++++++++++++++++++++ 2 files changed, 57 insertions(+) create mode 100644 application/utils/harvester/artifact_id.py diff --git a/application/utils/harvester/artifact_id.py b/application/utils/harvester/artifact_id.py new file mode 100644 index 000000000..fa3ed46f1 --- /dev/null +++ b/application/utils/harvester/artifact_id.py @@ -0,0 +1,11 @@ +def generate_artifact_id(repository: str, file_path: str) -> str: + """ + Generate a stable artifact identifier for a repository file. + + Example: + repository = "OWASP/ASVS" + file_path = "5.0/en/0x01-Frontispiece.md" + + -> art:OWASP/ASVS:5.0/en/0x01-Frontispiece.md + """ + return f"art:{repository}:{file_path}" diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index 0eca718c9..27689cda2 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -39,3 +39,49 @@ class DiffBlock: repository: str commit_sha: str committed_at: datetime | None = None + + +@dataclass(slots=True) +class SourceInfo: + type: str + repository: str + commit_sha: str + committed_at: datetime + + +@dataclass(slots=True) +class Locator: + kind: str + id: str + path: str + + +@dataclass(slots=True) +class SpanInfo: + heading_path: list[str] + start_line: int + end_line: int + index: int | None = None + total: int | None = None + start_char_idx: int | None = None + end_char_idx: int | None = None + + +@dataclass(slots=True) +class HeadingNode: + level: int + text: str + start_line: int + end_line: int + + +@dataclass(slots=True) +class Document: + schema_version: str + artifact_id: str + pipeline_run_id: str + text: str + source: SourceInfo + locator: Locator + heading_structure: list[HeadingNode] + span: SpanInfo From 6986ec1ef8d7ccd742c6da0ee1a4969d592e71cf Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Tue, 14 Jul 2026 18:46:21 +0530 Subject: [PATCH 06/27] feat(harvester): read files from repository commits --- .../git_repository_client_test.py | 28 +++++++++++++++++ .../utils/harvester/git_repository_client.py | 31 +++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/application/tests/harvester_test/git_repository_client_test.py b/application/tests/harvester_test/git_repository_client_test.py index 8bbff6ab8..d012594c9 100644 --- a/application/tests/harvester_test/git_repository_client_test.py +++ b/application/tests/harvester_test/git_repository_client_test.py @@ -4,6 +4,7 @@ import tempfile from pathlib import Path +from unittest.mock import MagicMock from application.utils.harvester.git_repository_client import ( GitRepositoryClient, ) @@ -153,6 +154,33 @@ def test_clone_runs_git_command(self, mock_run): mock_run.assert_called() + @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") + + client = GitRepositoryClient("OWASP", "ASVS", "master") + + client.get_local_path = MagicMock(return_value="/tmp/repo") + + 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, + ) + if __name__ == "__main__": unittest.main() diff --git a/application/utils/harvester/git_repository_client.py b/application/utils/harvester/git_repository_client.py index 468be2665..b650b74f7 100644 --- a/application/utils/harvester/git_repository_client.py +++ b/application/utils/harvester/git_repository_client.py @@ -284,3 +284,34 @@ def is_valid_repository(self, repository_path: Path) -> bool: def verify_repository_integrity(self) -> bool: return self.is_valid_repository(self.local_path) + + def get_file_at_commit(self, commit_sha: str, file_path: str) -> str: + """ + Retrieve the contents of a file at a specific commit. + + Args: + commit_sha: + Commit to read from. + + file_path: + Repository-relative file path. + + Returns: + File contents as a string. + """ + + result = subprocess.run( + [ + "git", + "-C", + str(self.get_local_path()), + "show", + f"{commit_sha}:{file_path}", + ], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + + return result.stdout From 0f24d718732540ee07bf7e6fa0215d0771539bdc Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Tue, 14 Jul 2026 20:04:28 +0530 Subject: [PATCH 07/27] feat(harvester): extract markdown heading hierarchy --- .../harvester_test/heading_extractor_test.py | 107 ++++++++++++++++++ application/utils/harvester/__init__.py | 7 ++ .../utils/harvester/heading_extractor.py | 58 ++++++++++ 3 files changed, 172 insertions(+) create mode 100644 application/tests/harvester_test/heading_extractor_test.py create mode 100644 application/utils/harvester/heading_extractor.py diff --git a/application/tests/harvester_test/heading_extractor_test.py b/application/tests/harvester_test/heading_extractor_test.py new file mode 100644 index 000000000..e1411b9cd --- /dev/null +++ b/application/tests/harvester_test/heading_extractor_test.py @@ -0,0 +1,107 @@ +import unittest + +from application.utils.harvester.heading_extractor import ( + HeadingExtractor, +) + + +class HeadingExtractorTests(unittest.TestCase): + def test_single_heading(self): + text = """ +# Title + +Hello + +World +""" + + headings = HeadingExtractor().extract(text) + + self.assertEqual(len(headings), 1) + + self.assertEqual(headings[0].text, "Title") + self.assertEqual(headings[0].level, 1) + self.assertEqual(headings[0].start_line, 2) + + def test_nested_headings(self): + text = """ +# Root + +## Child One + +content + +## Child Two + +more + +# Second Root +""" + + headings = HeadingExtractor().extract(text) + + self.assertEqual(len(headings), 4) + + self.assertEqual(headings[0].text, "Root") + self.assertEqual(headings[1].text, "Child One") + self.assertEqual(headings[2].text, "Child Two") + self.assertEqual(headings[3].text, "Second Root") + + def test_heading_ranges(self): + text = """ +# Root + +text + +## Child + +child + +# Next +""" + + headings = HeadingExtractor().extract(text) + self.assertEqual(headings[0].end_line, 9) + self.assertEqual(headings[1].end_line, 9) + self.assertEqual(headings[2].end_line, 10) + + def test_ignore_non_headings(self): + text = """ +Hello + +###Heading + +####NoSpace + +## Valid Heading +""" + + headings = HeadingExtractor().extract(text) + self.assertEqual(len(headings), 1) + self.assertEqual(headings[0].text, "Valid Heading") + + def test_heading_stops_at_same_level(self): + text = """ +# Root + +## A + +### X + +## B + + content + """ + + headings = HeadingExtractor().extract(text) + + self.assertEqual(headings[1].text, "A") + self.assertEqual(headings[2].text, "X") + self.assertEqual(headings[3].text, "B") + + self.assertEqual(headings[1].end_line, 7) + self.assertEqual(headings[2].end_line, 7) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 9961aae16..cd80077d7 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -26,6 +26,11 @@ FilteringBenchmarkResult, ) +from .heading_extractor import ( + HeadingExtractor, + HeadingNode, +) + __all__ = [ "build_repository_cache_path", "ChunkingConfig", @@ -36,6 +41,8 @@ "FilteringMetricsCollector", "FilteringBenchmark", "FilteringBenchmarkResult", + "HeadingExtractor", + "HeadingNode", "PathRules", "PollingConfig", "RepositoryClient", diff --git a/application/utils/harvester/heading_extractor.py b/application/utils/harvester/heading_extractor.py new file mode 100644 index 000000000..a4d578e0a --- /dev/null +++ b/application/utils/harvester/heading_extractor.py @@ -0,0 +1,58 @@ +from dataclasses import dataclass + + +@dataclass(slots=True) +class HeadingNode: + """ + Represents a Markdown heading within a document. + """ + + level: int + text: str + start_line: int + end_line: int + + +class HeadingExtractor: + """ + Extracts Markdown headings and their line ranges. + + Heading ranges extend until the next heading of the same + or higher level, or the end of the document. + """ + + def extract(self, text: str) -> list[HeadingNode]: + lines = text.splitlines() + + headings: list[HeadingNode] = [] + + for line_number, line in enumerate(lines, start=1): + stripped = line.lstrip() + + if not stripped.startswith("#"): + continue + + hashes = len(stripped) - len(stripped.lstrip("#")) + + if hashes == 0: + continue + + if len(stripped) > hashes and stripped[hashes] != " ": + continue + + headings.append( + HeadingNode( + level=hashes, + text=stripped[hashes:].strip(), + start_line=line_number, + end_line=len(lines), + ) + ) + + for index, heading in enumerate(headings): + for next_heading in headings[index + 1 :]: + if next_heading.level <= heading.level: + heading.end_line = next_heading.start_line - 1 + break + + return headings From 291d2d48b3035291c89f27c5184c08809ca2ac9d Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Tue, 14 Jul 2026 20:49:05 +0530 Subject: [PATCH 08/27] feat(harvester): build structured document objects --- .../harvester_test/document_builder_test.py | 65 +++++++++++++++++++ application/utils/harvester/__init__.py | 3 + .../utils/harvester/document_builder.py | 46 +++++++++++++ .../utils/harvester/heading_extractor.py | 13 +--- application/utils/harvester/models.py | 4 +- 5 files changed, 117 insertions(+), 14 deletions(-) create mode 100644 application/tests/harvester_test/document_builder_test.py create mode 100644 application/utils/harvester/document_builder.py diff --git a/application/tests/harvester_test/document_builder_test.py b/application/tests/harvester_test/document_builder_test.py new file mode 100644 index 000000000..ba2313c08 --- /dev/null +++ b/application/tests/harvester_test/document_builder_test.py @@ -0,0 +1,65 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.document_builder import ( + DocumentBuilder, +) +from application.utils.harvester.models import ( + DiffBlock, +) + + +class DocumentBuilderTests(unittest.TestCase): + def test_build_document(self): + block = DiffBlock( + file_path="README.md", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + added_lines=["Hello"], + ) + + document = DocumentBuilder().build( + block, + "# Title\n\nHello", + pipeline_run_id="20260714T120000Z", + ) + + self.assertEqual( + document.schema_version, + "0.2.0", + ) + + self.assertEqual( + document.artifact_id, + "art:OWASP/ASVS:README.md", + ) + + self.assertEqual( + document.pipeline_run_id, + "20260714T120000Z", + ) + + self.assertEqual( + document.text, + "# Title\n\nHello", + ) + + self.assertEqual( + document.source.repository, + "OWASP/ASVS", + ) + + self.assertEqual( + document.locator.path, + "README.md", + ) + + self.assertEqual( + len(document.heading_structure), + 1, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index cd80077d7..ea1f6cfe7 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -31,11 +31,14 @@ HeadingNode, ) +from .document_builder import DocumentBuilder + __all__ = [ "build_repository_cache_path", "ChunkingConfig", "ConfigLoaderError", "DiffRetriever", + "DocumentBuilder", "GitRepositoryClient", "FileFilter", "FilteringMetricsCollector", diff --git a/application/utils/harvester/document_builder.py b/application/utils/harvester/document_builder.py new file mode 100644 index 000000000..9988c9f13 --- /dev/null +++ b/application/utils/harvester/document_builder.py @@ -0,0 +1,46 @@ +from .artifact_id import generate_artifact_id +from .heading_extractor import HeadingExtractor +from .models import ( + DiffBlock, + Document, + Locator, + SourceInfo, +) + + +class DocumentBuilder: + """ + Builds structured Document objects from parsed diffs. + + This bridges raw git diff ingestion and downstream + semantic chunking. + """ + + SCHEMA_VERSION = "0.2.0" + + def build(self, block: DiffBlock, full_text: str, pipeline_run_id: str) -> Document: + artifact_id = generate_artifact_id( + block.repository, + block.file_path, + ) + + headings = HeadingExtractor().extract(full_text) + + return Document( + schema_version=self.SCHEMA_VERSION, + artifact_id=artifact_id, + pipeline_run_id=pipeline_run_id, + text=full_text, + heading_structure=headings, + source=SourceInfo( + type="github", + repository=block.repository, + commit_sha=block.commit_sha, + committed_at=block.committed_at, + ), + locator=Locator( + kind="repo_path", + id=block.file_path, + path=block.file_path, + ), + ) diff --git a/application/utils/harvester/heading_extractor.py b/application/utils/harvester/heading_extractor.py index a4d578e0a..7941a805b 100644 --- a/application/utils/harvester/heading_extractor.py +++ b/application/utils/harvester/heading_extractor.py @@ -1,16 +1,5 @@ from dataclasses import dataclass - - -@dataclass(slots=True) -class HeadingNode: - """ - Represents a Markdown heading within a document. - """ - - level: int - text: str - start_line: int - end_line: int +from .models import HeadingNode class HeadingExtractor: diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index 27689cda2..b30985a0d 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -46,7 +46,7 @@ class SourceInfo: type: str repository: str commit_sha: str - committed_at: datetime + committed_at: datetime | None @dataclass(slots=True) @@ -84,4 +84,4 @@ class Document: source: SourceInfo locator: Locator heading_structure: list[HeadingNode] - span: SpanInfo + span: SpanInfo | None = None From ea7189fc262079a2759ac3ea508d85b146614ba0 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Tue, 14 Jul 2026 21:21:06 +0530 Subject: [PATCH 09/27] feat(harvester): validate structured documents --- .../harvester_test/document_validator_test.py | 85 +++++++++++++++++++ application/utils/harvester/__init__.py | 2 + .../utils/harvester/document_validator.py | 42 +++++++++ 3 files changed, 129 insertions(+) create mode 100644 application/tests/harvester_test/document_validator_test.py create mode 100644 application/utils/harvester/document_validator.py diff --git a/application/tests/harvester_test/document_validator_test.py b/application/tests/harvester_test/document_validator_test.py new file mode 100644 index 000000000..c445d7245 --- /dev/null +++ b/application/tests/harvester_test/document_validator_test.py @@ -0,0 +1,85 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.document_validator import ( + DocumentValidator, +) +from application.utils.harvester.models import ( + Document, + HeadingNode, + Locator, + SourceInfo, +) + + +def make_document() -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:OWASP/ASVS:README.md", + pipeline_run_id="20260714T120000Z", + text="# Title", + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + ), + locator=Locator( + kind="repo_path", + id="README.md", + path="README.md", + ), + heading_structure=[ + HeadingNode( + level=1, + text="Title", + start_line=1, + end_line=1, + ) + ], + span=None, + ) + + +class DocumentValidatorTests(unittest.TestCase): + def test_valid_document(self): + validator = DocumentValidator() + + self.assertTrue(validator.validate(make_document())) + + def test_missing_artifact_id(self): + validator = DocumentValidator() + + document = make_document() + document.artifact_id = "" + + self.assertFalse(validator.validate(document)) + + def test_missing_text(self): + validator = DocumentValidator() + + document = make_document() + document.text = "" + + self.assertFalse(validator.validate(document)) + + def test_invalid_source_type(self): + validator = DocumentValidator() + + document = make_document() + document.source.type = "gitlab" + + self.assertFalse(validator.validate(document)) + + def test_non_markdown_document_is_valid(self): + validator = DocumentValidator() + + document = make_document() + document.heading_structure = [] + document.text = '{"hello": "world"}' + + self.assertTrue(validator.validate(document)) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index ea1f6cfe7..af2d96ded 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -32,6 +32,7 @@ ) from .document_builder import DocumentBuilder +from .document_validator import DocumentValidator __all__ = [ "build_repository_cache_path", @@ -39,6 +40,7 @@ "ConfigLoaderError", "DiffRetriever", "DocumentBuilder", + "DocumentValidator", "GitRepositoryClient", "FileFilter", "FilteringMetricsCollector", diff --git a/application/utils/harvester/document_validator.py b/application/utils/harvester/document_validator.py new file mode 100644 index 000000000..9e099a4aa --- /dev/null +++ b/application/utils/harvester/document_validator.py @@ -0,0 +1,42 @@ +from .models import Document + + +class DocumentValidator: + """ + Validates structured Document objects before indexing. + + Ensures every required metadata field has been populated. + """ + + def validate(self, document: Document) -> bool: + if not document.schema_version: + return False + + if not document.artifact_id.startswith("art:"): + return False + + if not document.pipeline_run_id: + return False + + if not document.text: + return False + + if document.source.type != "github": + return False + + if not document.source.repository: + return False + + if not document.source.commit_sha: + return False + + if document.source.committed_at is None: + return False + + if document.locator.kind != "repo_path": + return False + + if not document.locator.path: + return False + + return True From aa3f436eb2038fb423483dbf698d240f78cb573c Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 17 Jul 2026 00:11:20 +0530 Subject: [PATCH 10/27] feat(harvester): add content hashing for deduplication --- .../tests/harvester_test/content_hash_test.py | 35 +++++++++++++++++++ application/utils/harvester/__init__.py | 2 ++ application/utils/harvester/content_hash.py | 13 +++++++ 3 files changed, 50 insertions(+) create mode 100644 application/tests/harvester_test/content_hash_test.py create mode 100644 application/utils/harvester/content_hash.py 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/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index af2d96ded..039905183 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -33,6 +33,7 @@ from .document_builder import DocumentBuilder from .document_validator import DocumentValidator +from .content_hash import generate_content_hash __all__ = [ "build_repository_cache_path", @@ -46,6 +47,7 @@ "FilteringMetricsCollector", "FilteringBenchmark", "FilteringBenchmarkResult", + "generate_content_hash", "HeadingExtractor", "HeadingNode", "PathRules", 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() From 6abf604ab17575ae71c74291b355bdc163d3a460 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 17 Jul 2026 00:33:55 +0530 Subject: [PATCH 11/27] feat(harvester): add artifact registry --- .../harvester_test/artifact_registry_test.py | 99 +++++++++++++++++++ application/utils/harvester/__init__.py | 2 + .../utils/harvester/artifact_registry.py | 24 +++++ application/utils/harvester/models.py | 17 ++++ 4 files changed, 142 insertions(+) create mode 100644 application/tests/harvester_test/artifact_registry_test.py create mode 100644 application/utils/harvester/artifact_registry.py 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/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 039905183..1728abcc7 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -34,8 +34,10 @@ from .document_builder import DocumentBuilder from .document_validator import DocumentValidator from .content_hash import generate_content_hash +from .artifact_registry import ArtifactRegistry __all__ = [ + "ArtifactRegistry", "build_repository_cache_path", "ChunkingConfig", "ConfigLoaderError", 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/models.py b/application/utils/harvester/models.py index b30985a0d..86284bcf0 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -85,3 +85,20 @@ 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 across pipeline runs. + """ + + 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 From 5ae56e07e74c0599d0ae98d297ae890b9cc3a44a Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 17 Jul 2026 00:54:32 +0530 Subject: [PATCH 12/27] feat(harvester): implement document deduplication --- .../document_deduplicator_test.py | 71 +++++++++++++++++++ application/utils/harvester/__init__.py | 2 + .../utils/harvester/document_deduplicator.py | 59 +++++++++++++++ application/utils/harvester/models.py | 7 ++ 4 files changed, 139 insertions(+) create mode 100644 application/tests/harvester_test/document_deduplicator_test.py create mode 100644 application/utils/harvester/document_deduplicator.py 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..3aa4c56f6 --- /dev/null +++ b/application/tests/harvester_test/document_deduplicator_test.py @@ -0,0 +1,71 @@ +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(self): + registry = ArtifactRegistry() + + deduplicator = DocumentDeduplicator(registry) + document = self.create_document("hello") + + deduplicator.process(document) + result = deduplicator.process(document) + + self.assertEqual(result, DeduplicationStatus.UNCHANGED) + + 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/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 1728abcc7..35ff98ef8 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -35,6 +35,7 @@ from .document_validator import DocumentValidator from .content_hash import generate_content_hash from .artifact_registry import ArtifactRegistry +from .document_deduplicator import DocumentDeduplicator __all__ = [ "ArtifactRegistry", @@ -43,6 +44,7 @@ "ConfigLoaderError", "DiffRetriever", "DocumentBuilder", + "DocumentDeduplicator", "DocumentValidator", "GitRepositoryClient", "FileFilter", diff --git a/application/utils/harvester/document_deduplicator.py b/application/utils/harvester/document_deduplicator.py new file mode 100644 index 000000000..8150d994d --- /dev/null +++ b/application/utils/harvester/document_deduplicator.py @@ -0,0 +1,59 @@ +from .artifact_registry import ArtifactRegistry +from .content_hash import generate_content_hash +from .models import ( + ArtifactRegistryRecord, + DeduplicationStatus, + Document, +) +from datetime import datetime + + +class DocumentDeduplicator: + """ + Performs artifact-level deduplication. + + Documents are classified as: + - NEW + - UPDATED + - 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) + + 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=datetime.now(), + status=DeduplicationStatus.NEW.value, + ) + ) + + return DeduplicationStatus.NEW + + if existing.content_hash == content_hash: + existing.status = DeduplicationStatus.UNCHANGED.value + + self._registry.upsert(existing) + + return DeduplicationStatus.UNCHANGED + + existing.content_hash = content_hash + existing.last_commit_sha = document.source.commit_sha + existing.last_pipeline_run = document.pipeline_run_id + existing.status = DeduplicationStatus.UPDATED.value + + self._registry.upsert(existing) + + return DeduplicationStatus.UPDATED diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index 86284bcf0..df28e983f 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -1,6 +1,7 @@ from dataclasses import dataclass from datetime import datetime from pydantic import BaseModel +from enum import Enum @dataclass(slots=True) @@ -102,3 +103,9 @@ class ArtifactRegistryRecord: last_pipeline_run: str last_processed_at: datetime status: str + + +class DeduplicationStatus(str, Enum): + NEW = "new" + UPDATED = "updated" + UNCHANGED = "unchanged" From 57b34d79ac39daa8e2ea45cf7fceeb68bdec9b55 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 17 Jul 2026 01:02:44 +0530 Subject: [PATCH 13/27] feat(harvester): add checkpoint management --- .../harvester_test/checkpoint_manager_test.py | 68 +++++++++++++++++++ application/utils/harvester/__init__.py | 2 + .../utils/harvester/checkpoint_manager.py | 36 ++++++++++ application/utils/harvester/models.py | 9 +++ 4 files changed, 115 insertions(+) create mode 100644 application/tests/harvester_test/checkpoint_manager_test.py create mode 100644 application/utils/harvester/checkpoint_manager.py 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/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 35ff98ef8..4709a060c 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -36,10 +36,12 @@ from .content_hash import generate_content_hash from .artifact_registry import ArtifactRegistry from .document_deduplicator import DocumentDeduplicator +from .checkpoint_manager import CheckpointManager __all__ = [ "ArtifactRegistry", "build_repository_cache_path", + "CheckpointManager", "ChunkingConfig", "ConfigLoaderError", "DiffRetriever", 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/models.py b/application/utils/harvester/models.py index df28e983f..4a899d5a3 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -109,3 +109,12 @@ 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 From f0f93681319e2ed73836fbad7334225addb11c81 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 17 Jul 2026 01:13:45 +0530 Subject: [PATCH 14/27] feat(harvester): orchestrate incremental document processing --- .../incremental_pipeline_test.py | 76 +++++++++++++++++++ application/utils/harvester/__init__.py | 2 + .../utils/harvester/incremental_pipeline.py | 58 ++++++++++++++ 3 files changed, 136 insertions(+) create mode 100644 application/tests/harvester_test/incremental_pipeline_test.py create mode 100644 application/utils/harvester/incremental_pipeline.py 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..2b10814ed --- /dev/null +++ b/application/tests/harvester_test/incremental_pipeline_test.py @@ -0,0 +1,76 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.checkpoint_manager import CheckpointManager +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) -> 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_only_new_and_updated_are_emitted(self): + registry = ArtifactRegistry() + + dedup = DocumentDeduplicator( + registry, + ) + + checkpoints = CheckpointManager() + + pipeline = IncrementalPipeline( + dedup, + checkpoints, + ) + + docs = [ + self.make_document("hello"), + self.make_document("hello"), + self.make_document("changed"), + ] + + emitted = pipeline.process( + "OWASP/ASVS", + "run1", + docs, + ) + + self.assertEqual(len(emitted), 2) + checkpoint = checkpoints.get("OWASP/ASVS") + + assert checkpoint is not None + self.assertEqual(checkpoint.status, "completed") + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 4709a060c..77049b28e 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -37,6 +37,7 @@ from .artifact_registry import ArtifactRegistry from .document_deduplicator import DocumentDeduplicator from .checkpoint_manager import CheckpointManager +from .incremental_pipeline import IncrementalPipeline __all__ = [ "ArtifactRegistry", @@ -56,6 +57,7 @@ "generate_content_hash", "HeadingExtractor", "HeadingNode", + "IncrementalPipeline", "PathRules", "PollingConfig", "RepositoryClient", diff --git a/application/utils/harvester/incremental_pipeline.py b/application/utils/harvester/incremental_pipeline.py new file mode 100644 index 000000000..4402ebcba --- /dev/null +++ b/application/utils/harvester/incremental_pipeline.py @@ -0,0 +1,58 @@ +from datetime import datetime +from .checkpoint_manager import CheckpointManager +from .document_deduplicator import ( + DeduplicationStatus, + DocumentDeduplicator, +) +from .models import ( + CheckpointRecord, + Document, +) + + +class IncrementalPipeline: + """ + Coordinates document deduplication and checkpoint updates. + + Only NEW or UPDATED documents are emitted downstream. + """ + + def __init__( + self, deduplicator: DocumentDeduplicator, checkpoint_manager: CheckpointManager + ): + self._deduplicator = deduplicator + self._checkpoint_manager = checkpoint_manager + + def process( + self, repository: str, pipeline_run_id: str, documents: list[Document] + ) -> list[Document]: + + emitted: list[Document] = [] + + if documents: + self._checkpoint_manager.save( + CheckpointRecord( + repository=repository, + pipeline_run_id=pipeline_run_id, + last_processed_commit="", + status="running", + updated_at=datetime.now(), + ) + ) + + for document in documents: + status = self._deduplicator.process(document) + + self._checkpoint_manager.update_commit( + repository, + document.source.commit_sha, + ) + + if status != DeduplicationStatus.UNCHANGED: + emitted.append(document) + + self._checkpoint_manager.mark_completed( + repository, + ) + + return emitted From c46a8489bbd21dd525c10941fb0c302b6e8d0389 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 17 Jul 2026 01:26:51 +0530 Subject: [PATCH 15/27] feat(harvester): track deduplication metrics --- .../deduplication_metrics_test.py | 35 +++++++++++++++++++ application/utils/harvester/__init__.py | 3 ++ .../utils/harvester/deduplication_metrics.py | 29 +++++++++++++++ .../utils/harvester/incremental_pipeline.py | 6 ++++ 4 files changed, 73 insertions(+) create mode 100644 application/tests/harvester_test/deduplication_metrics_test.py create mode 100644 application/utils/harvester/deduplication_metrics.py 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/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 77049b28e..c73f833fc 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -21,6 +21,7 @@ from .filtering_metrics import FilteringMetricsCollector from .diff_retriever import DiffRetriever + from .filtering_benchmark import ( FilteringBenchmark, FilteringBenchmarkResult, @@ -38,6 +39,7 @@ from .document_deduplicator import DocumentDeduplicator from .checkpoint_manager import CheckpointManager from .incremental_pipeline import IncrementalPipeline +from .deduplication_metrics import DeduplicationMetrics __all__ = [ "ArtifactRegistry", @@ -45,6 +47,7 @@ "CheckpointManager", "ChunkingConfig", "ConfigLoaderError", + "DeduplicationMetrics", "DiffRetriever", "DocumentBuilder", "DocumentDeduplicator", 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/incremental_pipeline.py b/application/utils/harvester/incremental_pipeline.py index 4402ebcba..52a96c766 100644 --- a/application/utils/harvester/incremental_pipeline.py +++ b/application/utils/harvester/incremental_pipeline.py @@ -1,5 +1,6 @@ from datetime import datetime from .checkpoint_manager import CheckpointManager +from .deduplication_metrics import DeduplicationMetrics from .document_deduplicator import ( DeduplicationStatus, DocumentDeduplicator, @@ -20,14 +21,17 @@ class IncrementalPipeline: def __init__( self, deduplicator: DocumentDeduplicator, checkpoint_manager: CheckpointManager ): + self._deduplicator = deduplicator self._checkpoint_manager = checkpoint_manager + self.metrics = DeduplicationMetrics() def process( self, repository: str, pipeline_run_id: str, documents: list[Document] ) -> list[Document]: emitted: list[Document] = [] + metrics = DeduplicationMetrics() if documents: self._checkpoint_manager.save( @@ -43,6 +47,7 @@ def process( for document in documents: status = self._deduplicator.process(document) + metrics.record(status) self._checkpoint_manager.update_commit( repository, document.source.commit_sha, @@ -55,4 +60,5 @@ def process( repository, ) + self.metrics = metrics return emitted From 5750abc2a797c148d09e2a16b13df46253e32adb Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Mon, 24 Aug 2026 17:24:27 +0530 Subject: [PATCH 16/27] resolve the textacy stuff --- application/tests/harvester_test/diff_pipeline_test.py | 2 +- application/utils/harvester/diff_normalizer.py | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/application/tests/harvester_test/diff_pipeline_test.py b/application/tests/harvester_test/diff_pipeline_test.py index 37c7f25a9..e5420a4ed 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 -import os from application.utils.harvester.diff_normalizer import DiffNormalizer from application.utils.harvester.diff_parser import DiffParser diff --git a/application/utils/harvester/diff_normalizer.py b/application/utils/harvester/diff_normalizer.py index fe8773349..756d89147 100644 --- a/application/utils/harvester/diff_normalizer.py +++ b/application/utils/harvester/diff_normalizer.py @@ -1,6 +1,6 @@ -import textacy.preprocessing as prep +import re +import unicodedata -from application.utils.harvester import repository_client from .models import DiffBlock @@ -13,8 +13,8 @@ class DiffNormalizer: """ def normalize_line(self, line: str) -> str: - line = prep.normalize.unicode(line) - line = prep.normalize.whitespace(line) + line = unicodedata.normalize("NFKC", line) + line = re.sub(r"\s+", " ", line) return line.strip() def normalize(self, blocks: list[DiffBlock]) -> list[DiffBlock]: From 955fff61a61132cd933b04303f71191d02a2f77b Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 13:47:17 +0530 Subject: [PATCH 17/27] feat(harvester): add git diff retrieval pipeline --- application/tests/harvester_test/diff_retriever_test.py | 2 ++ application/utils/harvester/__init__.py | 1 - application/utils/harvester/diff_retriever.py | 1 + 3 files changed, 3 insertions(+), 1 deletion(-) diff --git a/application/tests/harvester_test/diff_retriever_test.py b/application/tests/harvester_test/diff_retriever_test.py index f8c3ea142..1a5e8a8db 100644 --- a/application/tests/harvester_test/diff_retriever_test.py +++ b/application/tests/harvester_test/diff_retriever_test.py @@ -18,6 +18,7 @@ def test_get_diff(self, mock_run): ] client = MagicMock() client.get_local_path.return_value = "/tmp/repo" + retriever = DiffRetriever(client) diff = retriever.get_diff( @@ -29,6 +30,7 @@ def test_get_diff(self, mock_run): diff, "diff --git a/README.md b/README.md\n", ) + mock_run.assert_has_calls( [ call( diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index c73f833fc..8e89a64e0 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -21,7 +21,6 @@ from .filtering_metrics import FilteringMetricsCollector from .diff_retriever import DiffRetriever - from .filtering_benchmark import ( FilteringBenchmark, FilteringBenchmarkResult, diff --git a/application/utils/harvester/diff_retriever.py b/application/utils/harvester/diff_retriever.py index 665d10d19..7f18bee7c 100644 --- a/application/utils/harvester/diff_retriever.py +++ b/application/utils/harvester/diff_retriever.py @@ -40,6 +40,7 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: base_commit, target_commit, ) + base_commit = self._resolve_commit(base_commit) target_commit = self._resolve_commit(target_commit) From e6465f2a1e2b524df72bbd5be04f028f81b22ede Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 14:00:26 +0530 Subject: [PATCH 18/27] feat(harvester): parse unified git diffs --- application/tests/harvester_test/diff_parser_test.py | 4 ---- application/utils/harvester/models.py | 3 ++- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/application/tests/harvester_test/diff_parser_test.py b/application/tests/harvester_test/diff_parser_test.py index 877d70a09..48f9d954d 100644 --- a/application/tests/harvester_test/diff_parser_test.py +++ b/application/tests/harvester_test/diff_parser_test.py @@ -5,10 +5,6 @@ DiffParser, ) -TEST_REPOSITORY = "OWASP/ASVS" -TEST_COMMIT_SHA = "abc123" -TEST_COMMITTED_AT = datetime.now(UTC) - TEST_REPOSITORY = "OWASP/ASVS" TEST_COMMIT_SHA = "abc123" diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index 4a899d5a3..1d8c2958b 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -1,8 +1,9 @@ from dataclasses import dataclass from datetime import datetime -from pydantic import BaseModel from enum import Enum +from pydantic import BaseModel + @dataclass(slots=True) class RepositoryCheckpoint: From a1100fa7fb96fbb2fda4d8de22c13da046758269 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 10 Jul 2026 17:14:07 +0530 Subject: [PATCH 19/27] Enhance diff pipeline with metadata and normalization --- application/tests/harvester_test/diff_parser_test.py | 4 ++++ application/tests/harvester_test/diff_pipeline_test.py | 5 ++++- application/tests/harvester_test/diff_retriever_test.py | 3 ++- application/utils/harvester/diff_normalizer.py | 1 + 4 files changed, 11 insertions(+), 2 deletions(-) diff --git a/application/tests/harvester_test/diff_parser_test.py b/application/tests/harvester_test/diff_parser_test.py index 48f9d954d..877d70a09 100644 --- a/application/tests/harvester_test/diff_parser_test.py +++ b/application/tests/harvester_test/diff_parser_test.py @@ -5,6 +5,10 @@ DiffParser, ) +TEST_REPOSITORY = "OWASP/ASVS" +TEST_COMMIT_SHA = "abc123" +TEST_COMMITTED_AT = datetime.now(UTC) + TEST_REPOSITORY = "OWASP/ASVS" TEST_COMMIT_SHA = "abc123" diff --git a/application/tests/harvester_test/diff_pipeline_test.py b/application/tests/harvester_test/diff_pipeline_test.py index e5420a4ed..cbc3a424f 100644 --- a/application/tests/harvester_test/diff_pipeline_test.py +++ b/application/tests/harvester_test/diff_pipeline_test.py @@ -13,7 +13,6 @@ 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 +62,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 1a5e8a8db..db78e6563 100644 --- a/application/tests/harvester_test/diff_retriever_test.py +++ b/application/tests/harvester_test/diff_retriever_test.py @@ -16,11 +16,11 @@ def test_get_diff(self, mock_run): MagicMock(stdout="def456\n"), MagicMock(stdout=b"diff --git a/README.md b/README.md\n"), ] + client = MagicMock() client.get_local_path.return_value = "/tmp/repo" retriever = DiffRetriever(client) - diff = retriever.get_diff( "abc123", "def456", @@ -87,6 +87,7 @@ def test_large_diff_raises(self, mock_run): client = MagicMock() client.get_local_path.return_value = "/tmp/repo" + retriever = DiffRetriever(client) with self.assertRaises(ValueError): diff --git a/application/utils/harvester/diff_normalizer.py b/application/utils/harvester/diff_normalizer.py index 756d89147..d3e923a01 100644 --- a/application/utils/harvester/diff_normalizer.py +++ b/application/utils/harvester/diff_normalizer.py @@ -1,6 +1,7 @@ import re import unicodedata +from application.utils.harvester import repository_client from .models import DiffBlock From 8fba3443882fe50e83aba9f3432f870bcf31c14d Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 14 Aug 2026 13:02:56 +0530 Subject: [PATCH 20/27] feat(harvester): add LlamaIndex document chunking foundation --- .../tests/harvester_test/chunker_test.py | 73 +++++++++++++++++ application/utils/harvester/__init__.py | 3 + application/utils/harvester/chunker.py | 79 +++++++++++++++++++ requirements-dev.txt | 4 + 4 files changed, 159 insertions(+) create mode 100644 application/tests/harvester_test/chunker_test.py create mode 100644 application/utils/harvester/chunker.py diff --git a/application/tests/harvester_test/chunker_test.py b/application/tests/harvester_test/chunker_test.py new file mode 100644 index 000000000..6c85a3565 --- /dev/null +++ b/application/tests/harvester_test/chunker_test.py @@ -0,0 +1,73 @@ +import unittest +from unittest.mock import patch + +from application.utils.harvester.chunker import ( + ChunkInfo, + DocumentChunker, +) + + +class DocumentChunkerTests(unittest.TestCase): + @patch("application.utils.harvester.chunker.HuggingFaceEmbedding") + @patch("application.utils.harvester.chunker.SemanticSplitterNodeParser") + def test_empty_document_returns_no_chunks( + self, + splitter_cls, + embedding_cls, + ): + chunker = DocumentChunker() + + self.assertEqual(chunker.chunk(""), []) + splitter_cls.assert_not_called() + + @patch("application.utils.harvester.chunker.HuggingFaceEmbedding") + @patch("application.utils.harvester.chunker.SemanticSplitterNodeParser") + def test_whitespace_document_returns_no_chunks( + self, + splitter_cls, + embedding_cls, + ): + chunker = DocumentChunker() + + self.assertEqual(chunker.chunk(" \n\n "), []) + splitter_cls.assert_not_called() + + @patch("application.utils.harvester.chunker.HuggingFaceEmbedding") + @patch("application.utils.harvester.chunker.SemanticSplitterNodeParser") + def test_chunks_preserve_node_boundaries( + self, + splitter_cls, + embedding_cls, + ): + node = type( + "Node", + (), + { + "start_char_idx": 0, + "end_char_idx": 12, + "get_content": lambda self: "# Heading\ntext", + }, + )() + + splitter_cls.return_value.get_nodes_from_documents.return_value = [ + node, + ] + + chunker = DocumentChunker() + + chunks = chunker.chunk("# Heading\ntext") + + self.assertEqual( + chunks, + [ + ChunkInfo( + text="# Heading\ntext", + start_char_idx=0, + end_char_idx=12, + ) + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 8e89a64e0..52e1c1d95 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -39,16 +39,19 @@ from .checkpoint_manager import CheckpointManager from .incremental_pipeline import IncrementalPipeline from .deduplication_metrics import DeduplicationMetrics +from .chunker import ChunkInfo, DocumentChunker __all__ = [ "ArtifactRegistry", "build_repository_cache_path", + "ChunkInfo", "CheckpointManager", "ChunkingConfig", "ConfigLoaderError", "DeduplicationMetrics", "DiffRetriever", "DocumentBuilder", + "DocumentChunker", "DocumentDeduplicator", "DocumentValidator", "GitRepositoryClient", diff --git a/application/utils/harvester/chunker.py b/application/utils/harvester/chunker.py new file mode 100644 index 000000000..a4a4046e8 --- /dev/null +++ b/application/utils/harvester/chunker.py @@ -0,0 +1,79 @@ +from dataclasses import dataclass + +from llama_index.core import Document as LlamaDocument +from llama_index.core.node_parser import SemanticSplitterNodeParser +from llama_index.embeddings.huggingface import HuggingFaceEmbedding +from llama_index.core.schema import TextNode +from typing import cast + + +@dataclass(slots=True) +class ChunkInfo: + text: str + start_char_idx: int + end_char_idx: int + + +class DocumentChunker: + """ + Splits documents into semantically coherent chunks while + preserving the original document text boundaries. + """ + + DEFAULT_BUFFER_SIZE = 1 + DEFAULT_BREAKPOINT_PERCENTILE = 95 + DEFAULT_EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2" + + def __init__( + self, + embedding_model: str = DEFAULT_EMBEDDING_MODEL, + buffer_size: int = DEFAULT_BUFFER_SIZE, + breakpoint_percentile_threshold: int = DEFAULT_BREAKPOINT_PERCENTILE, + ) -> None: + self._embedding_model = embedding_model + self._buffer_size = buffer_size + self._breakpoint_percentile_threshold = breakpoint_percentile_threshold + self._splitter: SemanticSplitterNodeParser | None = None + + def _get_splitter(self) -> SemanticSplitterNodeParser: + if self._splitter is None: + embed_model = HuggingFaceEmbedding( + model_name=self._embedding_model, + ) + + self._splitter = SemanticSplitterNodeParser( + buffer_size=self._buffer_size, + breakpoint_percentile_threshold=self._breakpoint_percentile_threshold, + embed_model=embed_model, + ) + + return self._splitter + + def chunk(self, text: str) -> list[ChunkInfo]: + if not text.strip(): + return [] + + document = LlamaDocument(text=text) + + nodes = self._get_splitter().get_nodes_from_documents([document]) + + chunks: list[ChunkInfo] = [] + + for node in nodes: + text_node = cast(TextNode, node) + + start = text_node.start_char_idx + end = text_node.end_char_idx + + if start is None or end is None: + continue + + chunks.append( + ChunkInfo( + text=text_node.get_content(), + start_char_idx=start, + end_char_idx=end, + ) + ) + + return chunks diff --git a/requirements-dev.txt b/requirements-dev.txt index 7121de20e..c2eebf6a4 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -89,3 +89,7 @@ pytest-playwright # Local stdio MCP server (issue #1003 v1) — not needed on Heroku slug mcp==2.0.0 + +# LlamaIndex +llama-index-core +llama-index-embeddings-huggingface From aae28c324aa87cd6e10b0e0260856df00bd0c140 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 14 Aug 2026 18:00:43 +0530 Subject: [PATCH 21/27] feat(harvester): preserve document structure in chunks --- .../chunk_record_builder_test.py | 274 ++++++++++++++++++ application/utils/harvester/chunk_pipeline.py | 26 ++ .../utils/harvester/chunk_record_builder.py | 149 ++++++++++ application/utils/harvester/models.py | 13 + 4 files changed, 462 insertions(+) create mode 100644 application/tests/harvester_test/chunk_record_builder_test.py create mode 100644 application/utils/harvester/chunk_pipeline.py create mode 100644 application/utils/harvester/chunk_record_builder.py 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..df982bd95 --- /dev/null +++ b/application/tests/harvester_test/chunk_record_builder_test.py @@ -0,0 +1,274 @@ +import unittest + +from application.utils.harvester.chunk_record_builder import ( + ChunkRecordBuilder, +) +from application.utils.harvester.chunker import ChunkInfo +from application.utils.harvester.models import ( + Document, + HeadingNode, + Locator, + SourceInfo, +) + + +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="abc123", + committed_at=None, + ), + locator=Locator( + kind="repo_path", + id="README.md", + path="README.md", + ), + heading_structure=headings, + ) + + def test_builds_chunk_record(self): + 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.artifact_id, + document.artifact_id, + ) + + self.assertEqual( + record.text, + text, + ) + + self.assertEqual( + record.span.index, + 0, + ) + + self.assertEqual( + record.span.total, + 1, + ) + + self.assertEqual( + record.span.heading_path, + ["Root"], + ) + + self.assertEqual( + record.span.start_char_idx, + 0, + ) + + self.assertEqual( + record.span.end_char_idx, + len(text), + ) + + self.assertEqual( + record.span.start_line, + 1, + ) + + self.assertEqual( + record.span.end_line, + 3, + ) + + def test_heading_path_follows_chunk_start(self): + text = "# Root\n\nroot content\n\n## Child\n\nchild content" + + document = self._document( + text, + [ + HeadingNode( + level=1, + text="Root", + start_line=1, + end_line=7, + ), + HeadingNode( + level=2, + text="Child", + start_line=5, + end_line=7, + ), + ], + ) + + root_end = text.index("## Child") + + chunks = [ + ChunkInfo( + text=text[:root_end], + start_char_idx=0, + end_char_idx=root_end, + ), + ChunkInfo( + text=text[root_end:], + start_char_idx=root_end, + end_char_idx=len(text), + ), + ] + + records = ChunkRecordBuilder().build( + document, + chunks, + ) + + self.assertEqual( + records[0].span.heading_path, + ["Root"], + ) + + self.assertEqual( + records[1].span.heading_path, + ["Root", "Child"], + ) + + def test_line_ranges_are_derived_from_character_offsets(self): + text = "one\ntwo\nthree\nfour" + + document = self._document(text, []) + + chunks = [ + ChunkInfo( + text="two\nthree", + start_char_idx=4, + end_char_idx=13, + ) + ] + + records = ChunkRecordBuilder().build( + document, + chunks, + ) + + self.assertEqual( + records[0].span.start_line, + 2, + ) + + self.assertEqual( + records[0].span.end_line, + 3, + ) + + def test_chunk_ids_are_deterministic(self): + text = "# Root\n\nContent" + + 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), + ) + + builder = ChunkRecordBuilder() + + first = builder.build(document, [chunk]) + second = builder.build(document, [chunk]) + + self.assertEqual( + first[0].chunk_id, + second[0].chunk_id, + ) + + def test_chunk_ids_include_heading_path_and_content_hash(self): + text = "# Root\n\nContent" + + 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), + ) + + record = ChunkRecordBuilder().build( + document, + [chunk], + )[0] + + self.assertTrue( + record.chunk_id.startswith("chk:art:OWASP/ASVS:README.md:Root:") + ) + + def test_empty_heading_path_is_allowed(self): + text = "Plain text without headings." + + document = self._document(text, []) + + chunk = ChunkInfo( + text=text, + start_char_idx=0, + end_char_idx=len(text), + ) + + record = ChunkRecordBuilder().build( + document, + [chunk], + )[0] + + self.assertEqual( + record.span.heading_path, + [], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/chunk_pipeline.py b/application/utils/harvester/chunk_pipeline.py new file mode 100644 index 000000000..2283c16ac --- /dev/null +++ b/application/utils/harvester/chunk_pipeline.py @@ -0,0 +1,26 @@ +from .chunk_record_builder import ChunkRecordBuilder +from .chunker import DocumentChunker +from .models import Document, IngestChunkRecord + + +class DocumentChunkPipeline: + """ + Runs semantic chunking followed by structure-aware RFC + chunk-record construction. + """ + + def __init__( + self, + chunker: DocumentChunker | None = None, + record_builder: ChunkRecordBuilder | None = None, + ) -> None: + self._chunker = chunker or DocumentChunker() + self._record_builder = record_builder or ChunkRecordBuilder() + + def chunk(self, document: Document) -> list[IngestChunkRecord]: + chunks = self._chunker.chunk(document.text) + + return self._record_builder.build( + document, + chunks, + ) diff --git a/application/utils/harvester/chunk_record_builder.py b/application/utils/harvester/chunk_record_builder.py new file mode 100644 index 000000000..bc3f9e461 --- /dev/null +++ b/application/utils/harvester/chunk_record_builder.py @@ -0,0 +1,149 @@ +import hashlib +from dataclasses import dataclass + +from .models import Document, IngestChunkRecord, SpanInfo +from .chunker import ChunkInfo + + +@dataclass(slots=True) +class ChunkRecordBuilder: + """ + Converts semantic ChunkInfo objects into structure-aware + ingestion chunk records. + """ + + SCHEMA_VERSION = "0.2.0" + + def build( + self, + document: Document, + chunks: list[ChunkInfo], + ) -> list[IngestChunkRecord]: + total = len(chunks) + + records: list[IngestChunkRecord] = [] + + for index, chunk in enumerate(chunks): + heading_path = self._heading_path_for_chunk( + document, + chunk, + ) + + content_hash = hashlib.sha256(chunk.text.encode("utf-8")).hexdigest() + + chunk_id = self._chunk_id( + artifact_id=document.artifact_id, + heading_path=heading_path, + content_hash=content_hash, + ) + + 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=chunk_id, + artifact_id=document.artifact_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, + ), + ) + ) + + return records + + @staticmethod + def _heading_path_for_chunk( + document: Document, + chunk: ChunkInfo, + ) -> list[str]: + """ + Determine the Markdown heading hierarchy containing the + beginning of the chunk. + + A heading is considered active when its range contains + the chunk's starting line. + """ + + 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]: + """ + Convert zero-based character offsets into one-based + inclusive line numbers. + + The end offset is treated as exclusive. + """ + + 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 + + @staticmethod + def _chunk_id( + artifact_id: str, + heading_path: list[str], + content_hash: str, + ) -> str: + heading = "/".join(heading_path) + + return f"chk:{artifact_id}:{heading}:{content_hash}" diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index 1d8c2958b..55fed4f14 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -119,3 +119,16 @@ class CheckpointRecord: last_processed_commit: str status: str updated_at: datetime + + +@dataclass(slots=True) +class IngestChunkRecord: + """ + RFC-facing representation of a semantically chunked document. + """ + + schema_version: str + chunk_id: str + artifact_id: str + text: str + span: SpanInfo From f29dd0c11510d8f9e35b3434bf62953dec574236 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 14 Aug 2026 18:24:15 +0530 Subject: [PATCH 22/27] feat(harvester): validate chunk records and benchmark chunking --- .../chunk_record_builder_test.py | 62 ++++++++++++++ .../chunk_record_validator_test.py | 81 +++++++++++++++++++ .../tests/harvester_test/chunker_test.py | 46 +++++++++++ .../harvester_test/chunking_benchmark_test.py | 46 +++++++++++ .../utils/harvester/chunk_record_validator.py | 48 +++++++++++ 5 files changed, 283 insertions(+) create mode 100644 application/tests/harvester_test/chunk_record_validator_test.py create mode 100644 application/tests/harvester_test/chunking_benchmark_test.py create mode 100644 application/utils/harvester/chunk_record_validator.py diff --git a/application/tests/harvester_test/chunk_record_builder_test.py b/application/tests/harvester_test/chunk_record_builder_test.py index df982bd95..62d33687d 100644 --- a/application/tests/harvester_test/chunk_record_builder_test.py +++ b/application/tests/harvester_test/chunk_record_builder_test.py @@ -269,6 +269,68 @@ def test_empty_heading_path_is_allowed(self): [], ) + def test_empty_document_produces_no_records(self): + document = self._document(text="", headings=[]) + + records = ChunkRecordBuilder().build(document, []) + + self.assertEqual(records, []) + + def test_chunk_text_matches_source_span(self): + text = "# Root\n\nFirst paragraph.\n\nSecond paragraph." + + document = self._document(text=text, headings=[]) + + chunks = [ + ChunkInfo( + text="First paragraph.", + start_char_idx=text.index("First"), + end_char_idx=text.index("First") + len("First paragraph."), + ) + ] + + records = ChunkRecordBuilder().build(document, chunks) + + record = records[0] + + self.assertEqual( + record.text, + text[record.span.start_char_idx : record.span.end_char_idx], + ) + + def test_chunk_order_is_deterministic(self): + text = "# Root\n\nFirst.\n\nSecond." + + document = self._document(text=text, headings=[]) + + first_start = text.index("First") + second_start = text.index("Second") + + chunks = [ + ChunkInfo( + text="First.", + start_char_idx=first_start, + end_char_idx=first_start + len("First."), + ), + ChunkInfo( + text="Second.", + start_char_idx=second_start, + end_char_idx=second_start + len("Second."), + ), + ] + + records = ChunkRecordBuilder().build(document, chunks) + + self.assertEqual( + [record.span.index for record in records], + [0, 1], + ) + + self.assertEqual( + [record.span.total for record in records], + [2, 2], + ) + 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..7ba07e9b9 --- /dev/null +++ b/application/tests/harvester_test/chunk_record_validator_test.py @@ -0,0 +1,81 @@ +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:abc123", + artifact_id="art:OWASP/OpenCRE:README.md", + 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, + ), + ) + + +class ChunkRecordValidatorTests(unittest.TestCase): + def test_valid_record(self): + ChunkRecordValidator().validate(valid_record()) + + def test_empty_text_is_rejected(self): + record = valid_record() + record.text = " " + + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_invalid_chunk_id_is_rejected(self): + record = valid_record() + record.chunk_id = "invalid-id" + + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_missing_span_index_is_rejected(self): + record = valid_record() + record.span.index = None + + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_index_outside_total_is_rejected(self): + 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): + record = valid_record() + record.span.start_char_idx = 25 + record.span.end_char_idx = 10 + + with self.assertRaises(ValueError): + ChunkRecordValidator().validate(record) + + def test_invalid_line_range_is_rejected(self): + 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 index 6c85a3565..de2344065 100644 --- a/application/tests/harvester_test/chunker_test.py +++ b/application/tests/harvester_test/chunker_test.py @@ -68,6 +68,52 @@ def test_chunks_preserve_node_boundaries( ], ) + @patch("application.utils.harvester.chunker.HuggingFaceEmbedding") + @patch("application.utils.harvester.chunker.SemanticSplitterNodeParser") + def test_chunking_preserves_node_order( + self, + splitter_cls, + embedding_cls, + ): + first = type( + "Node", + (), + { + "start_char_idx": 0, + "end_char_idx": 6, + "get_content": lambda self: "First.", + }, + )() + + second = type( + "Node", + (), + { + "start_char_idx": 8, + "end_char_idx": 15, + "get_content": lambda self: "Second.", + }, + )() + + splitter_cls.return_value.get_nodes_from_documents.return_value = [ + first, + second, + ] + + chunker = DocumentChunker() + + chunks = chunker.chunk("First.\n\nSecond.") + + self.assertEqual( + [(c.start_char_idx, c.end_char_idx) for c in chunks], + [(0, 6), (8, 15)], + ) + + self.assertEqual( + [c.text for c in chunks], + ["First.", "Second."], + ) + 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..2cd83c27a --- /dev/null +++ b/application/tests/harvester_test/chunking_benchmark_test.py @@ -0,0 +1,46 @@ +import time +import unittest + +from application.utils.harvester.chunker import DocumentChunker + + +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/utils/harvester/chunk_record_validator.py b/application/utils/harvester/chunk_record_validator.py new file mode 100644 index 000000000..d46c2c022 --- /dev/null +++ b/application/utils/harvester/chunk_record_validator.py @@ -0,0 +1,48 @@ +from .models import IngestChunkRecord + + +class ChunkRecordValidator: + """ + Validates RFC-facing ingestion chunk records. + """ + + 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.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 >= 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") From 4f4ffa841516120e6c4a89c86d5a8bba9b06e12c Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Thu, 20 Aug 2026 12:47:55 +0530 Subject: [PATCH 23/27] fix(harvester): address CodeRabbit chunking review --- .../harvester_test/chunk_pipeline_test.py | 36 +++++++++++++++++ .../chunk_record_builder_test.py | 40 +++++++++++++++++++ .../harvester_test/chunking_benchmark_test.py | 5 +++ application/utils/harvester/chunk_pipeline.py | 12 +++++- .../utils/harvester/chunk_record_builder.py | 9 ++++- 5 files changed, 99 insertions(+), 3 deletions(-) create mode 100644 application/tests/harvester_test/chunk_pipeline_test.py 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 index 62d33687d..6e006621a 100644 --- a/application/tests/harvester_test/chunk_record_builder_test.py +++ b/application/tests/harvester_test/chunk_record_builder_test.py @@ -331,6 +331,46 @@ def test_chunk_order_is_deterministic(self): [2, 2], ) + def test_chunk_ids_differ_for_identical_text_at_different_offsets(self): + text = "# Root\n\nRepeated.\n\nRepeated." + document = self._document( + text=text, + headings=[ + HeadingNode( + level=1, + text="Root", + start_line=1, + end_line=5, + ) + ], + ) + + first_start = text.index("Repeated.") + second_start = text.index("Repeated.", first_start + 1) + + chunks = [ + ChunkInfo( + text="Repeated.", + start_char_idx=first_start, + end_char_idx=first_start + len("Repeated."), + ), + ChunkInfo( + text="Repeated.", + start_char_idx=second_start, + end_char_idx=second_start + len("Repeated."), + ), + ] + + records = ChunkRecordBuilder().build( + document, + chunks, + ) + + self.assertNotEqual( + records[0].chunk_id, + records[1].chunk_id, + ) + 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 index 2cd83c27a..31b33615a 100644 --- a/application/tests/harvester_test/chunking_benchmark_test.py +++ b/application/tests/harvester_test/chunking_benchmark_test.py @@ -1,9 +1,14 @@ +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 = ( diff --git a/application/utils/harvester/chunk_pipeline.py b/application/utils/harvester/chunk_pipeline.py index 2283c16ac..340fe50d9 100644 --- a/application/utils/harvester/chunk_pipeline.py +++ b/application/utils/harvester/chunk_pipeline.py @@ -1,4 +1,5 @@ from .chunk_record_builder import ChunkRecordBuilder +from .chunk_record_validator import ChunkRecordValidator from .chunker import DocumentChunker from .models import Document, IngestChunkRecord @@ -6,21 +7,28 @@ class DocumentChunkPipeline: """ Runs semantic chunking followed by structure-aware RFC - chunk-record construction. + chunk-record construction and validation. """ def __init__( self, chunker: DocumentChunker | None = None, record_builder: ChunkRecordBuilder | None = None, + validator: ChunkRecordValidator | None = None, ) -> None: self._chunker = chunker or DocumentChunker() 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) - return self._record_builder.build( + 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 index bc3f9e461..22e303b8d 100644 --- a/application/utils/harvester/chunk_record_builder.py +++ b/application/utils/harvester/chunk_record_builder.py @@ -34,6 +34,8 @@ def build( chunk_id = self._chunk_id( artifact_id=document.artifact_id, heading_path=heading_path, + start_char_idx=chunk.start_char_idx, + end_char_idx=chunk.end_char_idx, content_hash=content_hash, ) @@ -142,8 +144,13 @@ def _line_range( def _chunk_id( artifact_id: str, heading_path: list[str], + start_char_idx: int, + end_char_idx: int, content_hash: str, ) -> str: heading = "/".join(heading_path) - return f"chk:{artifact_id}:{heading}:{content_hash}" + return ( + f"chk:{artifact_id}:{heading}:" + f"{start_char_idx}-{end_char_idx}:{content_hash}" + ) From 84d33b3c34208c6810ada63bdd1671ebcb9ac320 Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Thu, 20 Aug 2026 13:17:36 +0530 Subject: [PATCH 24/27] fix(harvester): adding dependencies to the harvester --- requirements.txt | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/requirements.txt b/requirements.txt index c55834701..53abe35e9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -55,3 +55,8 @@ compliance-trestle posthog typing_extensions + + +llama-index-core +llama-index-embeddings-huggingface +textacy From 6affe36d87ee6ee7ddb386374774bfec1ebf922e Mon Sep 17 00:00:00 2001 From: ParthAggarwal16 Date: Fri, 21 Aug 2026 18:55:04 +0530 Subject: [PATCH 25/27] test: make harvester benchmarks opt-in --- application/tests/harvester_test/chunking_benchmark_test.py | 6 ++---- application/tests/harvester_test/diff_normalizer_test.py | 1 - application/tests/harvester_test/diff_pipeline_test.py | 6 +++++- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/application/tests/harvester_test/chunking_benchmark_test.py b/application/tests/harvester_test/chunking_benchmark_test.py index 31b33615a..711316f2a 100644 --- a/application/tests/harvester_test/chunking_benchmark_test.py +++ b/application/tests/harvester_test/chunking_benchmark_test.py @@ -13,12 +13,10 @@ 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 + "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 + "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 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_pipeline_test.py b/application/tests/harvester_test/diff_pipeline_test.py index cbc3a424f..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,6 +10,10 @@ 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. From f86e5b88f1c7456b6bab28114fed0950f01f28d5 Mon Sep 17 00:00:00 2001 From: Spyros Date: Sat, 29 Aug 2026 18:57:10 +0100 Subject: [PATCH 26/27] =?UTF-8?q?fix(oie):=20hermetic=20A=E2=86=92B?= =?UTF-8?q?=E2=86=92C=20smoke=20and=20orchestrator=20session=20handoff?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Always pass a live SQLAlchemy session into A/B stages, treat Module C's declared NullSafetyGuard degradation as non-fatal, drop LlamaIndex from dev requirements, and document make oie-e2e-smoke. --- Makefile | 6 +- .../utils/oie_orchestrator/pipeline.py | 99 ++++++--- docs/gsoc_2026_module_a/runbook.md | 16 +- requirements-dev.txt | 4 - scripts/run_oie_e2e_smoke.py | 209 ++++++++++++++++++ scripts/run_oie_pipeline.py | 26 +-- 6 files changed, 304 insertions(+), 56 deletions(-) create mode 100644 scripts/run_oie_e2e_smoke.py diff --git a/Makefile b/Makefile index 5d08f100b..1eece7b79 100644 --- a/Makefile +++ b/Makefile @@ -183,7 +183,11 @@ alembic-guardrail: oie-pipeline: [ -d "./venv" ] && . ./venv/bin/activate &&\ - python scripts/run_oie_pipeline.py --cache_file "$(or $(CACHE_FILE),sqlite:///$(CURDIR)/standards_cache.sqlite)" $(OIE_ARGS) + 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 &&\ diff --git a/application/utils/oie_orchestrator/pipeline.py b/application/utils/oie_orchestrator/pipeline.py index 01d816873..df2213825 100644 --- a/application/utils/oie_orchestrator/pipeline.py +++ b/application/utils/oie_orchestrator/pipeline.py @@ -11,7 +11,6 @@ import json import logging -import uuid from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Any, Callable, Dict, List, Optional @@ -42,7 +41,7 @@ def to_dict(self) -> Dict[str, Any]: "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") for s in self.stages), + "ok": all(s.status in ("ok", "skipped", "degraded") for s in self.stages), } def to_json(self) -> str: @@ -57,6 +56,36 @@ def _summary_dict(summary: Any) -> Dict[str, Any]: 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, @@ -80,25 +109,16 @@ def _stage_module_a( fn = run_harvester try: - session: Any = None - if run_harvester_fn is None: - from application import sqla - from application.cmd.cre_main import db_connect - - db_connect(cache_file) - session = sqla.session + session = _connect(cache_file) summary = fn( session, run_id, dry_run=dry_run, sync_repos=sync_repos, ) - status = "ok" - if getattr(summary, "status", "ok") == "degraded": - status = "error" return StageResult( name="module_a_harvester", - status=status, + status=_stage_status_from_summary(summary), detail=f"run_harvester completed for run_id={run_id!r}", summary=_summary_dict(summary), ) @@ -126,7 +146,6 @@ def _stage_module_b( detail="skip_b=True; noise filter not invoked", ) - injected = run_noise_filter_fn is not None fn = run_noise_filter_fn if fn is None: from application.utils.noise_filter.pipeline import run_noise_filter @@ -134,20 +153,11 @@ def _stage_module_b( fn = run_noise_filter try: - session: Any = None - if not injected: - from application import sqla - from application.cmd.cre_main import db_connect - - db_connect(cache_file) - session = sqla.session + session = _connect(cache_file) summary = fn(session, run_id, dry_run=dry_run) - status = "ok" - if getattr(summary, "status", "ok") == "degraded": - status = "error" return StageResult( name="module_b_noise_filter", - status=status, + status=_stage_status_from_summary(summary), detail=f"run_noise_filter completed for run_id={run_id!r}", summary=_summary_dict(summary), ) @@ -177,23 +187,40 @@ def _stage_module_c( 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 run_librarian_live - - # run_librarian_live connects, drains knowledge_queue for run_id, - # writes decision_queue, prints JSON. - run_librarian_live( - cache_file, - pipeline_run_id=run_id, + 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, ) - summary = {"pipeline_run_id": run_id, "dry_run": dry_run} return StageResult( name="module_c_librarian", - status="ok", - detail=f"run_librarian_live completed for run_id={run_id!r}", + 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 ), @@ -203,7 +230,7 @@ def _stage_module_c( return StageResult( name="module_c_librarian", status="error", - detail=f"run_librarian_live failed: {exc}", + detail=f"run_librarian_queue failed: {exc}", ) diff --git a/docs/gsoc_2026_module_a/runbook.md b/docs/gsoc_2026_module_a/runbook.md index 63d78d1c3..e648c67c5 100644 --- a/docs/gsoc_2026_module_a/runbook.md +++ b/docs/gsoc_2026_module_a/runbook.md @@ -21,14 +21,26 @@ Optional: - `--harvester_dry_run` — classify path without inserting rows - `--harvester_repos_yaml PATH` — override `application/utils/harvester/repos.yaml` -Orchestrated: +Orchestrated (A → B → C): ```bash make oie-pipeline OIE_ARGS='--run_id 20260829T020000Z' # or -python scripts/run_oie_pipeline.py --cache_file --run_id +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 diff --git a/requirements-dev.txt b/requirements-dev.txt index c2eebf6a4..7121de20e 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -89,7 +89,3 @@ pytest-playwright # Local stdio MCP server (issue #1003 v1) — not needed on Heroku slug mcp==2.0.0 - -# LlamaIndex -llama-index-core -llama-index-embeddings-huggingface 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 index 8d960afc7..e58a566fb 100644 --- a/scripts/run_oie_pipeline.py +++ b/scripts/run_oie_pipeline.py @@ -44,23 +44,23 @@ def main() -> int: ) args = parser.parse_args() - # Ensure Flask app context for SQLAlchemy. + # 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") - from cre import app # noqa: WPS433 — CLI bootstrap + os.environ.setdefault("NO_LOAD_GRAPH_DB", "1") from application.utils.oie_orchestrator import run_oie_pipeline - with app.app_context(): - 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, - ) + 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 From 8e62ba1282ac0248f23f81e711bd7abb23bd277b Mon Sep 17 00:00:00 2001 From: Spyros Date: Sat, 29 Aug 2026 19:08:38 +0100 Subject: [PATCH 27/27] fix(harvester): import ChunkingConfig for chunk_document helper CI on Python 3.12 evaluated the unquoted annotation at import time and failed the harvester package load. --- application/utils/harvester/chunk_record_builder.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/application/utils/harvester/chunk_record_builder.py b/application/utils/harvester/chunk_record_builder.py index 0bccaed35..8cc292536 100644 --- a/application/utils/harvester/chunk_record_builder.py +++ b/application/utils/harvester/chunk_record_builder.py @@ -1,8 +1,15 @@ +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: @@ -100,7 +107,7 @@ def _line_range( def chunk_document( document: Document, - config: ChunkingConfig | None = None, + config: Optional["ChunkingConfig"] = None, ) -> list[IngestChunkRecord]: chunker = DocumentChunker(config) chunks = chunker.chunk(document.text, document=document)