Task 6. Xác minh replay idempotent

from pathlib import Path
import json
import shutil
import sqlite3
import sys

PROJECT_ROOT = Path.cwd()
if PROJECT_ROOT.name == "lab04-book":
    PROJECT_ROOT = PROJECT_ROOT.parent
SRC_ROOT = PROJECT_ROOT / "src"
if str(SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(SRC_ROOT))

from application.services.process_file import ProcessFileService
from application.services.replay_file import ReplayFileService
from domain.models import SourceFile
from infrastructure.filesystem.git_source_repository import GitSourceRepository
from infrastructure.messaging.event_validator import EventValidator
from infrastructure.messaging.jsonl_event_writer import JsonlEventWriter
from infrastructure.state.sqlite_state_store import SqliteStateStore
from parsing.cpg_parser import CpgParser
from parsing.identifiers import IdentifierGenerator

verification_dir = PROJECT_ROOT / "workspace" / "tmp" / "task6-replay-notebook"
source_root = verification_dir / "source"
output_dir = verification_dir / "events"
state_db = verification_dir / "state.sqlite3"
repo_id = "local/task6-replay-notebook"
target_file = Path("pkg/new_feature.py")
target_path = source_root / target_file

if verification_dir.exists():
    shutil.rmtree(verification_dir)
target_path.parent.mkdir(parents=True)
output_dir.mkdir(parents=True)

repo = GitSourceRepository(source_root, "", None)
parser = CpgParser(repository_id=repo_id)
state_store = SqliteStateStore(state_db, repo_id)
validator = EventValidator(PROJECT_ROOT / "schemas")
writer = JsonlEventWriter(output_dir)
process_service = ProcessFileService(repo, parser, state_store, validator, writer)
replay_service = ReplayFileService(repo, parser, state_store, process_service, repo_id)


def write_source(source: str) -> None:
    target_path.write_text(source, encoding="utf-8")


def make_source_file() -> SourceFile:
    return SourceFile(
        repository_id=repo_id,
        repository_root=str(source_root),
        relative_path=target_file.as_posix(),
        commit_sha=repo.get_commit_hash(),
        size_bytes=target_path.stat().st_size,
    )


def jsonl_count(filename: str) -> int:
    path = output_dir / filename
    if not path.exists():
        return 0
    return sum(1 for line in path.read_text(encoding="utf-8").splitlines() if line.strip())


def jsonl_counts() -> dict[str, int]:
    return {
        "nodes.jsonl": jsonl_count("nodes.jsonl"),
        "edges.jsonl": jsonl_count("edges.jsonl"),
        "metadata.jsonl": jsonl_count("metadata.jsonl"),
        "errors.jsonl": jsonl_count("errors.jsonl"),
    }


def sqlite_state_summary() -> dict[str, object]:
    file_id = IdentifierGenerator.generate_file_id(repo_id, target_file)
    with sqlite3.connect(state_db) as conn:
        rows = conn.execute(
            "SELECT file_id, file_path, content_hash, node_ids_json, edge_ids_json "
            "FROM file_state WHERE file_id = ?",
            (file_id,),
        ).fetchall()
    row = rows[0] if rows else None
    return {
        "row_count": len(rows),
        "file_path": row[1] if row else None,
        "content_hash": row[2] if row else None,
        "node_count": len(json.loads(row[3])) if row else 0,
        "edge_count": len(json.loads(row[4])) if row else 0,
    }

print("verification_dir=", verification_dir)

initial_source = """def normalize(value):
    total = value + 1
    return total
"""
modified_source = """def normalize(value):
    total = value + 1
    if total > 10:
        return total * 2
    return total

class Normalizer:
    def apply(self, value):
        return normalize(value)
"""

write_source(initial_source)
first_result = process_service.execute(make_source_file())
after_first_counts = jsonl_counts()

unchanged_result = process_service.execute(make_source_file())
after_unchanged_counts = jsonl_counts()

write_source(modified_source)
replay_result = replay_service.execute(target_file)
after_replay_counts = jsonl_counts()

