From a37524587d620179b2b85ee51fd4484c31a00da2 Mon Sep 17 00:00:00 2001 From: Shaggi Date: Wed, 29 Jul 2026 12:56:40 +0300 Subject: [PATCH] feat: build exploratory training datasets --- benchmarks/real_world/README.md | 60 ++ benchmarks/real_world/_secure_publish.py | 248 +++++++ .../real_world/build_training_dataset.py | 645 ++++++++++++++++++ benchmarks/real_world/collect_prs.py | 449 +++++++++++- .../benchmarks/test_build_training_dataset.py | 589 ++++++++++++++++ tests/benchmarks/test_collect_prs.py | 338 +++++++++ 6 files changed, 2293 insertions(+), 36 deletions(-) create mode 100644 benchmarks/real_world/_secure_publish.py create mode 100644 benchmarks/real_world/build_training_dataset.py create mode 100644 tests/benchmarks/test_build_training_dataset.py create mode 100644 tests/benchmarks/test_collect_prs.py diff --git a/benchmarks/real_world/README.md b/benchmarks/real_world/README.md index 63e2a7a..1132558 100644 --- a/benchmarks/real_world/README.md +++ b/benchmarks/real_world/README.md @@ -22,6 +22,66 @@ Third-party source and patches are fetched on demand and are not vendored. Re-running the collector creates a new corpus; it must not silently replace the frozen benchmark used in a published comparison. +## Three workflows (do not use the production path for data generation) + +There are three different goals here, and they do not need the same process: + +1. **Exploratory training data**: collect PR metadata, obtain diffs, and join + whatever completed labels we have. This is the cheap path. It does not run + the production custody/ledger workflow, does not write canonical truth, and + may use one reviewed label as an explicitly exploratory training target. +2. **Analyzer comparison**: run `run_current.py` and score with `evaluate.py` + against a deliberately frozen and fully adjudicated corpus. +3. **Publication-grade ground truth**: use the pilot/production protocols only + when we need an auditable public claim, not just examples for fine-tuning. + +The first path is intentionally small and ordinary. Its secure no-clobber +publisher currently requires Linux/POSIX filesystem semantics (directory file +descriptors, no-follow opens, and hard links), matching current CI coverage. +Use real, non-symlinked working and cache directories for these exploratory +scripts; no cross-platform fallback is provided. + +To use the frozen corpus, join it to its identity-matched adjudicated labels: + +```bash +.venv/bin/python benchmarks/real_world/build_training_dataset.py \ + --corpus benchmarks/real_world/corpus.json \ + --labels benchmarks/real_world/adjudicated.jsonl \ + --diff-dir /tmp/blast-radius-frozen-diffs --fetch-diffs \ + --output /tmp/blast-radius-frozen-train.jsonl +``` + +For a fresh collection, use a separate completed review JSONL produced for the +same repository/PR identities. Do not join a latest-N collection to the frozen +historical labels merely because both files use the same format: + +```bash +# Pick a fresh output; the collector defaults to candidate-corpus.json and is +# no-clobber, so corpus.json remains the frozen artifact. +.venv/bin/python benchmarks/real_world/collect_prs.py \ + --repository open-webui/open-webui --repository langflow-ai/langflow \ + --limit 20 --output /tmp/blast-radius-corpus.json + +.venv/bin/python benchmarks/real_world/build_training_dataset.py \ + --corpus /tmp/blast-radius-corpus.json \ + --labels /tmp/completed-labels-for-blast-radius-corpus.jsonl \ + --diff-dir /tmp/blast-radius-fresh-diffs --fetch-diffs \ + --output /tmp/blast-radius-fresh-train.jsonl +``` + +Only identity-matched completed labels produce examples. Corpus identities with +no completed matching label are skipped and reported in `missing_labels`; they +are never treated as negatives. + +`build_training_dataset.py` emits one JSON object per PR and a small `target` +containing the claims present in the validated completed label. Its `diff` is +nullable unless the corpus embeds it, the cache already contains it, or +`--fetch-diffs` fetches and exclusively publishes it. Use `--scope all` when +training a multi-surface model; the default `fastapi` scope keeps only +HTTP/WebSocket claims addressable by this tool. The output is not a score, and a +single-review or exploratory set must not be presented as canonical benchmark +truth. + ## Frozen 50-project expansion Issue #103 adds a disjoint, metadata-only expansion with 50 projects and 100 diff --git a/benchmarks/real_world/_secure_publish.py b/benchmarks/real_world/_secure_publish.py new file mode 100644 index 0000000..80a0f0f --- /dev/null +++ b/benchmarks/real_world/_secure_publish.py @@ -0,0 +1,248 @@ +"""Private race-safe file publication helpers for exploratory benchmark tools.""" + +from __future__ import annotations + +import errno +import os +import secrets +import stat +from contextlib import suppress +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable + + +class SecurePathError(ValueError): + """Raised when a path cannot be accessed without following aliases.""" + + +def _absolute(path: Path) -> Path: + # Resolving would follow precisely the symlinks this helper must reject. + return Path(os.path.abspath(path.expanduser())) # noqa: PTH100 + + +def _identity_from_stat(value: os.stat_result) -> tuple[int, int]: + return value.st_dev, value.st_ino + + +def _existing_identities(paths: Iterable[Path]) -> set[tuple[int, int]]: + identities: set[tuple[int, int]] = set() + for path in paths: + try: + identities.add(_identity_from_stat(os.stat(path))) # noqa: PTH116 + except FileNotFoundError: + continue + except OSError as error: + raise SecurePathError(f"cannot inspect protected path {path}: {error}") from error + return identities + + +def _open_parent( + path: Path, + *, + create: bool, + forbidden_roots: tuple[Path, ...] = (), +) -> tuple[int, str, Path] | None: + absolute = _absolute(path) + if not absolute.name: + raise SecurePathError(f"destination must name a file: {path}") + root_identities = _existing_identities(forbidden_roots) + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW + descriptor = os.open(absolute.anchor, flags) + try: + if _identity_from_stat(os.fstat(descriptor)) in root_identities: + raise SecurePathError(f"path is inside a protected directory: {path}") + for component in absolute.parent.parts[1:]: + try: + child = os.open(component, flags, dir_fd=descriptor) + except FileNotFoundError: + if not create: + return None + with suppress(FileExistsError): + os.mkdir(component, dir_fd=descriptor) + child = os.open(component, flags, dir_fd=descriptor) + except OSError as error: + if error.errno in {errno.ELOOP, errno.ENOTDIR}: + raise SecurePathError( + f"path contains a symlink or non-directory component: {path}" + ) from error + raise + os.close(descriptor) + descriptor = child + if _identity_from_stat(os.fstat(descriptor)) in root_identities: + raise SecurePathError(f"path is inside a protected directory: {path}") + return descriptor, absolute.name, absolute + except BaseException: + os.close(descriptor) + raise + + +def _forbidden_file_addresses(paths: Iterable[Path]) -> set[tuple[int, int, str]]: + addresses: set[tuple[int, int, str]] = set() + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW + for path in paths: + absolute = _absolute(path) + try: + descriptor = os.open(absolute.parent, flags) + except FileNotFoundError: + continue + except OSError as error: + raise SecurePathError(f"cannot inspect protected path {path}: {error}") from error + try: + device, inode = _identity_from_stat(os.fstat(descriptor)) + addresses.add((device, inode, absolute.name)) + finally: + os.close(descriptor) + return addresses + + +def ensure_publishable( + destination: Path, + *, + forbidden_files: tuple[Path, ...] = (), + forbidden_roots: tuple[Path, ...] = (), +) -> None: + """Check a destination without following symlinked path components.""" + try: + opened = _open_parent(destination, create=False, forbidden_roots=forbidden_roots) + except OSError as error: + raise SecurePathError(f"cannot inspect destination {destination}: {error}") from error + if opened is None: + return + descriptor, name, _absolute_path = opened + try: + device, inode = _identity_from_stat(os.fstat(descriptor)) + if (device, inode, name) in _forbidden_file_addresses(forbidden_files): + raise SecurePathError(f"refusing to target protected file: {destination}") + try: + os.stat(name, dir_fd=descriptor, follow_symlinks=False) + except FileNotFoundError: + return + raise SecurePathError(f"destination already exists: {destination}") + except OSError as error: + raise SecurePathError(f"cannot inspect destination {destination}: {error}") from error + finally: + os.close(descriptor) + + +def publish_exclusive_bytes( # noqa: PLR0912, PLR0915 + destination: Path, + content: bytes, + *, + forbidden_files: tuple[Path, ...] = (), + forbidden_roots: tuple[Path, ...] = (), +) -> None: + """Publish bytes exclusively through one stable destination-directory FD.""" + try: + opened = _open_parent(destination, create=True, forbidden_roots=forbidden_roots) + assert opened is not None + directory_fd, name, _absolute_path = opened + temporary_name: str | None = None + file_fd: int | None = None + try: + device, inode = _identity_from_stat(os.fstat(directory_fd)) + if (device, inode, name) in _forbidden_file_addresses(forbidden_files): + raise SecurePathError(f"refusing to target protected file: {destination}") + try: + os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError: + pass + else: + raise SecurePathError(f"destination already exists: {destination}") + + for _attempt in range(100): + candidate = f".{name}.{secrets.token_hex(12)}.tmp" + try: + file_fd = os.open( + candidate, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW, + 0o600, + dir_fd=directory_fd, + ) + except FileExistsError: + continue + temporary_name = candidate + break + if file_fd is None or temporary_name is None: + raise SecurePathError(f"cannot allocate temporary file for {destination}") + + view = memoryview(content) + while view: + written = os.write(file_fd, view) + if written == 0: + raise OSError("short write while publishing") + view = view[written:] + os.fsync(file_fd) + os.close(file_fd) + file_fd = None + try: + os.link( + temporary_name, + name, + src_dir_fd=directory_fd, + dst_dir_fd=directory_fd, + follow_symlinks=False, + ) + except FileExistsError as error: + raise SecurePathError(f"destination already exists: {destination}") from error + os.unlink(temporary_name, dir_fd=directory_fd) + temporary_name = None + os.fsync(directory_fd) + finally: + if file_fd is not None: + os.close(file_fd) + if temporary_name is not None: + with suppress(FileNotFoundError): + os.unlink(temporary_name, dir_fd=directory_fd) + os.close(directory_fd) + except SecurePathError: + raise + except OSError as error: + raise SecurePathError(f"cannot publish {destination}: {error}") from error + + +def read_secure_regular_file( + path: Path, + *, + forbidden_files: tuple[Path, ...] = (), +) -> bytes | None: + """Read a regular single-link file without following any symlink component.""" + try: + opened = _open_parent(path, create=False) + if opened is None: + return None + directory_fd, name, _absolute_path = opened + file_fd: int | None = None + try: + try: + file_fd = os.open( + name, + os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, + dir_fd=directory_fd, + ) + except FileNotFoundError: + return None + info = os.fstat(file_fd) + if not stat.S_ISREG(info.st_mode): + raise SecurePathError(f"cache is not a regular file: {path}") + if info.st_nlink != 1: + raise SecurePathError(f"cache must have exactly one hard link: {path}") + if _identity_from_stat(info) in _existing_identities(forbidden_files): + raise SecurePathError(f"cache aliases a protected or input file: {path}") + chunks: list[bytes] = [] + while True: + chunk = os.read(file_fd, 1024 * 1024) + if not chunk: + break + chunks.append(chunk) + return b"".join(chunks) + finally: + if file_fd is not None: + os.close(file_fd) + os.close(directory_fd) + except SecurePathError: + raise + except OSError as error: + raise SecurePathError(f"cannot read cache {path}: {error}") from error diff --git a/benchmarks/real_world/build_training_dataset.py b/benchmarks/real_world/build_training_dataset.py new file mode 100644 index 0000000..a224aad --- /dev/null +++ b/benchmarks/real_world/build_training_dataset.py @@ -0,0 +1,645 @@ +#!/usr/bin/env python3 +"""Build inexpensive exploratory fine-tuning data from PR metadata and labels. + +This is deliberately separate from publication-grade blind-review and production +custody workflows. It joins an existing corpus with completed legacy review JSONL +or a v2 compatibility projection, optionally fetches PR diffs, and emits ordinary +JSONL examples. It never writes canonical truth or treats incomplete labels as +negative examples. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from pathlib import Path +from typing import Any, cast +from urllib.error import URLError +from urllib.request import Request, urlopen + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from benchmarks.real_world._secure_publish import ( + SecurePathError, + ensure_publishable, + publish_exclusive_bytes, + read_secure_regular_file, +) +from benchmarks.real_world.benchmark_schema import BenchmarkSchemaError, strict_json_loads +from benchmarks.real_world.benchmark_scope import SCOPES, filter_record + +HERE = Path(__file__).resolve().parent +DEFAULT_OUTPUT = HERE / "blast-radius-dataset.jsonl" +MAX_DIFF_BYTES = 8 * 1024 * 1024 + +TARGET_FIELDS = ( + "status", + "affected_entrypoints", + "changed_symbols", + "affected_tests", + "contract_changes", + "unknowns", + "orphans", +) +_LIST_TARGET_FIELDS = TARGET_FIELDS[1:] +_LEGACY_REQUIRED_LIST_FIELDS = (*_LIST_TARGET_FIELDS, "cross_repository_consumers") +_COMPLETED_LEGACY_STATUSES = {"reviewed", "adjudicated"} +_SKIPPABLE_STATUSES = {"pending", "pending_double_review", "unknown", "not_evaluable"} +_ALLOWED_CONFIDENCE = {"confirmed", "probable", "possible"} +_ALLOWED_KINDS = {"http", "graphql", "task", "event", "cli", "cron", "sdk", "other"} +_PROTECTED_ARTIFACTS = ( + HERE / "corpus.json", + HERE / "review-a.jsonl", + HERE / "review-b.jsonl", + HERE / "adjudicated.jsonl", + HERE / "adjudication-amendments.jsonl", + HERE / "reachability-supplements.jsonl", + HERE / "review-queue.json", +) +_PROTECTED_ROOTS = tuple( + HERE / name + for name in ( + "expansion", + "ground_truth_v2", + "pilot_v2", + "pilot_v3", + "production_v1", + "scopes", + "verification_sets", + ) +) + + +class DatasetError(ValueError): + """Raised when corpus or label data cannot form a safe training example.""" + + +def _strict_json(content: str, source: str) -> Any: + try: + return strict_json_loads(content, source) + except BenchmarkSchemaError as error: + raise DatasetError(str(error)) from error + + +def _protected_files(input_paths: tuple[Path, ...]) -> tuple[Path, ...]: + return (*_PROTECTED_ARTIFACTS, *input_paths) + + +def _validate_destination(path: Path, *, input_paths: tuple[Path, ...] = ()) -> None: + absolute = Path(path).expanduser().absolute() + if absolute in {item.expanduser().absolute() for item in input_paths}: + raise DatasetError(f"refusing to overwrite input file: {path}") + if absolute in {item.expanduser().absolute() for item in _PROTECTED_ARTIFACTS} or any( + absolute.is_relative_to(root.expanduser().absolute()) for root in _PROTECTED_ROOTS + ): + raise DatasetError(f"refusing to target frozen benchmark artifact: {path}") + try: + ensure_publishable( + path, + forbidden_files=_protected_files(input_paths), + forbidden_roots=_PROTECTED_ROOTS, + ) + except SecurePathError as error: + raise DatasetError(str(error)) from error + + +def _publish_exclusive_bytes( + destination: Path, + content: bytes, + *, + input_paths: tuple[Path, ...] = (), +) -> None: + """Durably publish exact bytes through a stable destination-directory FD.""" + try: + publish_exclusive_bytes( + destination, + content, + forbidden_files=_protected_files(input_paths), + forbidden_roots=_PROTECTED_ROOTS, + ) + except SecurePathError as error: + raise DatasetError(str(error)) from error + + +def record_key(record: dict[str, Any]) -> tuple[str, int]: + repository = record.get("repository") + if ( + "pr" in record + and "number" in record + and (type(record["pr"]) is not type(record["number"]) or record["pr"] != record["number"]) + ): + raise DatasetError(f"record has conflicting pr and number aliases for {repository}") + raw_pr = record.get("pr", record.get("number")) + if not isinstance(repository, str) or not repository.strip(): + raise DatasetError("record has no repository") + if type(raw_pr) is not int or raw_pr < 1: + raise DatasetError(f"record has invalid PR number for {repository}") + return repository, raw_pr + + +def read_jsonl(path: Path) -> list[dict[str, Any]]: + try: + lines = path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeError) as error: + raise DatasetError(f"cannot read {path}: {error}") from error + + records: list[dict[str, Any]] = [] + for line_number, line in enumerate(lines, start=1): + if not line.strip(): + continue + value = _strict_json(line, f"{path}:{line_number}") + if not isinstance(value, dict): + raise DatasetError(f"{path}:{line_number} must contain a JSON object") + records.append(value) + return records + + +def load_corpus(path: Path) -> list[dict[str, Any]]: + try: + content = path.read_text(encoding="utf-8") + except (OSError, UnicodeError) as error: + raise DatasetError(f"cannot read corpus {path}: {error}") from error + value = _strict_json(content, str(path)) + if not isinstance(value, dict) or not isinstance(value.get("entries"), list): + raise DatasetError("corpus must be an object with an entries list") + + entries: list[dict[str, Any]] = [] + seen: set[tuple[str, int]] = set() + for entry in value["entries"]: + if not isinstance(entry, dict): + raise DatasetError("corpus entries must be objects") + key = record_key(entry) + merge_sha(entry) + if key in seen: + raise DatasetError(f"duplicate corpus entry: {key[0]}#{key[1]}") + seen.add(key) + entries.append(entry) + return entries + + +def _validate_string_list(label: dict[str, Any], field: str, identity: str) -> None: + value = label.get(field) + if not isinstance(value, list) or any( + not isinstance(item, str) or not item.strip() for item in value + ): + raise DatasetError(f"completed label {identity} requires {field} as a string list") + + +def _validate_entrypoints( + value: object, + identity: str, + *, + require_evidence: bool, +) -> list[dict[str, Any]]: + if not isinstance(value, list): + raise DatasetError(f"completed label {identity} requires affected_entrypoints as a list") + seen: set[str] = set() + for index, item in enumerate(value): + location = f"completed label {identity} affected_entrypoints[{index}]" + if not isinstance(item, dict): + raise DatasetError(f"{location} must be an object") + entrypoint_id = item.get("id") + if not isinstance(entrypoint_id, str) or not entrypoint_id.strip(): + raise DatasetError(f"{location}.id must be a non-empty string") + if entrypoint_id in seen: + raise DatasetError(f"completed label {identity} has duplicate entrypoint id") + seen.add(entrypoint_id) + if item.get("kind") not in _ALLOWED_KINDS: + raise DatasetError(f"{location}.kind is invalid") + if item.get("confidence") not in _ALLOWED_CONFIDENCE: + raise DatasetError(f"{location}.confidence is invalid") + if require_evidence: + evidence = item.get("evidence") + if ( + not isinstance(evidence, list) + or not evidence + or any( + not isinstance(fragment, str) or not fragment.strip() for fragment in evidence + ) + ): + raise DatasetError(f"{location}.evidence must contain non-empty strings") + return value + + +def _validate_reviewer(label: dict[str, Any], identity: str) -> None: + reviewer = label.get("reviewer") + if not isinstance(reviewer, dict): + raise DatasetError(f"completed label {identity} requires reviewer provenance") + if reviewer.get("kind") not in {"agent", "human"}: + raise DatasetError(f"completed label {identity} reviewer.kind is invalid") + for field in ("name", "version"): + value = reviewer.get(field) + if not isinstance(value, str) or not value.strip(): + raise DatasetError(f"completed label {identity} reviewer.{field} is required") + + +def classify_label(label: dict[str, Any]) -> tuple[str, str | None]: + """Classify a label as completed or explicitly skippable, rejecting ambiguity.""" + status = label.get("status") + if not isinstance(status, str): + raise DatasetError("label status must be a string") + has_terminal = "terminal_status" in label + terminal_status = label.get("terminal_status") + if has_terminal and not isinstance(terminal_status, str): + raise DatasetError("label terminal_status must be a string when present") + + if not has_terminal and status in _COMPLETED_LEGACY_STATUSES: + return "completed", status + if status == "adjudicated" and terminal_status in {"positive", "negative_control"}: + return "completed", cast("str", terminal_status) + if not has_terminal and status in _SKIPPABLE_STATUSES: + return "skippable", None + if status in {"unknown", "not_evaluable"} and terminal_status == status: + return "skippable", None + supported = sorted(_COMPLETED_LEGACY_STATUSES | _SKIPPABLE_STATUSES) + raise DatasetError( + f"invalid status/terminal_status combination: status={status!r}, " + f"terminal_status={terminal_status!r}; supported statuses are {supported}" + ) + + +def completed_label_status(label: dict[str, Any]) -> str | None: + """Return a validated completed target status, or None for an explicit skip.""" + classification, target_status = classify_label(label) + return target_status if classification == "completed" else None + + +def _validate_optional_v2_targets(label: dict[str, Any], identity: str) -> None: + if "affected_entrypoints" in label: + _validate_entrypoints(label["affected_entrypoints"], identity, require_evidence=False) + for field in _LIST_TARGET_FIELDS: + if field != "affected_entrypoints" and field in label: + _validate_string_list(label, field, identity) + + +def _validate_skippable_label(label: dict[str, Any]) -> None: + repository, pr = record_key(label) + identity = f"{repository}#{pr}" + status = cast("str", label["status"]) + if "terminal_status" in label: + _validate_optional_v2_targets(label, identity) + if label.get("affected_entrypoints") != []: + raise DatasetError(f"skippable v2 label {identity} requires no affected entrypoints") + elif status in {"unknown", "not_evaluable"} and "affected_entrypoints" in label: + _validate_optional_v2_targets(label, identity) + + +def validate_completed_label(label: dict[str, Any]) -> str: + """Validate one completed legacy review or real v2 compatibility projection.""" + repository, pr = record_key(label) + identity = f"{repository}#{pr}" + target_status = completed_label_status(label) + if target_status is None: + raise DatasetError(f"label {identity} is not a completed review") + + if target_status in {"positive", "negative_control"}: + _validate_optional_v2_targets(label, identity) + entrypoints = label.get("affected_entrypoints") + if not isinstance(entrypoints, list): + raise DatasetError( + f"completed v2 label {identity} requires affected_entrypoints as a list" + ) + if target_status == "positive" and not entrypoints: + raise DatasetError(f"completed v2 positive label {identity} requires entrypoints") + if target_status == "negative_control" and entrypoints: + raise DatasetError( + f"completed v2 negative-control label {identity} forbids entrypoints" + ) + return target_status + + _validate_reviewer(label, identity) + _validate_entrypoints(label.get("affected_entrypoints"), identity, require_evidence=True) + for field in _LEGACY_REQUIRED_LIST_FIELDS: + if field != "affected_entrypoints": + _validate_string_list(label, field, identity) + notes = label.get("notes") + if not isinstance(notes, str): + raise DatasetError(f"completed label {identity} requires notes as a string") + return target_status + + +def load_labels(path: Path) -> dict[tuple[str, int], dict[str, Any]]: + labels: dict[tuple[str, int], dict[str, Any]] = {} + seen: set[tuple[str, int]] = set() + for label in read_jsonl(path): + key = record_key(label) + if key in seen: + raise DatasetError(f"duplicate label: {key[0]}#{key[1]}") + seen.add(key) + classification, _target_status = classify_label(label) + if classification == "skippable": + # Only explicit pending/unknown/not-evaluable rows are absent from + # the join; malformed status shapes fail closed above. + _validate_skippable_label(label) + continue + validate_completed_label(label) + labels[key] = label + return labels + + +def _normalized_oid(value: object, location: str) -> str: + if ( + not isinstance(value, str) + or re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", value) is None + ): + raise DatasetError(f"{location} must be a full 40- or 64-character hexadecimal Git OID") + return value.lower() + + +def merge_sha(entry: dict[str, Any]) -> str | None: + """Return a normalized corpus merge OID, rejecting every malformed present value.""" + if "mergeCommit" not in entry or entry["mergeCommit"] is None: + return None + value = entry["mergeCommit"] + if isinstance(value, dict): + if set(value) != {"oid"}: + raise DatasetError("corpus mergeCommit must contain only an oid") + return _normalized_oid(value["oid"], "corpus mergeCommit.oid") + if isinstance(value, str): + return _normalized_oid(value, "corpus mergeCommit") + raise DatasetError("corpus mergeCommit must be an object with an oid, a Git OID, or null") + + +def diff_filename(repository: str, pr: int, merge_commit: str | None = None) -> str: + """Return one separator-free cache basename bound to the full target commit.""" + normalized_commit = ( + _normalized_oid(merge_commit, "cache merge commit") if merge_commit is not None else None + ) + repository_component = repository.replace("/", "--").replace("\\", "--") + suffix = f"--{normalized_commit}" if normalized_commit else "" + name = f"{repository_component}--{pr}{suffix}.diff" + if name in {".", ".."} or "/" in name or "\\" in name or Path(name).name != name: + raise DatasetError("diff cache name must be a single separator-free basename") + return name + + +def _diff_cache_path( + diff_dir: Path, + repository: str, + pr: int, + merge_commit: str | None, +) -> Path: + cache_root = diff_dir.expanduser().absolute() + candidate = (cache_root / diff_filename(repository, pr, merge_commit)).absolute() + if candidate.parent != cache_root: + raise DatasetError(f"diff cache path escapes --diff-dir: {candidate}") + return candidate + + +def _read_diff( + path: Path, + *, + input_paths: tuple[Path, ...] = (), +) -> str | None: + try: + raw = read_secure_regular_file( + path, + forbidden_files=_protected_files(input_paths), + ) + except SecurePathError as error: + raise DatasetError(str(error)) from error + if raw is None: + return None + if len(raw) > MAX_DIFF_BYTES: + raise DatasetError(f"diff is larger than {MAX_DIFF_BYTES} bytes: {path}") + try: + return raw.decode("utf-8") + except UnicodeDecodeError as error: + raise DatasetError(f"diff is not valid UTF-8: {path}") from error + + +def fetch_diff( + url: str, + destination: Path, + timeout: float = 30.0, + *, + input_paths: tuple[Path, ...] = (), +) -> str: + """Fetch one public diff and publish its cache entry without clobbering.""" + _validate_destination(destination, input_paths=input_paths) + request = Request(url, headers={"Accept": "text/plain", "User-Agent": "blast-radius-dataset"}) + try: + with urlopen(request, timeout=timeout) as response: + raw = cast("bytes", response.read(MAX_DIFF_BYTES + 1)) + except (OSError, URLError) as error: + raise DatasetError(f"cannot fetch diff {url}: {error}") from error + if len(raw) > MAX_DIFF_BYTES: + raise DatasetError(f"diff is larger than {MAX_DIFF_BYTES} bytes: {url}") + try: + decoded = raw.decode("utf-8") + except UnicodeDecodeError as error: + raise DatasetError(f"downloaded diff is not valid UTF-8: {url}") from error + _publish_exclusive_bytes(destination, raw, input_paths=input_paths) + return decoded + + +def _embedded_diff(entry: dict[str, Any]) -> str | None: + value = entry.get("diff", entry.get("patch")) + return value if isinstance(value, str) else None + + +def get_diff( + entry: dict[str, Any], + *, + diff_dir: Path | None, + fetch_missing: bool, + input_paths: tuple[Path, ...] = (), +) -> tuple[str | None, str]: + embedded = _embedded_diff(entry) + if embedded is not None: + return embedded, "corpus" + + repository, pr = record_key(entry) + target_commit = merge_sha(entry) + if diff_dir is not None: + cached = _diff_cache_path(diff_dir, repository, pr, target_commit) + cached_diff = _read_diff(cached, input_paths=input_paths) + if cached_diff is not None: + return cached_diff, "cache" + + if not fetch_missing: + return None, "missing" + + # Construct the URL from the validated repository identity instead of + # following an arbitrary URL embedded in a corpus record. + url = f"https://github.com/{repository}/pull/{pr}.diff" + if diff_dir is None: + raise DatasetError("--fetch-diffs requires --diff-dir") + return ( + fetch_diff( + url, + _diff_cache_path(diff_dir, repository, pr, target_commit), + input_paths=input_paths, + ), + # Serialized provenance describes stable data availability, not whether + # this invocation happened to populate the cache. + "cache", + ) + + +def target_from_label(label: dict[str, Any], scope: str) -> dict[str, Any]: + validate_completed_label(label) + filtered = filter_record(label, scope) + target: dict[str, Any] = {"status": label["status"]} + if "terminal_status" in label: + target["terminal_status"] = label["terminal_status"] + for field in _LIST_TARGET_FIELDS: + if field in filtered: + target[field] = filtered[field] + return target + + +def build_examples( + corpus: list[dict[str, Any]], + labels: dict[tuple[str, int], dict[str, Any]], + *, + scope: str = "fastapi", + diff_dir: Path | None = None, + fetch_missing: bool = False, + limit: int | None = None, + input_paths: tuple[Path, ...] = (), +) -> tuple[list[dict[str, Any]], int]: + """Join corpus and completed labels, returning examples and missing count.""" + if scope not in SCOPES: + raise DatasetError(f"unknown scope: {scope}") + # Callers may construct corpus records directly instead of using load_corpus; + # validate all merge identities before filtering, cache lookup, or fetching. + for entry in corpus: + merge_sha(entry) + examples: list[dict[str, Any]] = [] + missing_labels = 0 + for entry in corpus: + key = record_key(entry) + label = labels.get(key) + if label is None or completed_label_status(label) is None: + missing_labels += 1 + continue + validate_completed_label(label) + if limit is not None and len(examples) >= limit: + break + diff, diff_source = get_diff( + entry, + diff_dir=diff_dir, + fetch_missing=fetch_missing, + input_paths=input_paths, + ) + target = target_from_label(label, scope) + examples.append( + { + "schema_version": 1, + "id": f"{key[0]}#{key[1]}", + "input": { + "repository": key[0], + "pr": key[1], + "title": entry.get("title", ""), + "body": entry.get("body", ""), + "changed_files": entry.get("files", []), + "diff": diff, + }, + "target": target, + "metadata": { + "label_source_status": label["status"], + "label_terminal_status": label.get("terminal_status"), + "diff_source": diff_source, + "diff_sha256": hashlib.sha256(diff.encode("utf-8")).hexdigest() + if diff is not None + else None, + "merge_commit": entry.get("mergeCommit"), + "base_ref": entry.get("baseRefName"), + "head_ref": entry.get("headRefName"), + }, + } + ) + return examples, missing_labels + + +def _write_jsonl( + path: Path, + records: list[dict[str, Any]], + *, + input_paths: tuple[Path, ...] = (), +) -> None: + _validate_destination(path, input_paths=input_paths) + try: + content = "".join( + json.dumps(record, sort_keys=True, allow_nan=False) + "\n" for record in records + ).encode("utf-8") + except (TypeError, ValueError) as error: + raise DatasetError(f"dataset output is not strict JSON: {error}") from error + _publish_exclusive_bytes(path, content, input_paths=input_paths) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Build exploratory blast-radius fine-tuning JSONL without production gates." + ) + parser.add_argument( + "--corpus", + type=Path, + required=True, + help="Exploratory corpus metadata JSON (frozen destinations remain protected).", + ) + parser.add_argument( + "--labels", + type=Path, + required=True, + help="Completed review JSONL or a v2 compatibility projection.", + ) + parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) + parser.add_argument("--diff-dir", type=Path) + parser.add_argument( + "--fetch-diffs", + action="store_true", + help="Fetch missing public PR diffs into --diff-dir (otherwise diff is null).", + ) + parser.add_argument("--scope", choices=SCOPES, default="fastapi") + parser.add_argument("--limit", type=int) + args = parser.parse_args(argv) + + if args.limit is not None and args.limit < 1: + parser.error("--limit must be positive") + if args.fetch_diffs and args.diff_dir is None: + parser.error("--fetch-diffs requires --diff-dir") + + input_paths = (args.corpus, args.labels) + cache_protected_paths = (*input_paths, args.output) + try: + _validate_destination(args.output, input_paths=input_paths) + corpus = load_corpus(args.corpus) + labels = load_labels(args.labels) + examples, missing_labels = build_examples( + corpus, + labels, + scope=args.scope, + diff_dir=args.diff_dir, + fetch_missing=args.fetch_diffs, + limit=args.limit, + input_paths=cache_protected_paths, + ) + _write_jsonl(args.output, examples, input_paths=input_paths) + except DatasetError as error: + parser.error(str(error)) + + missing_diffs = sum(example["metadata"]["diff_source"] == "missing" for example in examples) + print( + json.dumps( + { + "examples": len(examples), + "missing_labels": missing_labels, + "missing_diffs": missing_diffs, + "scope": args.scope, + "output": str(args.output), + }, + sort_keys=True, + allow_nan=False, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/real_world/collect_prs.py b/benchmarks/real_world/collect_prs.py index bfb78d0..43c0d0c 100644 --- a/benchmarks/real_world/collect_prs.py +++ b/benchmarks/real_world/collect_prs.py @@ -1,32 +1,317 @@ #!/usr/bin/env python3 -"""Collect a reproducible corpus of recent merged PRs from FastAPI projects. +"""Collect a cheap exploratory corpus of recent merged PRs. -Requires an authenticated GitHub CLI (`gh`). The collector stores immutable PR -and commit identifiers plus changed-file metadata, but not third-party source. -Diffs can be fetched on demand during evaluation from each PR's diff URL. +Requires an authenticated GitHub CLI (`gh`). The collector stores PR and commit +identifiers plus changed-file metadata, but not third-party source. The latest-N +selection is intentionally a convenient sampling strategy, not a publication +lock; diffs can be fetched later from each PR's diff URL. """ from __future__ import annotations import argparse import json +import re import subprocess -from datetime import UTC, datetime +import sys +from datetime import datetime, timezone from pathlib import Path from typing import Any +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from benchmarks.real_world._secure_publish import ( + SecurePathError, + ensure_publishable, + publish_exclusive_bytes, +) + HERE = Path(__file__).resolve().parent +DEFAULT_OUTPUT = HERE / "candidate-corpus.json" + +_PROTECTED_ARTIFACTS = ( + HERE / "corpus.json", + HERE / "review-a.jsonl", + HERE / "review-b.jsonl", + HERE / "adjudicated.jsonl", + HERE / "adjudication-amendments.jsonl", + HERE / "reachability-supplements.jsonl", + HERE / "review-queue.json", +) +_PROTECTED_ROOTS = tuple( + HERE / name + for name in ( + "expansion", + "ground_truth_v2", + "pilot_v2", + "pilot_v3", + "production_v1", + "scopes", + "verification_sets", + ) +) + + +class CollectorError(ValueError): + """Raised when exploratory collection input or publication is unsafe.""" + + +def _reject_duplicate_members(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for name, value in pairs: + if name in result: + raise CollectorError(f"duplicate JSON key: {name}") + result[name] = value + return result + + +def _reject_non_finite(token: str) -> None: + raise CollectorError(f"non-finite JSON number: {token}") + + +def _strict_json_loads(content: str, source: str) -> Any: + try: + return json.loads( + content, + object_pairs_hook=_reject_duplicate_members, + parse_constant=_reject_non_finite, + ) + except (json.JSONDecodeError, CollectorError) as error: + raise CollectorError(f"invalid JSON in {source}: {error}") from error + + +def _protected_files(input_paths: tuple[Path, ...]) -> tuple[Path, ...]: + return (*_PROTECTED_ARTIFACTS, *input_paths) + + +def _validate_destination(path: Path, *, input_paths: tuple[Path, ...] = ()) -> None: + absolute = Path(path).expanduser().absolute() + if absolute in {item.expanduser().absolute() for item in input_paths}: + raise CollectorError(f"refusing to overwrite input file: {path}") + if absolute in {item.expanduser().absolute() for item in _PROTECTED_ARTIFACTS} or any( + absolute.is_relative_to(root.expanduser().absolute()) for root in _PROTECTED_ROOTS + ): + raise CollectorError(f"refusing to target frozen benchmark artifact: {path}") + try: + ensure_publishable( + path, + forbidden_files=_protected_files(input_paths), + forbidden_roots=_PROTECTED_ROOTS, + ) + except SecurePathError as error: + raise CollectorError(str(error)) from error + + +def _publish_exclusive_bytes( + output: Path, + content: bytes, + *, + input_paths: tuple[Path, ...] = (), +) -> None: + """Durably publish exact bytes through a stable destination-directory FD.""" + _validate_destination(output, input_paths=input_paths) + try: + publish_exclusive_bytes( + output, + content, + forbidden_files=_protected_files(input_paths), + forbidden_roots=_PROTECTED_ROOTS, + ) + except SecurePathError as error: + raise CollectorError(str(error)) from error + + +def _publish_output( + output: Path, + payload: dict[str, Any], + *, + input_paths: tuple[Path, ...] = (), +) -> None: + try: + content = (json.dumps(payload, indent=2, sort_keys=True, allow_nan=False) + "\n").encode( + "utf-8" + ) + except (TypeError, ValueError) as error: + raise CollectorError(f"collector output is not strict JSON: {error}") from error + _publish_exclusive_bytes(output, content, input_paths=input_paths) def gh_json(*args: str) -> Any: - """Run gh and decode its JSON response.""" - result = subprocess.run( - ["gh", *args], - check=True, - capture_output=True, - text=True, - ) - return json.loads(result.stdout) + """Run gh and strictly decode its JSON response as UTF-8.""" + try: + result = subprocess.run( + ["gh", *args], + check=True, + capture_output=True, + text=True, + encoding="utf-8", + errors="strict", + ) + except UnicodeError as error: + raise CollectorError("gh response is not valid UTF-8") from error + return _strict_json_loads(result.stdout, "gh response") + + +def _positive_integer(value: object, location: str) -> int: + if type(value) is not int or value < 1: + raise CollectorError(f"{location} must be a positive integer") + return value + + +def _nonnegative_integer(value: object, location: str) -> int: + if type(value) is not int or value < 0: + raise CollectorError(f"{location} must be a nonnegative integer") + return value + + +def _required_string(record: dict[str, Any], field: str, location: str) -> str: + value = record.get(field) + if not isinstance(value, str) or not value.strip(): + raise CollectorError(f"{location}.{field} must be a non-empty string") + return value + + +def _oid(value: object, location: str) -> str: + if ( + not isinstance(value, str) + or re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", value) is None + ): + raise CollectorError(f"{location} must be a Git object ID") + return value + + +def _validate_author(value: object, location: str) -> None: + if value is None: + return + if not isinstance(value, dict): + raise CollectorError(f"{location} must be an object or null") + _required_string(value, "id", location) + _required_string(value, "login", location) + if "name" not in value: + raise CollectorError(f"{location}.name is required") + name = value.get("name") + if name is not None and not isinstance(name, str): + raise CollectorError(f"{location}.name must be a string or null") + if not isinstance(value.get("is_bot"), bool): + raise CollectorError(f"{location}.is_bot must be a boolean") + + +def _normalized_author(value: object) -> dict[str, object] | None: + if value is None: + return None + assert isinstance(value, dict) + return {field: value[field] for field in ("id", "is_bot", "login", "name")} + + +def _validate_pr_summary(pr: dict[str, Any], repository: str, index: int) -> int: + location = f"gh PR list {repository}[{index}]" + number = _positive_integer(pr.get("number"), f"{location}.number") + for field in ("title", "url", "mergedAt", "baseRefName", "headRefName"): + _required_string(pr, field, location) + merge_commit = pr.get("mergeCommit") + if not isinstance(merge_commit, dict): + raise CollectorError(f"{location}.mergeCommit must be an object") + _oid(merge_commit.get("oid"), f"{location}.mergeCommit.oid") + if "author" not in pr: + raise CollectorError(f"{location}.author is required") + _validate_author(pr["author"], f"{location}.author") + return number + + +def _validate_detail(detail: dict[str, Any], identity: str) -> None: + for field in ("additions", "deletions", "changedFiles"): + _nonnegative_integer(detail.get(field), f"gh PR detail {identity}.{field}") + body = detail.get("body") + if not isinstance(body, str): + raise CollectorError(f"gh PR detail {identity}.body must be a string") + files = detail.get("files") + if not isinstance(files, list): + raise CollectorError(f"gh PR detail {identity}.files must be a list") + seen_paths: set[str] = set() + for index, item in enumerate(files): + location = f"gh PR detail {identity}.files[{index}]" + if not isinstance(item, dict): + raise CollectorError(f"{location} must be an object") + path = _required_string(item, "path", location) + if path in seen_paths: + raise CollectorError(f"gh PR detail {identity} has duplicate file path: {path}") + seen_paths.add(path) + for field in ("additions", "deletions"): + _nonnegative_integer(item.get(field), f"{location}.{field}") + _required_string(item, "changeType", location) + commits = detail.get("commits") + if not isinstance(commits, list) or not commits: + raise CollectorError(f"gh PR detail {identity}.commits must be a non-empty list") + seen_oids: set[str] = set() + for index, commit in enumerate(commits): + location = f"gh PR detail {identity}.commits[{index}]" + if not isinstance(commit, dict): + raise CollectorError(f"{location} must be an object") + oid = _oid(commit.get("oid"), f"{location}.oid") + if oid in seen_oids: + raise CollectorError(f"gh PR detail {identity} has duplicate commit OID") + seen_oids.add(oid) + + +def _validate_collected_entry(entry: dict[str, Any], index: int) -> tuple[str, int]: + location = f"collector entries[{index}]" + repository = _required_string(entry, "repository", location) + number = _validate_pr_summary(entry, repository, index) + for field in ("additions", "deletions", "changedFiles"): + _nonnegative_integer(entry.get(field), f"{location}.{field}") + if not isinstance(entry.get("body"), str): + raise CollectorError(f"{location}.body must be a string") + files = entry.get("files") + if not isinstance(files, list): + raise CollectorError(f"{location}.files must be a list") + seen_paths: set[str] = set() + for file_index, item in enumerate(files): + file_location = f"{location}.files[{file_index}]" + if not isinstance(item, dict): + raise CollectorError(f"{file_location} must be an object") + path = _required_string(item, "path", file_location) + if path in seen_paths: + raise CollectorError(f"{location} has duplicate file path: {path}") + seen_paths.add(path) + for field in ("additions", "deletions"): + _nonnegative_integer(item.get(field), f"{file_location}.{field}") + _required_string(item, "changeType", file_location) + commits = entry.get("commits") + if not isinstance(commits, list) or not commits: + raise CollectorError(f"{location}.commits must be a non-empty list") + seen_oids: set[str] = set() + for commit_index, commit in enumerate(commits): + oid = _oid(commit, f"{location}.commits[{commit_index}]") + if oid in seen_oids: + raise CollectorError(f"{location} has duplicate commit OID") + seen_oids.add(oid) + expected_url = f"https://github.com/{repository}/pull/{number}.diff" + if entry.get("diff_url") != expected_url: + raise CollectorError(f"{location}.diff_url does not match its identity") + ground_truth = entry.get("ground_truth") + if ground_truth != { + "status": "pending_double_review", + "review_a": None, + "review_b": None, + "adjudicated": None, + }: + raise CollectorError(f"{location}.ground_truth has an invalid pending shape") + return repository, number + + +def _validate_output_payload(payload: dict[str, Any]) -> None: + entries = payload.get("entries") + if not isinstance(entries, list): + raise CollectorError("collector output entries must be a list") + seen: set[tuple[str, int]] = set() + for index, entry in enumerate(entries): + if not isinstance(entry, dict): + raise CollectorError(f"collector entries[{index}] must be an object") + identity = _validate_collected_entry(entry, index) + if identity in seen: + raise CollectorError(f"collector output has duplicate PR: {identity[0]}#{identity[1]}") + seen.add(identity) def collect_repository(repository: str, limit: int) -> list[dict[str, Any]]: @@ -43,9 +328,20 @@ def collect_repository(repository: str, limit: int) -> list[dict[str, Any]]: "--json", "number,title,url,mergedAt,mergeCommit,baseRefName,headRefName,author", ) + if not isinstance(prs, list) or any(not isinstance(pr, dict) for pr in prs): + raise CollectorError(f"gh returned an invalid PR list for {repository}") corpus: list[dict[str, Any]] = [] - for pr in prs: + seen_numbers: set[int] = set() + validated: list[tuple[dict[str, Any], int]] = [] + for index, pr in enumerate(prs): + number = _validate_pr_summary(pr, repository, index) + if number in seen_numbers: + raise CollectorError(f"gh returned duplicate PR number for {repository}: {number}") + seen_numbers.add(number) + validated.append((pr, number)) + + for pr, number in validated: detail = gh_json( "pr", "view", @@ -55,14 +351,32 @@ def collect_repository(repository: str, limit: int) -> list[dict[str, Any]]: "--json", "additions,deletions,changedFiles,files,body,commits", ) + if not isinstance(detail, dict): + raise CollectorError(f"gh returned invalid PR detail for {repository}#{number}") + _validate_detail(detail, f"{repository}#{number}") corpus.append( { "repository": repository, - **pr, + "number": number, + "title": pr["title"], + "url": pr["url"], + "mergedAt": pr["mergedAt"], + "mergeCommit": {"oid": pr["mergeCommit"]["oid"]}, + "baseRefName": pr["baseRefName"], + "headRefName": pr["headRefName"], + "author": _normalized_author(pr["author"]), "additions": detail["additions"], "deletions": detail["deletions"], "changedFiles": detail["changedFiles"], - "files": detail["files"], + "files": [ + { + "path": item["path"], + "additions": item["additions"], + "deletions": item["deletions"], + "changeType": item["changeType"], + } + for item in detail["files"] + ], "body": detail["body"], "commits": [commit["oid"] for commit in detail["commits"]], "diff_url": f"https://github.com/{repository}/pull/{pr['number']}.diff", @@ -77,28 +391,91 @@ def collect_repository(repository: str, limit: int) -> list[dict[str, Any]]: return corpus -def main() -> None: - parser = argparse.ArgumentParser() +def _validate_config(config: dict[str, Any]) -> dict[str, Any]: + limit = config.get("prs_per_repository") + if type(limit) is not int or limit < 1: + raise CollectorError("prs_per_repository must be a positive integer") + repositories = config.get("repositories") + if not isinstance(repositories, list) or not repositories: + raise CollectorError("repositories must be a non-empty list of objects") + for index, repository in enumerate(repositories): + if not isinstance(repository, dict): + raise CollectorError(f"repositories[{index}] must be an object") + name = repository.get("name") + if not isinstance(name, str) or not name.strip(): + raise CollectorError(f"repositories[{index}].name must be a non-empty string") + return config + + +def _resolve_config( + path: Path, + repositories: list[str] | None, + limit: int | None, +) -> dict[str, Any]: + try: + raw_config = _strict_json_loads(path.read_text(encoding="utf-8"), str(path)) + except (OSError, UnicodeError) as error: + raise CollectorError(f"cannot read config {path}: {error}") from error + if not isinstance(raw_config, dict): + raise CollectorError("config must be a JSON object") + config: dict[str, Any] = raw_config + if repositories: + config = { + **config, + "repositories": [{"name": repository} for repository in repositories], + } + if limit is not None: + config = {**config, "prs_per_repository": limit} + return _validate_config(config) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description=( + "Collect a cheap, exploratory PR metadata corpus without changing frozen artifacts." + ) + ) parser.add_argument("--config", type=Path, default=HERE / "repos.json") - parser.add_argument("--output", type=Path, default=HERE / "corpus.json") - args = parser.parse_args() - - config = json.loads(args.config.read_text(encoding="utf-8")) - limit = config["prs_per_repository"] - entries: list[dict[str, Any]] = [] - for repository in config["repositories"]: - entries.extend(collect_repository(repository["name"], limit)) - - output = { - "schema_version": 1, - "selection": "Latest N merged PRs returned by GitHub at collection time; no content filtering.", - "collected_at": datetime.now(UTC).isoformat(), - "config": config, - "entries": entries, - } - args.output.write_text(json.dumps(output, indent=2) + "\n", encoding="utf-8") + parser.add_argument( + "--repository", + action="append", + dest="repositories", + help="Repository to collect (repeatable); overrides repositories in --config.", + ) + parser.add_argument( + "--limit", + type=int, + help="Merged PRs per repository; overrides prs_per_repository in --config.", + ) + parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) + args = parser.parse_args(argv) + + try: + _validate_destination(args.output, input_paths=(args.config,)) + config = _resolve_config(args.config, args.repositories, args.limit) + limit = config["prs_per_repository"] + entries: list[dict[str, Any]] = [] + for repository in config["repositories"]: + entries.extend(collect_repository(repository["name"], limit)) + + output = { + "schema_version": 1, + "dataset_kind": "exploratory_pr_metadata", + "selection": ( + "Latest N merged PRs returned by GitHub at collection time; no content filtering." + ), + "collected_at": datetime.now(timezone.utc).isoformat(), + "config": config, + "entries": entries, + } + _validate_output_payload(output) + _publish_output(args.output, output, input_paths=(args.config,)) + except (CollectorError, OSError, subprocess.SubprocessError, KeyError, TypeError) as error: + parser.error(str(error)) + print(f"Wrote {len(entries)} PRs to {args.output}") + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/tests/benchmarks/test_build_training_dataset.py b/tests/benchmarks/test_build_training_dataset.py new file mode 100644 index 0000000..183d2ea --- /dev/null +++ b/tests/benchmarks/test_build_training_dataset.py @@ -0,0 +1,589 @@ +"""Tests for the cheap exploratory training-dataset builder.""" + +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any +from unittest.mock import patch + +from benchmarks.real_world import build_training_dataset as builder + + +def completed_label( + pr: int, + *, + entrypoints: list[dict[str, Any]] | None = None, + status: str = "reviewed", +) -> dict[str, Any]: + return { + "repository": "owner/repo", + "pr": pr, + "status": status, + "reviewer": {"kind": "agent", "name": "reviewer", "version": "v1"}, + "changed_symbols": ["app.handler"], + "affected_entrypoints": entrypoints or [], + "affected_tests": [], + "contract_changes": [], + "cross_repository_consumers": [], + "unknowns": [], + "orphans": [], + "notes": "completed review", + } + + +def http_entrypoint(path: str = "/items") -> dict[str, Any]: + return { + "id": f"HTTP GET {path}", + "kind": "http", + "confidence": "confirmed", + "evidence": ["app.py:10 handler"], + } + + +class _FakeResponse: + def __init__(self, content: bytes, barrier: threading.Barrier | None = None) -> None: + self.content = content + self.barrier = barrier + + def __enter__(self) -> _FakeResponse: + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self, _limit: int) -> bytes: + if self.barrier is not None: + self.barrier.wait() + return self.content + + +class TrainingDatasetTests(unittest.TestCase): + def test_build_examples_filters_scope_and_never_invents_negatives(self) -> None: + corpus = [ + { + "repository": "owner/repo", + "number": 1, + "title": "change handler", + "body": "body", + "files": [{"path": "app.py"}], + }, + {"repository": "owner/repo", "number": 2, "files": []}, + {"repository": "owner/repo", "number": 3, "files": []}, + ] + label = completed_label( + 1, + entrypoints=[ + http_entrypoint(), + { + "id": "Web UI /", + "kind": "other", + "confidence": "confirmed", + "evidence": ["ui.py:1 page"], + }, + ], + ) + label["unknowns"] = ["dynamic registration"] + labels = { + ("owner/repo", 1): label, + ("owner/repo", 3): { + "repository": "owner/repo", + "pr": 3, + "status": "not_evaluable", + "affected_entrypoints": [], + }, + } + + examples, missing = builder.build_examples(corpus, labels) + + self.assertEqual(missing, 2) + self.assertEqual(len(examples), 1) + self.assertEqual(examples[0]["id"], "owner/repo#1") + self.assertEqual(examples[0]["target"]["affected_entrypoints"], [http_entrypoint()]) + self.assertIsNone(examples[0]["input"]["diff"]) + self.assertEqual(examples[0]["metadata"]["diff_source"], "missing") + + def test_local_diff_is_used_and_hashed(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + diff_dir = Path(temporary_name) + diff = "diff --git a/app.py b/app.py\n+return 1\n" + merge_commit = "A" * 40 + cache_name = builder.diff_filename("owner/repo", 7, merge_commit) + self.assertEqual(cache_name, f"owner--repo--7--{'a' * 40}.diff") + (diff_dir / cache_name).write_text(diff, encoding="utf-8") + corpus = [ + { + "repository": "owner/repo", + "number": 7, + "mergeCommit": {"oid": merge_commit}, + "files": [], + } + ] + label = completed_label(7, status="adjudicated") + labels = {("owner/repo", 7): label} + + examples, missing = builder.build_examples(corpus, labels, diff_dir=diff_dir) + + self.assertEqual(missing, 0) + self.assertEqual(examples[0]["input"]["diff"], diff) + self.assertEqual(examples[0]["metadata"]["diff_source"], "cache") + self.assertEqual( + examples[0]["metadata"]["diff_sha256"], + hashlib.sha256(diff.encode()).hexdigest(), + ) + + def test_cache_path_accepts_supported_oid_lengths_and_stays_confined(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + diff_dir = Path(temporary_name) / "diffs" + for length in (40, 64): + with self.subTest(length=length): + oid = "A" * length + path = builder._diff_cache_path(diff_dir, "owner/repo", 7, oid) + self.assertEqual(path.parent, diff_dir.absolute()) + self.assertEqual(path.name, f"owner--repo--7--{'a' * length}.diff") + self.assertNotIn("/", path.name) + self.assertNotIn("\\", path.name) + + def test_v2_dual_status_preserves_terminal_and_filters_scope(self) -> None: + corpus = [{"repository": "owner/repo", "number": 4, "files": []}] + labels = { + ("owner/repo", 4): { + "repository": "owner/repo", + "pr": 4, + "status": "adjudicated", + "terminal_status": "positive", + "affected_entrypoints": [ + { + "id": "HTTP GET /health", + "kind": "http", + "confidence": "confirmed", + }, + { + "id": "CLI inspect", + "kind": "cli", + "confidence": "probable", + }, + ], + } + } + + examples, missing = builder.build_examples(corpus, labels, scope="fastapi") + + self.assertEqual(missing, 0) + self.assertEqual(examples[0]["target"]["status"], "adjudicated") + self.assertEqual(examples[0]["target"]["terminal_status"], "positive") + self.assertEqual( + examples[0]["target"]["affected_entrypoints"], + [{"id": "HTTP GET /health", "kind": "http", "confidence": "confirmed"}], + ) + self.assertNotIn("changed_symbols", examples[0]["target"]) + self.assertEqual(examples[0]["metadata"]["label_source_status"], "adjudicated") + self.assertEqual(examples[0]["metadata"]["label_terminal_status"], "positive") + + def test_status_only_completed_label_is_rejected(self) -> None: + corpus = [{"repository": "owner/repo", "number": 5, "files": []}] + labels = { + ("owner/repo", 5): { + "repository": "owner/repo", + "pr": 5, + "status": "reviewed", + } + } + + with self.assertRaisesRegex(builder.DatasetError, "requires reviewer provenance"): + builder.build_examples(corpus, labels) + + def test_endpoint_evidence_is_required_for_legacy_review(self) -> None: + label = completed_label( + 6, + entrypoints=[ + { + "id": "HTTP GET /items", + "kind": "http", + "confidence": "confirmed", + } + ], + ) + + with self.assertRaisesRegex(builder.DatasetError, "evidence"): + builder.validate_completed_label(label) + + def test_duplicate_labels_are_rejected_before_status_filtering(self) -> None: + pending = {"repository": "owner/repo", "pr": 1, "status": "pending_double_review"} + complete = completed_label(1) + with tempfile.TemporaryDirectory() as temporary_name: + path = Path(temporary_name) / "labels.jsonl" + for records in ((pending, complete), (complete, pending), (pending, pending)): + with self.subTest(statuses=[record["status"] for record in records]): + path.write_text( + "".join(json.dumps(record) + "\n" for record in records), + encoding="utf-8", + ) + with self.assertRaisesRegex(builder.DatasetError, "duplicate label"): + builder.load_labels(path) + + def test_label_statuses_are_explicit_and_contradictions_fail_closed(self) -> None: + invalid = ( + {"repository": "owner/repo", "pr": 1}, + {"repository": "owner/repo", "pr": 1, "status": 1}, + {"repository": "owner/repo", "pr": 1, "status": "reviewd"}, + { + "repository": "owner/repo", + "pr": 1, + "status": "adjudicated", + "terminal_status": "postive", + }, + { + "repository": "owner/repo", + "pr": 1, + "status": "reviewed", + "terminal_status": "positive", + }, + ) + with tempfile.TemporaryDirectory() as temporary_name: + path = Path(temporary_name) / "labels.jsonl" + for label in invalid: + with self.subTest(label=label): + path.write_text(json.dumps(label) + "\n", encoding="utf-8") + with self.assertRaises(builder.DatasetError): + builder.load_labels(path) + + def test_only_enumerated_skippable_shapes_are_accepted(self) -> None: + rows = [ + {"repository": "owner/repo", "pr": 1, "status": "pending_double_review"}, + {"repository": "owner/repo", "pr": 2, "status": "unknown"}, + { + "repository": "owner/repo", + "pr": 3, + "status": "not_evaluable", + "terminal_status": "not_evaluable", + "affected_entrypoints": [], + }, + ] + with tempfile.TemporaryDirectory() as temporary_name: + path = Path(temporary_name) / "labels.jsonl" + path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + self.assertEqual(builder.load_labels(path), {}) + + def test_record_key_rejects_conflicting_aliases(self) -> None: + with self.assertRaisesRegex(builder.DatasetError, "conflicting"): + builder.record_key({"repository": "owner/repo", "pr": 1, "number": 999}) + + def test_v2_optional_target_fields_are_validated(self) -> None: + base: dict[str, Any] = { + "repository": "owner/repo", + "pr": 4, + "status": "adjudicated", + "terminal_status": "positive", + "affected_entrypoints": [ + {"id": "HTTP GET /", "kind": "http", "confidence": "confirmed"} + ], + } + for field in ( + "changed_symbols", + "affected_tests", + "contract_changes", + "unknowns", + "orphans", + ): + with self.subTest(field=field): + label = {**base, field: "not-a-list"} + with self.assertRaisesRegex(builder.DatasetError, field): + builder.validate_completed_label(label) + + def test_legacy_completed_contract_requires_cross_consumers_and_notes(self) -> None: + label = completed_label(8) + del label["cross_repository_consumers"] + with self.assertRaisesRegex(builder.DatasetError, "cross_repository_consumers"): + builder.validate_completed_label(label) + label = completed_label(8) + del label["notes"] + with self.assertRaisesRegex(builder.DatasetError, "notes"): + builder.validate_completed_label(label) + + def test_strict_json_rejects_duplicate_keys_and_non_finite_values(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + duplicate = root / "duplicate.jsonl" + duplicate.write_text( + '{"repository":"owner/repo","pr":1,"pr":2,"status":"reviewed"}\n', + encoding="utf-8", + ) + non_finite = root / "corpus.json" + non_finite.write_text('{"entries":[],"value":NaN}\n', encoding="utf-8") + + with self.assertRaisesRegex(builder.DatasetError, "duplicate JSON member"): + builder.read_jsonl(duplicate) + with self.assertRaisesRegex(builder.DatasetError, "non-finite JSON number"): + builder.load_corpus(non_finite) + + def test_corpus_merge_commit_rejects_malformed_present_values(self) -> None: + malformed: tuple[object, ...] = ( + {"oid": "/../../secret"}, + {"oid": "..\\..\\secret"}, + {"oid": "." * 40}, + {"oid": "a" * 39}, + {"oid": "g" * 40}, + {"sha": "a" * 40}, + "/../../secret", + "", + [], + 1, + ) + with tempfile.TemporaryDirectory() as temporary_name: + path = Path(temporary_name) / "corpus.json" + for merge_commit in malformed: + with self.subTest(merge_commit=merge_commit): + path.write_text( + json.dumps( + { + "entries": [ + { + "repository": "owner/repo", + "number": 7, + "mergeCommit": merge_commit, + } + ] + } + ), + encoding="utf-8", + ) + with self.assertRaisesRegex(builder.DatasetError, "mergeCommit"): + builder.load_corpus(path) + + def test_merge_commit_traversal_cannot_read_or_write_outside_diff_dir(self) -> None: + labels = {("owner/repo", 7): completed_label(7)} + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + diff_dir = root / "diffs" + diff_dir.mkdir() + # This component made the old `...--/../../...` cache path traversable. + (diff_dir / "owner--repo--7--").mkdir() + outside_secret = root / "secre.diff" + outside_secret.write_text("SECRET OUTSIDE CACHE\n", encoding="utf-8") + outside_write = root / "writt.diff" + + for merge_commit in ( + {"oid": "/../../secret"}, + {"oid": "/../../written"}, + {"oid": "..\\..\\secret"}, + {"oid": "." * 40}, + ): + corpus = [ + { + "repository": "owner/repo", + "number": 7, + "mergeCommit": merge_commit, + "files": [], + } + ] + with ( + self.subTest(merge_commit=merge_commit), + patch.object(builder, "_read_diff") as read_diff, + patch.object(builder, "urlopen") as network, + self.assertRaisesRegex(builder.DatasetError, "mergeCommit"), + ): + builder.build_examples( + corpus, + labels, + diff_dir=diff_dir, + fetch_missing=True, + ) + read_diff.assert_not_called() + network.assert_not_called() + self.assertEqual( + outside_secret.read_text(encoding="utf-8"), "SECRET OUTSIDE CACHE\n" + ) + self.assertFalse(outside_write.exists()) + + def test_jsonl_serialization_rejects_non_finite_values(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + output = Path(temporary_name) / "dataset.jsonl" + + with self.assertRaisesRegex(builder.DatasetError, "strict JSON"): + builder._write_jsonl(output, [{"value": float("nan")}]) + + self.assertFalse(output.exists()) + + def test_output_rejects_existing_protected_and_input_destinations(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + existing = root / "dataset.jsonl" + existing.write_text("old\n", encoding="utf-8") + source = root / "labels.jsonl" + source.write_text("input\n", encoding="utf-8") + + with self.assertRaisesRegex(builder.DatasetError, "already exists"): + builder._write_jsonl(existing, []) + with self.assertRaisesRegex(builder.DatasetError, "overwrite input"): + builder._write_jsonl(source, [], input_paths=(source,)) + with self.assertRaisesRegex(builder.DatasetError, "frozen benchmark"): + builder._write_jsonl(builder.HERE / "corpus.json", []) + + self.assertEqual(existing.read_text(encoding="utf-8"), "old\n") + self.assertEqual(source.read_text(encoding="utf-8"), "input\n") + + def test_jsonl_publication_race_has_exactly_one_winner(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + output = Path(temporary_name) / "dataset.jsonl" + barrier = threading.Barrier(2) + + def publish(value: int) -> str: + barrier.wait() + try: + builder._write_jsonl(output, [{"value": value}]) + except builder.DatasetError: + return "lost" + return "won" + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(publish, (1, 2))) + + self.assertEqual(sorted(results), ["lost", "won"]) + self.assertIn(output.read_text(encoding="utf-8"), {'{"value": 1}\n', '{"value": 2}\n'}) + + def test_download_cache_is_exclusive_and_race_safe_without_network(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + destination = Path(temporary_name) / "cache.diff" + barrier = threading.Barrier(2) + + def fake_urlopen(_request: object, *, timeout: float) -> _FakeResponse: + self.assertEqual(timeout, 30.0) + return _FakeResponse(b"complete diff\n", barrier) + + def fetch() -> str: + try: + builder.fetch_diff("https://github.com/owner/repo/pull/1.diff", destination) + except builder.DatasetError: + return "lost" + return "won" + + with ( + patch.object(builder, "urlopen", side_effect=fake_urlopen), + ThreadPoolExecutor(max_workers=2) as executor, + ): + results = list(executor.map(lambda _index: fetch(), range(2))) + + self.assertEqual(sorted(results), ["lost", "won"]) + self.assertEqual(destination.read_bytes(), b"complete diff\n") + + def test_publication_parent_swap_cannot_redirect_dataset(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + parent = root / "parent" + held = root / "held" + protected = root / "protected" + parent.mkdir() + protected.mkdir() + output = parent / "dataset.jsonl" + real_link = os.link + + def swap_then_link(*args: Any, **kwargs: Any) -> None: + parent.rename(held) + parent.symlink_to(protected, target_is_directory=True) + real_link(*args, **kwargs) + + with patch("benchmarks.real_world._secure_publish.os.link", side_effect=swap_then_link): + builder._write_jsonl(output, [{"stable": True}]) + + self.assertTrue((held / "dataset.jsonl").is_file()) + self.assertFalse((protected / "dataset.jsonl").exists()) + + def test_download_parent_swap_cannot_redirect_cache(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + parent = root / "parent" + held = root / "held" + protected = root / "protected" + parent.mkdir() + protected.mkdir() + destination = parent / "cache.diff" + real_link = os.link + + def swap_then_link(*args: Any, **kwargs: Any) -> None: + parent.rename(held) + parent.symlink_to(protected, target_is_directory=True) + real_link(*args, **kwargs) + + with ( + patch.object(builder, "urlopen", return_value=_FakeResponse(b"diff\n")), + patch( + "benchmarks.real_world._secure_publish.os.link", + side_effect=swap_then_link, + ), + ): + builder.fetch_diff("https://github.com/owner/repo/pull/1.diff", destination) + + self.assertEqual((held / "cache.diff").read_bytes(), b"diff\n") + self.assertFalse((protected / "cache.diff").exists()) + + def test_cache_rejects_symlinks_hardlinks_aliases_and_invalid_utf8(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + source = root / "source.diff" + source.write_bytes(b"diff\n") + symlink = root / "symlink.diff" + symlink.symlink_to(source) + with self.assertRaises(builder.DatasetError): + builder._read_diff(symlink) + + hardlink = root / "hardlink.diff" + os.link(source, hardlink) + with self.assertRaisesRegex(builder.DatasetError, "exactly one hard link"): + builder._read_diff(hardlink, input_paths=(source,)) + + invalid = root / "invalid.diff" + invalid.write_bytes(b"abc\xffdef") + with self.assertRaisesRegex(builder.DatasetError, "valid UTF-8"): + builder._read_diff(invalid) + + def test_invalid_download_utf8_does_not_consume_cache_path(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + destination = Path(temporary_name) / "cache.diff" + with ( + patch.object(builder, "urlopen", return_value=_FakeResponse(b"abc\xffdef")), + self.assertRaisesRegex(builder.DatasetError, "valid UTF-8"), + ): + builder.fetch_diff("https://github.com/owner/repo/pull/1.diff", destination) + self.assertFalse(destination.exists()) + + def test_first_fetch_and_later_cache_examples_are_byte_identical(self) -> None: + corpus = [{"repository": "owner/repo", "number": 9, "files": []}] + labels = {("owner/repo", 9): completed_label(9)} + with tempfile.TemporaryDirectory() as temporary_name: + diff_dir = Path(temporary_name) / "diffs" + with patch.object(builder, "urlopen", return_value=_FakeResponse(b"same diff\n")): + first, _missing = builder.build_examples( + corpus, labels, diff_dir=diff_dir, fetch_missing=True + ) + second, _missing = builder.build_examples( + corpus, labels, diff_dir=diff_dir, fetch_missing=True + ) + first_bytes = json.dumps(first, sort_keys=True, allow_nan=False).encode() + second_bytes = json.dumps(second, sort_keys=True, allow_nan=False).encode() + self.assertEqual(first_bytes, second_bytes) + self.assertEqual(first[0]["metadata"]["diff_source"], "cache") + + def test_existing_download_destination_fails_before_network(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + destination = Path(temporary_name) / "cache.diff" + destination.write_bytes(b"trusted cache\n") + + with ( + patch.object(builder, "urlopen") as mocked_urlopen, + self.assertRaisesRegex(builder.DatasetError, "already exists"), + ): + builder.fetch_diff("https://github.com/owner/repo/pull/1.diff", destination) + + mocked_urlopen.assert_not_called() + self.assertEqual(destination.read_bytes(), b"trusted cache\n") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/benchmarks/test_collect_prs.py b/tests/benchmarks/test_collect_prs.py new file mode 100644 index 0000000..618d153 --- /dev/null +++ b/tests/benchmarks/test_collect_prs.py @@ -0,0 +1,338 @@ +"""Offline tests for the exploratory PR metadata collector.""" + +from __future__ import annotations + +import io +import json +import os +import sys +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from contextlib import redirect_stderr +from pathlib import Path +from typing import Any +from unittest.mock import patch + +from benchmarks.real_world import collect_prs as collector + +OID = "a" * 40 + + +def _fake_gh(path: Path, output: bytes) -> None: + path.write_text( + f"#!{sys.executable}\nimport os\nos.write(1, {output!r})\n", + encoding="utf-8", + ) + path.chmod(0o755) + + +def pr_summary(number: int = 1) -> dict[str, object]: + return { + "number": number, + "title": "title", + "url": f"https://github.com/owner/repo/pull/{number}", + "mergedAt": "2025-01-01T00:00:00Z", + "mergeCommit": {"oid": OID}, + "baseRefName": "main", + "headRefName": "feature", + "author": {"id": "user", "login": "author", "name": None, "is_bot": False}, + } + + +def pr_detail() -> dict[str, object]: + return { + "additions": 1, + "deletions": 0, + "changedFiles": 1, + "files": [{"path": "app.py", "additions": 1, "deletions": 0, "changeType": "MODIFIED"}], + "body": "body", + "commits": [{"oid": OID}], + } + + +class CollectorTests(unittest.TestCase): + def test_config_requires_positive_limit_and_nonempty_repository_objects(self) -> None: + invalid_configs = ( + {"prs_per_repository": 0, "repositories": [{"name": "owner/repo"}]}, + {"prs_per_repository": True, "repositories": [{"name": "owner/repo"}]}, + {"prs_per_repository": 1, "repositories": []}, + {"prs_per_repository": 1, "repositories": ["owner/repo"]}, + {"prs_per_repository": 1, "repositories": [{"name": " "}]}, + ) + with tempfile.TemporaryDirectory() as temporary_name: + path = Path(temporary_name) / "config.json" + for config in invalid_configs: + with self.subTest(config=config): + path.write_text(json.dumps(config), encoding="utf-8") + with self.assertRaises(collector.CollectorError): + collector._resolve_config(path, None, None) + + def test_cli_config_error_is_controlled_and_never_collects(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + config = root / "config.json" + output = root / "output.json" + config.write_text( + json.dumps({"prs_per_repository": -1, "repositories": []}), encoding="utf-8" + ) + + with ( + patch.object(collector, "collect_repository") as collect, + self.assertRaises(SystemExit) as raised, + ): + collector.main(["--config", str(config), "--output", str(output)]) + + self.assertEqual(raised.exception.code, 2) + collect.assert_not_called() + self.assertFalse(output.exists()) + + def test_invalid_config_utf8_is_a_controlled_cli_error(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + config = root / "config.json" + output = root / "output.json" + config.write_bytes(b'{"repositories": ["\xff"]}') + stderr = io.StringIO() + + with ( + redirect_stderr(stderr), + patch.object(collector, "collect_repository") as collect, + self.assertRaises(SystemExit) as raised, + ): + collector.main(["--config", str(config), "--output", str(output)]) + + self.assertEqual(raised.exception.code, 2) + self.assertIn("cannot read config", stderr.getvalue()) + self.assertNotIn("Traceback", stderr.getvalue()) + collect.assert_not_called() + self.assertFalse(output.exists()) + + def test_cli_overrides_are_validated_without_network(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + config = root / "config.json" + output = root / "output.json" + config.write_text( + json.dumps({"prs_per_repository": 20, "repositories": [{"name": "owner/repo"}]}), + encoding="utf-8", + ) + + with ( + patch.object(collector, "collect_repository") as collect, + self.assertRaises(SystemExit) as raised, + ): + collector.main( + [ + "--config", + str(config), + "--repository", + "", + "--output", + str(output), + ] + ) + + self.assertEqual(raised.exception.code, 2) + collect.assert_not_called() + + def test_strict_json_rejects_duplicate_keys_and_non_finite_values(self) -> None: + with self.assertRaisesRegex(collector.CollectorError, "duplicate JSON key"): + collector._strict_json_loads('{"limit":1,"limit":2}', "config") + with self.assertRaisesRegex(collector.CollectorError, "non-finite JSON number"): + collector._strict_json_loads('{"limit":Infinity}', "config") + + def test_gh_subprocess_decodes_utf8_independently_of_ascii_locale(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + _fake_gh(root / "gh", '{"title":"café"}\n'.encode()) + environment = { + "PATH": f"{root}{os.pathsep}{os.environ.get('PATH', '')}", + "LC_ALL": "C", + "LANG": "C", + "PYTHONUTF8": "0", + "PYTHONCOERCECLOCALE": "0", + } + with patch.dict(os.environ, environment): + self.assertEqual(collector.gh_json("api", "test"), {"title": "café"}) + + def test_invalid_gh_utf8_is_collector_error_and_controlled_cli_error(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + config = root / "config.json" + output = root / "output.json" + config.write_text( + json.dumps({"prs_per_repository": 1, "repositories": [{"name": "owner/repo"}]}), + encoding="utf-8", + ) + _fake_gh(root / "gh", b'{"title":"\xff"}\n') + environment = { + "PATH": f"{root}{os.pathsep}{os.environ.get('PATH', '')}", + "LC_ALL": "C", + "LANG": "C", + "PYTHONUTF8": "0", + "PYTHONCOERCECLOCALE": "0", + } + with ( + patch.dict(os.environ, environment), + self.assertRaisesRegex(collector.CollectorError, "not valid UTF-8") as raised, + ): + collector.gh_json("api", "test") + self.assertIsInstance(raised.exception.__cause__, UnicodeDecodeError) + + stderr = io.StringIO() + with ( + patch.dict(os.environ, environment), + redirect_stderr(stderr), + self.assertRaises(SystemExit) as exited, + ): + collector.main(["--config", str(config), "--output", str(output)]) + self.assertEqual(exited.exception.code, 2) + self.assertIn("gh response is not valid UTF-8", stderr.getvalue()) + self.assertNotIn("Traceback", stderr.getvalue()) + self.assertFalse(output.exists()) + + def test_publication_rejects_existing_protected_and_input_destinations(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + existing = root / "candidate.json" + existing.write_text("old\n", encoding="utf-8") + config = root / "config.json" + config.write_text("input\n", encoding="utf-8") + + with self.assertRaisesRegex(collector.CollectorError, "already exists"): + collector._publish_output(existing, {"entries": []}) + with self.assertRaisesRegex(collector.CollectorError, "overwrite input"): + collector._publish_output(config, {"entries": []}, input_paths=(config,)) + with self.assertRaisesRegex(collector.CollectorError, "frozen benchmark"): + collector._publish_output(collector.HERE / "corpus.json", {"entries": []}) + + self.assertEqual(existing.read_text(encoding="utf-8"), "old\n") + self.assertEqual(config.read_text(encoding="utf-8"), "input\n") + + def test_serialization_rejects_non_finite_values(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + output = Path(temporary_name) / "candidate.json" + + with self.assertRaisesRegex(collector.CollectorError, "strict JSON"): + collector._publish_output(output, {"value": float("nan")}) + + self.assertFalse(output.exists()) + + def test_collect_repository_validates_all_consumed_gh_fields(self) -> None: + malformed_responses = ( + ([{**pr_summary(), "number": True}],), + ([pr_summary(), pr_summary()],), + ([{**pr_summary(), "mergeCommit": {"oid": "bad"}}],), + ([pr_summary()], {**pr_detail(), "additions": -1}), + ([pr_summary()], {**pr_detail(), "changedFiles": "1"}), + ([pr_summary()], {**pr_detail(), "files": "app.py"}), + ( + [pr_summary()], + { + **pr_detail(), + "files": [ + { + "path": "", + "additions": 1, + "deletions": 0, + "changeType": "MODIFIED", + } + ], + }, + ), + ([pr_summary()], {**pr_detail(), "commits": [{"oid": "bad"}]}), + ([{**pr_summary(), "author": "author"}],), + ) + for responses in malformed_responses: + with ( + self.subTest(responses=responses), + patch.object(collector, "gh_json", side_effect=responses), + self.assertRaises(collector.CollectorError), + ): + collector.collect_repository("owner/repo", 20) + + def test_collect_repository_accepts_valid_offline_responses(self) -> None: + with patch.object(collector, "gh_json", side_effect=([pr_summary()], pr_detail())): + entries = collector.collect_repository("owner/repo", 20) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["number"], 1) + collector._validate_output_payload({"entries": entries}) + + def test_main_revalidates_assembled_payload_before_publication(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + config = root / "config.json" + output = root / "output.json" + config.write_text( + json.dumps({"prs_per_repository": 1, "repositories": [{"name": "owner/repo"}]}), + encoding="utf-8", + ) + malformed = [{"repository": "owner/repo", "number": True}] + with ( + patch.object(collector, "collect_repository", return_value=malformed), + self.assertRaises(SystemExit) as raised, + ): + collector.main(["--config", str(config), "--output", str(output)]) + self.assertEqual(raised.exception.code, 2) + self.assertFalse(output.exists()) + + def test_publication_parent_swap_cannot_redirect_collector_output(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + parent = root / "parent" + held = root / "held" + protected = root / "protected" + parent.mkdir() + protected.mkdir() + output = parent / "candidate.json" + real_link = os.link + + def swap_then_link(*args: Any, **kwargs: Any) -> None: + parent.rename(held) + parent.symlink_to(protected, target_is_directory=True) + real_link(*args, **kwargs) + + with patch("benchmarks.real_world._secure_publish.os.link", side_effect=swap_then_link): + collector._publish_output(output, {"value": 1}) + + self.assertEqual(json.loads((held / "candidate.json").read_text())["value"], 1) + self.assertFalse((protected / "candidate.json").exists()) + + def test_symlinked_parent_component_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + root = Path(temporary_name) + real_parent = root / "real" + real_parent.mkdir() + alias = root / "alias" + alias.symlink_to(real_parent, target_is_directory=True) + with self.assertRaises(collector.CollectorError): + collector._publish_output(alias / "candidate.json", {"value": 1}) + self.assertFalse((real_parent / "candidate.json").exists()) + + def test_collector_publication_race_has_exactly_one_winner(self) -> None: + with tempfile.TemporaryDirectory() as temporary_name: + output = Path(temporary_name) / "candidate.json" + barrier = threading.Barrier(2) + + def publish(value: int) -> str: + barrier.wait() + try: + collector._publish_output(output, {"value": value}) + except collector.CollectorError: + return "lost" + return "won" + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(publish, (1, 2))) + + self.assertEqual(sorted(results), ["lost", "won"]) + self.assertIn( + json.loads(output.read_text(encoding="utf-8"))["value"], + {1, 2}, + ) + + +if __name__ == "__main__": + unittest.main()