Skip to content

GSoC Module A : Week 3: feat(harvester): implement incremental change detection (stacked on #983) #985

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ParthAggarwal16 wants to merge 11 commits into
base: main
Choose a base branch
from
Open

GSoC Module A : Week 3: feat(harvester): implement incremental change detection (stacked on #983) #985

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
5cab1cf
fix(harvester): address repository client review feedback
ParthAggarwal16 Jul 18, 2026
2f981e2
feat(harvester): implement incremental change detection
ParthAggarwal16 Jul 8, 2026
eb72939
fix(harvester): write checkpoints atomically
ParthAggarwal16 Jul 18, 2026
877bf7a
Use immutable commit ranges for change detection
ParthAggarwal16 Jul 23, 2026
967f146
test: replace mocked /tmp/repo path
ParthAggarwal16 Jul 23, 2026
bb90b14
feat(db): add artifact ingestion persistence models
ParthAggarwal16 Jul 23, 2026
ab615d5
Address CodeRabbit review feedback
ParthAggarwal16 Jul 23, 2026
a6929df
feat(harvester): replace filesystem checkpoint store with postgres
ParthAggarwal16 Jul 25, 2026
92f7f0d
test(harvester): match checkout invocation
ParthAggarwal16 Jul 25, 2026
189d84b
Address review feedback for Week 3
ParthAggarwal16 Jul 26, 2026
9819313
docs : fix Revision ID in the migration file
ParthAggarwal16 Jul 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
165 changes: 165 additions & 0 deletions application/database/db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import time
import yaml

from datetime import datetime, timezone
from pprint import pprint

from collections import Counter, defaultdict
Expand Down Expand Up @@ -257,6 +258,108 @@ class StagedChangeSet(BaseModel): # type: ignore
created_at = sqla.Column(sqla.DateTime, nullable=False)


class ArtifactIngestEvent(BaseModel): # type: ignore
"""Tracks one harvested artifact persisted per import run."""

__tablename__ = "artifact_ingest_event"
id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
run_id = sqla.Column(
sqla.String,
sqla.ForeignKey("import_run.id", onupdate="CASCADE", ondelete="CASCADE"),
nullable=False,
)
artifact_id = sqla.Column(sqla.String, nullable=False)
harvest_mode = sqla.Column(sqla.String, nullable=False)
event_type = sqla.Column(sqla.String, nullable=False)
source_json = sqla.Column(sqla.Text, nullable=False)
locator_json = sqla.Column(sqla.Text, nullable=False)
artifact_json = sqla.Column(sqla.Text, nullable=False)
harvest_json = sqla.Column(sqla.Text, nullable=False)
observed_at = sqla.Column(sqla.DateTime, nullable=False)
created_at = sqla.Column(sqla.DateTime, nullable=False)

__table_args__ = (
sqla.UniqueConstraint(
run_id,
artifact_id,
name="uq_artifact_ingest_event_run_artifact",
),
)


class IngestChunk(BaseModel): # type: ignore
"""Tracks every chunk belonging to an artifact ingest event."""

__tablename__ = "ingest_chunk"
id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
artifact_event_id = sqla.Column(
sqla.String,
sqla.ForeignKey(
"artifact_ingest_event.id",
onupdate="CASCADE",
ondelete="CASCADE",
),
nullable=False,
)
chunk_id = sqla.Column(sqla.String, nullable=False)
text = sqla.Column(sqla.Text, nullable=False)
char_count = sqla.Column(sqla.Integer, nullable=False)
span_json = sqla.Column(sqla.Text, nullable=False)
delta_json = sqla.Column(sqla.Text, nullable=True)
created_at = sqla.Column(sqla.DateTime, nullable=False)

__table_args__ = (
sqla.UniqueConstraint(
artifact_event_id,
chunk_id,
name="uq_ingest_chunk_artifact_chunk",
),
)


class HarvesterCheckpoint(BaseModel): # type: ignore
__tablename__ = "harvester_checkpoint"
repository_id = sqla.Column(sqla.String, primary_key=True)
provider = sqla.Column(sqla.String, nullable=False)
owner = sqla.Column(sqla.String, nullable=False)
repository = sqla.Column(sqla.String, nullable=False)
branch = sqla.Column(sqla.String, nullable=False)
last_processed_commit = sqla.Column(sqla.String, nullable=True)
created_at = sqla.Column(
sqla.DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
updated_at = sqla.Column(
sqla.DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
__table_args__ = (
sqla.UniqueConstraint(
"provider",
"owner",
"repository",
"branch",
name="uq_harvester_checkpoint_canonical_source",
),
)


def _serialize_json_value(value: Any) -> str:
return flask_json.dumps(value)
Comment thread
coderabbitai[bot] marked this conversation as resolved.


def _normalize_utc_datetime(value: Any) -> Any:
from datetime import datetime, timezone

if isinstance(value, datetime):
if value.tzinfo is None:
return value
return value.astimezone(timezone.utc)
return value
Comment on lines +353 to +360

@coderabbitai coderabbitai Bot Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "Files:"
git ls-files | grep -E '(^|/)application/database/db\.py$|(^|/)application/database/|gap-analysis|harvester|observations|import' || true

echo
echo "Relevant db.py outline:"
if [ -f application/database/db.py ]; then
  ast-grep outline application/database/db.py --match _normalize_utc_datetime --view expanded || true
  ast-grep outline application/database/db.py --match create_import_run --view expanded || true
  wc -l application/database/db.py
  echo
  echo "Relevant lines 240-350:"
  sed -n '240,350p' application/database/db.py | nl -ba -v240
  echo
  echo "Search observed_at/create_import_run/normalize_utc:"
  rg -n "observed_at|created_at|_normalize_utc_datetime|create_import_run|astimezone|timezone=True|datetime\\(" application/database/db.py
fi

echo
echo "Search local timezone/utc assumptions in repo:"
rg -n "observed_at|created_at|astimezone\\(|timezone\\(\\)|UTC|naive|timezone-aware|utc" -g '*.py' -g '!*frontend*' .

Repository: OWASP/OpenCRE

Length of output: 3770

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "Lines 240-350:"
sed -n '240,350p' application/database/db.py

echo
echo "All normalize_utc_datetime usages:"
python3 - <<'PY'
from pathlib import Path
p = Path('application/database/db.py')
text = p.read_text()
for i,line in enumerate(text.splitlines(), start=1):
    if '_normalize_utc_datetime' in line or 'observed_at' in line or 'created_at' in line or 'create_import_run' in line:
        print(f"{i}: {line}")
PY

echo
echo "All observed_at/create_import_run usages outside db.py:"
rg -n "observed_at|created_at|create_import_run|_normalize_utc_datetime" -g '*.py' application migrations scripts docs

Repository: OWASP/OpenCRE

Length of output: 11277

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "db.py lines 378-415:"
sed -n '378,415p' application/database/db.py

echo
echo "Harvester observed_at references:"
rg -n "observed_at|datetime.now|datetime.utcnow|timezone\\(|utc" application/utils/harvester -g '*.py'

echo
echo "Tests around record_artifact_event observed_at:"
sed -n '35,75p' application/tests/import_run_test.py

Repository: OWASP/OpenCRE

Length of output: 1204

Reject naive datetimes in UTC normalization.

artifact_ingest_event.observed_at is normalized through _normalize_utc_datetime, but naïve inputs are returned unchanged and persisted without validation. The public ingestion path currently passes tz-aware datetimes, so this is mainly about a future caller/local timestamp silently bypassing conversion. Raise an explicit error for naive inputs or document the UTC assumption, and consider sqla.DateTime(timezone=True) for timestamp columns so offset metadata is preserved.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@application/database/db.py` around lines 324 - 331, Update
_normalize_utc_datetime to reject datetime values without tzinfo by raising an
explicit error instead of returning them unchanged; continue converting
timezone-aware values to UTC and leaving non-datetime values unchanged. Use
timezone-aware SQLAlchemy DateTime columns for persisted timestamps if needed to
preserve offset metadata.

@ParthAggarwal16 ParthAggarwal16 Jul 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the RFC doesnt say, the ingestion pipeline already supplies timezone-aware UTC timestamps,
@northdpole @Pa04rth would like to know your thoughts on this one before i do it



def create_import_run(source: str, version: Optional[str] = None) -> ImportRun:
"""Create and persist an import run record. Returns the new ImportRun."""
from datetime import datetime, timezone
Expand Down Expand Up @@ -304,6 +407,68 @@ def get_previous_import_run(source: str, current_run_id: str) -> Optional[Import
)


def create_artifact_ingest_event(
*,
run_id: str,
artifact_id: str,
harvest_mode: str,
event_type: str,
source_json: Any,
locator_json: Any,
artifact_json: Any,
harvest_json: Any,
observed_at: Any,
) -> ArtifactIngestEvent:
from datetime import datetime, timezone

observed_at = _normalize_utc_datetime(observed_at)

event = ArtifactIngestEvent(
id=generate_uuid(),
run_id=run_id,
artifact_id=artifact_id,
harvest_mode=harvest_mode,
event_type=event_type,
source_json=_serialize_json_value(source_json),
locator_json=_serialize_json_value(locator_json),
artifact_json=_serialize_json_value(artifact_json),
harvest_json=_serialize_json_value(harvest_json),
observed_at=observed_at,
created_at=_normalize_utc_datetime(datetime.now(timezone.utc)),
)
sqla.session.add(event)
sqla.session.commit()
return event


def create_ingest_chunk(
*,
artifact_event_id: str,
chunk_id: str,
text: str,
char_count: int,
span_json: Any,
delta_json: Optional[Any] = None,
) -> IngestChunk:
from datetime import datetime, timezone

chunk = IngestChunk(
id=generate_uuid(),
artifact_event_id=artifact_event_id,
chunk_id=chunk_id,
text=text,
char_count=char_count,
span_json=_serialize_json_value(span_json),
delta_json=(
_serialize_json_value(delta_json) if delta_json is not None else None
),
created_at=_normalize_utc_datetime(datetime.now(timezone.utc)),
)
sqla.session.add(chunk)
sqla.session.commit()
return chunk
Comment on lines +410 to +469

@coderabbitai coderabbitai Bot Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

No rollback on commit failure for either persistence helper.

Both create_artifact_ingest_event and create_ingest_chunk call sqla.session.add() followed directly by sqla.session.commit() with no try/except. A unique-constraint violation (duplicate run_id+artifact_id, or duplicate artifact_event_id+chunk_id — plausible on retries in an incremental harvester) raises an uncaught IntegrityError and leaves the session in a failed state, breaking subsequent operations on the same session until an explicit rollback() is issued.

🛠️ Suggested pattern (applies to both functions) 
     sqla.session.add(event)
-    sqla.session.commit()
+    try:
+        sqla.session.commit()
+    except Exception:
+        sqla.session.rollback()
+        raise
     return event
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@application/database/db.py` around lines 381 - 440, Wrap the add/commit flow
in both create_artifact_ingest_event and create_ingest_chunk with exception
handling that rolls back sqla.session when commit fails, then re-raise the
original exception so callers retain the failure behavior while the session is
usable for subsequent operations.



def persist_standard_snapshot(
*,
run_id: str,
Expand Down
168 changes: 168 additions & 0 deletions application/tests/harvester_test/change_detector_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,168 @@
import unittest
from unittest.mock import MagicMock
from unittest.mock import call
from unittest.mock import patch

from application.utils.harvester.change_detector import (
ChangeDetector,
)


class ChangeDetectorTests(unittest.TestCase):
@patch("application.utils.harvester.change_detector.subprocess.run")
def test_get_modified_files_since(self, mock_run):
client = MagicMock()
client.get_local_path.return_value = "repo-under-test"

mock_run.side_effect = [
MagicMock(stdout="resolved_base\n"),
MagicMock(stdout="resolved_target\n"),
MagicMock(stdout="a.md\nb.md\na.md\n"),
]

detector = ChangeDetector(client)

files = detector.get_modified_files_since(
"base",
"target",
)

self.assertEqual(
files,
[
"a.md",
"b.md",
],
)

mock_run.assert_has_calls(
[
call(
[
"git",
"-C",
"repo-under-test",
"rev-parse",
"--verify",
"--end-of-options",
"base^{commit}",
],
capture_output=True,
text=True,
check=True,
timeout=60,
),
call(
[
"git",
"-C",
"repo-under-test",
"rev-parse",
"--verify",
"--end-of-options",
"target^{commit}",
],
capture_output=True,
text=True,
check=True,
timeout=60,
),
call(
[
"git",
"-C",
"repo-under-test",
"diff",
"--name-only",
"resolved_base",
"resolved_target",
],
capture_output=True,
text=True,
check=True,
timeout=60,
),
]
)

@patch("application.utils.harvester.change_detector.subprocess.run")
def test_get_commits_since(self, mock_run):
client = MagicMock()
client.get_local_path.return_value = "repo-under-test"

mock_run.side_effect = [
MagicMock(stdout="resolved_base\n"),
MagicMock(stdout="resolved_target\n"),
MagicMock(stdout="111\n222\n333\n"),
]

detector = ChangeDetector(client)

commits = detector.get_commits_since(
"base",
"target",
)

self.assertEqual(
commits,
[
"111",
"222",
"333",
],
)

self.assertEqual(
mock_run.call_args_list,
[
call(
[
"git",
"-C",
"repo-under-test",
"rev-parse",
"--verify",
"--end-of-options",
"base^{commit}",
],
capture_output=True,
text=True,
check=True,
timeout=60,
),
call(
[
"git",
"-C",
"repo-under-test",
"rev-parse",
"--verify",
"--end-of-options",
"target^{commit}",
],
capture_output=True,
text=True,
check=True,
timeout=60,
),
call(
[
"git",
"-C",
"repo-under-test",
"log",
"--reverse",
"--format=%H",
"resolved_base..resolved_target",
],
capture_output=True,
text=True,
check=True,
timeout=60,
),
],
)


if __name__ == "__main__":
unittest.main()
Loading
Loading