run_summary = {
    "first": {
        "status": first_result.status.value,
        "node_count": first_result.node_count,
        "edge_count": first_result.edge_count,
        "emitted_event_counts": first_result.emitted_event_counts,
        "jsonl_counts": after_first_counts,
    },
    "unchanged": {
        "status": unchanged_result.status.value,
        "emitted_event_counts": unchanged_result.emitted_event_counts,
        "jsonl_counts": after_unchanged_counts,
    },
    "replay_after_edit": replay_result,
    "after_replay_jsonl_counts": after_replay_counts,
    "sqlite_state": sqlite_state_summary(),
}
print(json.dumps(run_summary, indent=2, ensure_ascii=False))

assert run_summary["first"]["status"] == "SUCCESS"
assert run_summary["first"]["node_count"] > 0
assert run_summary["first"]["edge_count"] > 0
assert run_summary["first"]["jsonl_counts"]["metadata.jsonl"] == 1

assert run_summary["unchanged"]["status"] == "SKIPPED_UNCHANGED"
assert run_summary["unchanged"]["emitted_event_counts"] == {}
assert run_summary["unchanged"]["jsonl_counts"] == run_summary["first"]["jsonl_counts"]

replay = run_summary["replay_after_edit"]
assert replay["status"] == "SUCCESS"
assert replay["old_content_hash"] != replay["new_content_hash"]
assert replay["removed_node_count"] > 0
assert replay["removed_edge_count"] > 0
assert replay["upsert_node_count"] > run_summary["first"]["node_count"]
assert replay["upsert_edge_count"] > run_summary["first"]["edge_count"]

assert run_summary["after_replay_jsonl_counts"]["errors.jsonl"] == 0
assert run_summary["after_replay_jsonl_counts"]["metadata.jsonl"] == 2
assert run_summary["sqlite_state"]["row_count"] == 1
assert run_summary["sqlite_state"]["file_path"] == "pkg/new_feature.py"
assert run_summary["sqlite_state"]["node_count"] == replay["upsert_node_count"]
assert run_summary["sqlite_state"]["edge_count"] == replay["upsert_edge_count"]

verification_summary = {
    "add_new_python_file": "passed",
    "skip_unchanged_file": "passed",
    "edit_file_replay_diff": "passed",
    "jsonl_error_count": run_summary["after_replay_jsonl_counts"]["errors.jsonl"],
    "sqlite_state_rows": run_summary["sqlite_state"]["row_count"],
    "sqlite_state_path_posix": run_summary["sqlite_state"]["file_path"],
}
print(json.dumps(verification_summary, indent=2, ensure_ascii=False))

# Optional live checklist. This cell is intentionally non-destructive and only reports the
# commands/evidence points used when the Docker stack is available.
from pathlib import Path
import json

LIVE_TARGET_FILE = Path("benchmark/__init__.py")
LIVE_CHECKPOINT_DIR = Path("workspace/checkpoints/task6-live")

live_task6_commands = {
    "start_stack": "docker compose --env-file .env -f infra/docker-compose.yml -f infra/docker-compose.neo4j.yml up -d kafka neo4j kafka-connect mongodb",
    "deploy_connectors": "uv run python scripts/deploy_connectors.py",
    "baseline_publish": f"uv run lab04 parse-file --file {LIVE_TARGET_FILE.as_posix()} --no-dry-run",
    "spark_available_now": "docker run --rm --network infra_default -v ${PWD}:/app -w /app -e MONGODB_URI='mongodb://root:${MONGO_ROOT_PASSWORD}@cpg-mongodb:27017/?authSource=admin' apache/spark:3.5.1 /opt/spark/bin/spark-submit --conf spark.jars.ivy=/tmp/.ivy2 --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1,org.mongodb.spark:mongo-spark-connector_2.12:10.3.0 spark_jobs/metadata_to_mongodb.py --bootstrap-servers kafka:29092 --checkpoint-dir /app/workspace/checkpoints/task6-live --available-now",
    "edit_file": f"Modify cloned repository file transformers-pr-agent/{LIVE_TARGET_FILE.as_posix()} using a small semantic change.",
    "replay_modified": f"uv run lab04 replay-file --file {LIVE_TARGET_FILE.as_posix()} --no-dry-run",
    "neo4j_verify": "uv run python scripts/inspect_neo4j_graph.py --fail-on-duplicates",
    "mongodb_verify": "mongosh mongodb://localhost:27017/cpg --file scripts/verify_mongodb.js",
    "unchanged_rerun": f"uv run lab04 replay-file --file {LIVE_TARGET_FILE.as_posix()} --no-dry-run",
    "spark_checkpoint_resume": "Repeat the same Docker Spark command with checkpoint /app/workspace/checkpoints/task6-live; log shows committed offsets equal available offsets.",
}
print(json.dumps(live_task6_commands, indent=2, ensure_ascii=False))

import json

task6_live_evidence = {
    "target_file": "benchmark/__init__.py",
    "file_id": "896d3ec1720eef15089fe647d75979f1f3ce86cc6741a003826e0bbeb871f5d3",
    "baseline_publish": {
        "status": "SUCCESS",
        "content_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
        "neo4j_node_count": 1,
        "neo4j_edge_count": 0,
        "spark_batch_id": 0,
        "spark_num_input_rows": 1,
        "spark_num_output_rows": 1,
    },
    "modified_replay": {
        "status": "SUCCESS",
        "old_content_hash": "23d0bf2831f046dd2b51078d155f51a18e789e731d131bbfc86fe95e4bfef5f6",
        "new_content_hash": "6629334444aa2f8366cb86f51102a0128e281608f2e85f590bba18d27e50adbc",
        "removed_node_count": 0,
        "removed_edge_count": 1,
        "upsert_node_count": 52,
        "upsert_edge_count": 59,
    },
    "neo4j_after_replay": {
        "node_count": 52,
        "edge_count": 59,
        "duplicate_nodes": [],
        "duplicate_edges": [],
        "duplicate_edge_tombstones": [],
    },
    "mongodb_after_replay": {
        "document_count_by_file_id": 1,
        "document_count_by_posix_path": 1,
        "total_documents": 1,
        "file_path": "benchmark/__init__.py",
        "content_hash": "6629334444aa2f8366cb86f51102a0128e281608f2e85f590bba18d27e50adbc",
        "function_count": 2,
        "class_count": 1,
        "node_count": 52,
        "edge_count": 59,
    },
    "spark_modified_replay_batch": {
        "batch_id": 1,
        "start_offset": {"source.metadata": {"0": 1}},
        "end_offset": {"source.metadata": {"0": 3}},
        "num_input_rows": 2,
        "num_output_rows": 2,
    },
    "unchanged_rerun": {
        "status": "SKIPPED_UNCHANGED",
        "content_hash": "6629334444aa2f8366cb86f51102a0128e281608f2e85f590bba18d27e50adbc",
    },
    "spark_checkpoint_resume": {
        "resuming_batch": 2,
        "committed_offsets": {"source.metadata": {"0": 3}},
        "available_offsets": {"source.metadata": {"0": 3}},
        "new_input_rows": 0,
        "checkpoint_files_present": ["commits/0", "commits/1", "offsets/0", "offsets/1", "sources/0/0"],
    },
}

assert task6_live_evidence["modified_replay"]["status"] == "SUCCESS"
assert task6_live_evidence["neo4j_after_replay"]["node_count"] == task6_live_evidence["modified_replay"]["upsert_node_count"]
assert task6_live_evidence["neo4j_after_replay"]["edge_count"] == task6_live_evidence["modified_replay"]["upsert_edge_count"]
assert task6_live_evidence["neo4j_after_replay"]["duplicate_nodes"] == []
assert task6_live_evidence["neo4j_after_replay"]["duplicate_edges"] == []
assert task6_live_evidence["mongodb_after_replay"]["document_count_by_file_id"] == 1
assert task6_live_evidence["mongodb_after_replay"]["document_count_by_posix_path"] == 1
assert task6_live_evidence["mongodb_after_replay"]["content_hash"] == task6_live_evidence["modified_replay"]["new_content_hash"]
assert task6_live_evidence["unchanged_rerun"]["status"] == "SKIPPED_UNCHANGED"
assert task6_live_evidence["spark_checkpoint_resume"]["committed_offsets"] == task6_live_evidence["spark_checkpoint_resume"]["available_offsets"]
assert task6_live_evidence["spark_checkpoint_resume"]["new_input_rows"] == 0

print(json.dumps(task6_live_evidence, indent=2, ensure_ascii=False))