From acbbe5d5f740455934381ccf75fd369f5e1d13cc Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 07:31:18 +0000 Subject: [PATCH] feat(#339): add GRACE semantic code-contract anchor eval (slice 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure data + deterministic-logic layer for the GRACE eval harness. No external model calls in this slice. - grace_eval/schema.py: GraceFixture, VariantRunRecord, VariantAggregate, SweepFinding, GraceReport, aggregate_runs(), make_run_id() — six variants (baseline, inline, lex, min, inj, lie) with variant-set classification, per-run JSONL row schema, rolled-up stats, and vs-baseline delta notes (improvement/regression/tie/no_baseline). - grace_eval/sweep.py: sweep_source() / sweep_variant_sources() — heuristic stale-anchor detection via CONTRACT/ANCHOR comment patterns with 7 contradiction signal pairs and an 8-line lookahead window. - 97 tests across test_grace_eval_schema.py + test_grace_eval_sweep.py. - docs/ARCHITECTURE.md: mark #339 closed (slice 1). Co-Authored-By: Claude Sonnet 4.6 Claude-Session: https://claude.ai/code/session_011jmbBkJwJByEJGXFAihgh8 --- docs/ARCHITECTURE.md | 4 +- src/mapify_cli/grace_eval/__init__.py | 42 ++ src/mapify_cli/grace_eval/schema.py | 437 +++++++++++++++++++++ src/mapify_cli/grace_eval/sweep.py | 136 +++++++ tests/test_grace_eval_schema.py | 530 ++++++++++++++++++++++++++ tests/test_grace_eval_sweep.py | 271 +++++++++++++ 6 files changed, 1418 insertions(+), 2 deletions(-) create mode 100644 src/mapify_cli/grace_eval/__init__.py create mode 100644 src/mapify_cli/grace_eval/schema.py create mode 100644 src/mapify_cli/grace_eval/sweep.py create mode 100644 tests/test_grace_eval_schema.py create mode 100644 tests/test_grace_eval_sweep.py diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 905bffe6..6a8d25a0 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -2571,8 +2571,8 @@ All remaining open issues are enhancements (no bugs as of 2026-07-18). Prioritiz ### #353 — Eval-gated prompt profile canary and rollback — CLOSED (slice 1) **COMPLETE (PR #374, 2026-07-18)**: `mapify prompt-profile list` command shipped. Manifest format: `.map/prompt-profiles//manifest.json` (required: `id`, `title`, `version`; optional: `description`, `owner`, `targets`, `eval_requirements`, `rollback_notes`). Active pointer: `.map/prompt-profiles/active.json` (`{"active": ""|null}`). `list` command shows table with ID/title/version/status/description and active marker; `--json` for machine-readable output; stale-pointer warning when `active.json` names a non-existent profile. 20 tests in `tests/test_prompt_profile_commands.py`. Future slices: `diff`, `activate`, `rollback`, `report` commands + integration with #291 preset composition as rendering substrate. -### #339 — GRACE semantic code-contract anchor eval -Eval harness comparing inline code contracts vs prompt-injected vs no-anchor variants for bug-fix workflows. Variants: baseline, inline (LEX/MIN), prompt-injected (INJ), stale (LIE). First slice: fixture structure + report schema (JSON). No external model calls required for slice 1. +### #339 — GRACE semantic code-contract anchor eval — CLOSED (slice 1) +**COMPLETE (PR #375, 2026-07-18)**: `src/mapify_cli/grace_eval/` package shipped. Pure data + deterministic logic layer; no external model calls. Contracts: `GraceFixture` (fixture.json schema), `VariantRunRecord` (per-run JSONL row), `VariantAggregate` (rolled-up stats with vs-baseline deltas), `SweepFinding` (stale/contradictory contract detection result), `GraceReport` (root report object with schema_version). Six variants: `baseline`, `inline`, `lex`, `min`, `inj`, `lie`; variant sets classify each as `CODE_LOCAL`, `PROMPT_INJECTED`, or `NO_ANCHOR`. `aggregate_runs()` computes success_rate, mean_retries, mean tokens, stale_detections, and trajectory_delta_note (improvement/regression/tie/no_baseline) vs a baseline aggregate. `sweep_source()` and `sweep_variant_sources()` perform heuristic stale-anchor detection using `# CONTRACT:` / `# ANCHOR:` comment patterns and 7 contradiction signal pairs with an 8-line lookahead. 97 tests in `tests/test_grace_eval_schema.py` + `tests/test_grace_eval_sweep.py`. Future slices: CLI entry point (`mapify grace-eval run/report`), real token-log replay, model-call dispatcher, HTML report viewer. --- diff --git a/src/mapify_cli/grace_eval/__init__.py b/src/mapify_cli/grace_eval/__init__.py new file mode 100644 index 00000000..07eeecf2 --- /dev/null +++ b/src/mapify_cli/grace_eval/__init__.py @@ -0,0 +1,42 @@ +"""grace_eval — GRACE semantic code-contract anchor evaluation. + +Exports the shared data contracts and sweep utilities. This module is a +pure data + deterministic-logic layer for slice 1; no external model calls, +no I/O, no subprocess, no clock access (INV-2/INV-3). +""" + +from __future__ import annotations + +from mapify_cli.grace_eval.schema import ( + VARIANT_NAMES, + CODE_LOCAL_VARIANTS, + PROMPT_INJECTED_VARIANTS, + NO_ANCHOR_VARIANTS, + GraceFixture, + GraceReport, + SweepFinding, + VariantAggregate, + VariantRunRecord, + aggregate_runs, + make_run_id, +) +from mapify_cli.grace_eval.sweep import ( + sweep_source, + sweep_variant_sources, +) + +__all__ = [ + "VARIANT_NAMES", + "CODE_LOCAL_VARIANTS", + "PROMPT_INJECTED_VARIANTS", + "NO_ANCHOR_VARIANTS", + "GraceFixture", + "GraceReport", + "SweepFinding", + "VariantAggregate", + "VariantRunRecord", + "aggregate_runs", + "make_run_id", + "sweep_source", + "sweep_variant_sources", +] diff --git a/src/mapify_cli/grace_eval/schema.py b/src/mapify_cli/grace_eval/schema.py new file mode 100644 index 00000000..2a82fd2b --- /dev/null +++ b/src/mapify_cli/grace_eval/schema.py @@ -0,0 +1,437 @@ +"""Data contracts for the GRACE semantic code-contract anchor eval. + +Pure data layer — no I/O, no dispatch, no subprocess, no clock/random. +Producers and consumers both import from here (contract-first pattern). + +GRACE variants (from the upstream experiment notes in issue #339): + - baseline : no semantic anchors beyond normal code/docs + - inline : code-local contracts near relevant implementation points + - lex : inline contracts enriched with domain terms / bug symptoms + - min : trap-only anchors around non-obvious or bug-prone paths + - inj : top contracts injected into prompt, not placed near code + - lie : controlled stale/false-anchor variant for sweep/audit tests +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +_SCHEMA_VERSION = "1.0" +_SENTINEL: object = object() + +# Canonical variant names in canonical order. +VARIANT_NAMES: tuple[str, ...] = ( + "baseline", + "inline", + "lex", + "min", + "inj", + "lie", +) + +# Anchor-placement strategies (split used in side-by-side reporting). +CODE_LOCAL_VARIANTS: frozenset[str] = frozenset({"inline", "lex", "min", "lie"}) +PROMPT_INJECTED_VARIANTS: frozenset[str] = frozenset({"inj"}) +NO_ANCHOR_VARIANTS: frozenset[str] = frozenset({"baseline"}) + + +# --------------------------------------------------------------------------- +# GraceFixture — defines one bug-fix task +# --------------------------------------------------------------------------- + + +@dataclass +class GraceFixture: + """Description of one bug-fix eval task. + + Loaded from ``/fixture.json``; the variant subdirectories + (``variants//``) supply the annotated source files for code-local + variants, and ``variants/inj/prompt_fragment.txt`` supplies the contract + text for the ``inj`` variant. + + ``expected_changed_files`` is a list of repo-relative glob patterns that + the agent is expected to modify when the fix is complete. Used by the + (future) deterministic scope gate — not validated in slice 1. + """ + + fixture_id: str + title: str + description: str + bug_summary: str + expected_changed_files: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + + def __post_init__(self) -> None: + if not isinstance(self.fixture_id, str) or not self.fixture_id: + raise ValueError("GraceFixture.fixture_id must be a non-empty str") + if not isinstance(self.title, str) or not self.title: + raise ValueError("GraceFixture.title must be a non-empty str") + if not isinstance(self.description, str): + raise ValueError("GraceFixture.description must be str") + if not isinstance(self.bug_summary, str): + raise ValueError("GraceFixture.bug_summary must be str") + if not isinstance(self.expected_changed_files, list): + raise ValueError("GraceFixture.expected_changed_files must be a list") + if not isinstance(self.tags, list): + raise ValueError("GraceFixture.tags must be a list") + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": _SCHEMA_VERSION, + "fixture_id": self.fixture_id, + "title": self.title, + "description": self.description, + "bug_summary": self.bug_summary, + "expected_changed_files": list(self.expected_changed_files), + "tags": list(self.tags), + } + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "GraceFixture": + return cls( + fixture_id=str(d["fixture_id"]), + title=str(d["title"]), + description=str(d.get("description", "")), + bug_summary=str(d.get("bug_summary", "")), + expected_changed_files=list(d.get("expected_changed_files", [])), + tags=list(d.get("tags", [])), + ) + + +# --------------------------------------------------------------------------- +# VariantRunRecord — one (fixture, variant, run) result row +# --------------------------------------------------------------------------- + + +@dataclass +class VariantRunRecord: + """One completed GRACE run result. + + Append-only JSONL row keyed by ``run_id``. In slice 1 this is populated + by fixture-driven mock runners only; in later slices a real dispatcher + fills ``total_tokens``, ``repeated_reads``, and ``retry_count`` from + token_log.jsonl and ``step_state.json``. + + ``stale_detection`` is meaningful only for the ``lie`` variant: True when + the sweep flagged a contradicted or stale contract in the variant source. + """ + + run_id: str + fixture_id: str + variant: str + success: bool + retry_count: int = 0 + total_tokens: int = 0 + output_tokens: int = 0 + repeated_reads: int = 0 + stale_detection: bool = False + error: str | None = None + + def __post_init__(self) -> None: + if self.variant not in VARIANT_NAMES: + raise ValueError( + f"VariantRunRecord.variant must be one of {VARIANT_NAMES}, " + f"got {self.variant!r}" + ) + if not isinstance(self.run_id, str) or not self.run_id: + raise ValueError("VariantRunRecord.run_id must be a non-empty str") + if not isinstance(self.fixture_id, str) or not self.fixture_id: + raise ValueError("VariantRunRecord.fixture_id must be a non-empty str") + if not isinstance(self.retry_count, int) or self.retry_count < 0: + raise ValueError("VariantRunRecord.retry_count must be a non-negative int") + if not isinstance(self.total_tokens, int) or self.total_tokens < 0: + raise ValueError("VariantRunRecord.total_tokens must be a non-negative int") + + def to_dict(self) -> dict[str, Any]: + return { + "run_id": self.run_id, + "fixture_id": self.fixture_id, + "variant": self.variant, + "success": bool(self.success), + "retry_count": self.retry_count, + "total_tokens": self.total_tokens, + "output_tokens": self.output_tokens, + "repeated_reads": self.repeated_reads, + "stale_detection": bool(self.stale_detection), + "error": self.error, + } + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "VariantRunRecord": + raw_err = d.get("error", _SENTINEL) + error: str | None = None if (raw_err is _SENTINEL or raw_err is None) else str(raw_err) + return cls( + run_id=str(d["run_id"]), + fixture_id=str(d["fixture_id"]), + variant=str(d["variant"]), + success=bool(d["success"]), + retry_count=int(d.get("retry_count", 0)), + total_tokens=int(d.get("total_tokens", 0)), + output_tokens=int(d.get("output_tokens", 0)), + repeated_reads=int(d.get("repeated_reads", 0)), + stale_detection=bool(d.get("stale_detection", False)), + error=error, + ) + + +# --------------------------------------------------------------------------- +# VariantAggregate — rolled-up stats across N runs for one variant +# --------------------------------------------------------------------------- + + +@dataclass +class VariantAggregate: + """Aggregated statistics for one variant across all runs. + + ``vs_baseline`` is ``None`` for the baseline variant itself (no self-diff). + ``trajectory_delta_note`` classifies the token-trajectory change relative + to baseline: ``improvement``, ``regression``, ``tie``, or ``no_baseline``. + """ + + variant: str + n: int + success_rate: float + mean_retries: float + mean_total_tokens: float + mean_repeated_reads: float + stale_detections: int + vs_baseline_correctness_delta: float | None = None + vs_baseline_tokens_delta: float | None = None + trajectory_delta_note: str = "no_baseline" + + def __post_init__(self) -> None: + if self.variant not in VARIANT_NAMES: + raise ValueError( + f"VariantAggregate.variant must be one of {VARIANT_NAMES}, " + f"got {self.variant!r}" + ) + valid_notes = {"improvement", "regression", "tie", "no_baseline"} + if self.trajectory_delta_note not in valid_notes: + raise ValueError( + f"VariantAggregate.trajectory_delta_note must be one of {valid_notes}, " + f"got {self.trajectory_delta_note!r}" + ) + + def to_dict(self) -> dict[str, Any]: + return { + "variant": self.variant, + "n": self.n, + "success_rate": round(self.success_rate, 4), + "mean_retries": round(self.mean_retries, 4), + "mean_total_tokens": round(self.mean_total_tokens, 2), + "mean_repeated_reads": round(self.mean_repeated_reads, 4), + "stale_detections": self.stale_detections, + "vs_baseline_correctness_delta": ( + round(self.vs_baseline_correctness_delta, 4) + if self.vs_baseline_correctness_delta is not None + else None + ), + "vs_baseline_tokens_delta": ( + round(self.vs_baseline_tokens_delta, 2) + if self.vs_baseline_tokens_delta is not None + else None + ), + "trajectory_delta_note": self.trajectory_delta_note, + } + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "VariantAggregate": + raw_cd = d.get("vs_baseline_correctness_delta", _SENTINEL) + cd: float | None = None if (raw_cd is _SENTINEL or raw_cd is None) else float(raw_cd) + raw_td = d.get("vs_baseline_tokens_delta", _SENTINEL) + td: float | None = None if (raw_td is _SENTINEL or raw_td is None) else float(raw_td) + return cls( + variant=str(d["variant"]), + n=int(d["n"]), + success_rate=float(d["success_rate"]), + mean_retries=float(d["mean_retries"]), + mean_total_tokens=float(d["mean_total_tokens"]), + mean_repeated_reads=float(d["mean_repeated_reads"]), + stale_detections=int(d.get("stale_detections", 0)), + vs_baseline_correctness_delta=cd, + vs_baseline_tokens_delta=td, + trajectory_delta_note=str(d.get("trajectory_delta_note", "no_baseline")), + ) + + +# --------------------------------------------------------------------------- +# SweepFinding — stale/contradictory contract detected by the sweep +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class SweepFinding: + """One stale or contradictory contract detected by the sweep checker. + + ``severity``: ``warning`` (mild drift) or ``critical`` (outright contradiction). + ``location``: file path + line range, e.g. ``"src/utils.py:L42-45"``. + ``detail``: human-readable description of the conflict. + ``variant``: which variant source the finding was detected in. + """ + + severity: str + location: str + detail: str + variant: str + + def __post_init__(self) -> None: + if self.severity not in ("warning", "critical"): + raise ValueError( + f"SweepFinding.severity must be warning|critical, got {self.severity!r}" + ) + if self.variant not in VARIANT_NAMES: + raise ValueError( + f"SweepFinding.variant must be one of {VARIANT_NAMES}, " + f"got {self.variant!r}" + ) + + def to_dict(self) -> dict[str, Any]: + return { + "severity": self.severity, + "location": self.location, + "detail": self.detail, + "variant": self.variant, + } + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "SweepFinding": + return cls( + severity=str(d["severity"]), + location=str(d["location"]), + detail=str(d["detail"]), + variant=str(d["variant"]), + ) + + +# --------------------------------------------------------------------------- +# GraceReport — root report object +# --------------------------------------------------------------------------- + + +@dataclass +class GraceReport: + """Root GRACE eval report object. + + Written to ``.map/grace-eval//report.json`` after an eval run. + ``aggregates`` is keyed by variant name in VARIANT_NAMES order. + ``sweep_findings`` lists stale/contradictory contract detections (empty when + the sweep finds no issues). + ``generated_at`` is an ISO-8601 timestamp supplied by the CLI boundary + (not by this module — INV-2: no clock access in the data layer). + """ + + fixture_id: str + generated_at: str + aggregates: list[VariantAggregate] + sweep_findings: list[SweepFinding] = field(default_factory=list) + schema_version: str = _SCHEMA_VERSION + notes: str = "" + + def aggregate_for(self, variant: str) -> VariantAggregate | None: + for agg in self.aggregates: + if agg.variant == variant: + return agg + return None + + @property + def n_stale_detections(self) -> int: + return sum(agg.stale_detections for agg in self.aggregates) + + @property + def n_sweep_findings(self) -> int: + return len(self.sweep_findings) + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": self.schema_version, + "fixture_id": self.fixture_id, + "generated_at": self.generated_at, + "aggregates": [a.to_dict() for a in self.aggregates], + "sweep_findings": [f.to_dict() for f in self.sweep_findings], + "notes": self.notes, + } + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "GraceReport": + return cls( + schema_version=str(d.get("schema_version", _SCHEMA_VERSION)), + fixture_id=str(d["fixture_id"]), + generated_at=str(d["generated_at"]), + aggregates=[VariantAggregate.from_dict(a) for a in d.get("aggregates", [])], + sweep_findings=[SweepFinding.from_dict(f) for f in d.get("sweep_findings", [])], + notes=str(d.get("notes", "")), + ) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_run_id(fixture_id: str, variant: str, run: int) -> str: + """Deterministic, human-readable run identifier. + + Example: ``make_run_id("off-by-one", "lex", 2)`` → ``"off-by-one-lex-r2"`` + """ + slug = "".join(ch if ch.isalnum() or ch == "-" else "-" for ch in fixture_id) + return f"{slug}-{variant}-r{run}" + + +def aggregate_runs( + records: list[VariantRunRecord], + baseline_agg: VariantAggregate | None = None, +) -> VariantAggregate: + """Aggregate a list of same-fixture, same-variant run records. + + All records must share the same ``variant`` and ``fixture_id``. + Raises ``ValueError`` on empty list or mixed variants. + ``baseline_agg`` supplies the baseline aggregate used for delta fields; pass + ``None`` when aggregating the baseline variant itself. + """ + if not records: + raise ValueError("aggregate_runs requires at least one record") + variants = {r.variant for r in records} + if len(variants) > 1: + raise ValueError(f"aggregate_runs: mixed variants {variants!r}") + variant = records[0].variant + + n = len(records) + success_rate = sum(1 for r in records if r.success) / n + mean_retries = sum(r.retry_count for r in records) / n + mean_total_tokens = sum(r.total_tokens for r in records) / n + mean_repeated_reads = sum(r.repeated_reads for r in records) / n + stale_detections = sum(1 for r in records if r.stale_detection) + + vs_cd: float | None = None + vs_td: float | None = None + note = "no_baseline" + + if baseline_agg is not None and variant != "baseline": + vs_cd = round(success_rate - baseline_agg.success_rate, 4) + vs_td = round(mean_total_tokens - baseline_agg.mean_total_tokens, 2) + _TOKEN_TIE_EPSILON = 500.0 + _TOKEN_REGRESSION_DELTA = 2000.0 + if abs(vs_td) < _TOKEN_TIE_EPSILON: + note = "tie" + elif vs_td <= -_TOKEN_REGRESSION_DELTA: + note = "improvement" + elif vs_td >= _TOKEN_REGRESSION_DELTA: + note = "regression" + else: + note = "tie" + elif variant == "baseline": + note = "no_baseline" + + return VariantAggregate( + variant=variant, + n=n, + success_rate=success_rate, + mean_retries=mean_retries, + mean_total_tokens=mean_total_tokens, + mean_repeated_reads=mean_repeated_reads, + stale_detections=stale_detections, + vs_baseline_correctness_delta=vs_cd, + vs_baseline_tokens_delta=vs_td, + trajectory_delta_note=note, + ) diff --git a/src/mapify_cli/grace_eval/sweep.py b/src/mapify_cli/grace_eval/sweep.py new file mode 100644 index 00000000..6c24555a --- /dev/null +++ b/src/mapify_cli/grace_eval/sweep.py @@ -0,0 +1,136 @@ +"""Contract sweep / stale-anchor audit for GRACE eval. + +Pure, deterministic, no I/O, no subprocess, no clock/random. +Detects contradictions or stale contracts in annotated source text. + +A GRACE contract anchor is a comment matching the pattern: + + # CONTRACT: + # ANCHOR: + +The sweep extracts these anchors and checks whether the immediately-following +code lines contradict the claim in a few heuristic ways (see ``_CONTRADICTION_PATTERNS``). +This is intentionally lightweight — the goal is to find obviously wrong +contracts quickly, not to verify semantic correctness. + +For the ``lie`` fixture variant, stale/false anchors are injected +intentionally; the sweep should detect them. For all other variants the +sweep acts as a quality gate: if contracts exist but are clearly wrong, +``SweepFinding`` objects with ``severity="critical"`` are returned. +""" + +from __future__ import annotations + +import re +from typing import Any + +from mapify_cli.grace_eval.schema import SweepFinding, VARIANT_NAMES + +# Lines we recognise as contract anchor comments. +_ANCHOR_RE = re.compile( + r"^\s*#\s*(?:CONTRACT|ANCHOR):\s*(.+)$", + re.IGNORECASE, +) + +# Heuristic contradiction signals: (anchor_keyword, code_counter_pattern) +# If the anchor claim contains `anchor_keyword` and the following N lines +# match `code_counter_pattern`, we flag a likely contradiction. +_CONTRADICTION_SIGNALS: tuple[tuple[str, re.Pattern[str]], ...] = ( + ("never returns none", re.compile(r"\breturn\s+None\b")), + ("always raises", re.compile(r"\breturn\b(?!\s+None)")), + ("idempotent", re.compile(r"\bappend\s*\(|\.extend\s*\(")), + ("no side effect", re.compile(r"\bself\.\w+\s*=")), + ("thread.safe", re.compile(r"\bself\.\w+\s*\+?=|list\b|dict\b")), + ("returns true", re.compile(r"\breturn\s+False\b")), + ("returns false", re.compile(r"\breturn\s+True\b")), +) + +_LOOKAHEAD_LINES = 8 + + +def _extract_anchors( + lines: list[str], +) -> list[tuple[int, str]]: + """Return (line_number_1indexed, claim) pairs for each anchor found.""" + found = [] + for i, line in enumerate(lines, start=1): + m = _ANCHOR_RE.match(line) + if m: + found.append((i, m.group(1).strip())) + return found + + +def _check_claim_vs_lookahead( + claim: str, + lookahead: list[str], +) -> str | None: + """Return a contradiction description or None if none detected.""" + claim_lower = claim.lower() + for keyword, pattern in _CONTRADICTION_SIGNALS: + if keyword in claim_lower: + for line in lookahead: + if pattern.search(line): + return ( + f"anchor claims '{keyword}' but code contains '{line.strip()}'" + ) + return None + + +def sweep_source( + source: str, + *, + variant: str, + location_prefix: str = "", +) -> list[SweepFinding]: + """Sweep a single source file's text for stale/contradictory contracts. + + ``variant`` is stored on each finding. ``location_prefix`` is a + human-readable path prefix prepended to the line-range in + ``SweepFinding.location`` (e.g. ``"src/utils.py"``). + Returns a (possibly empty) list of ``SweepFinding`` objects. + """ + if variant not in VARIANT_NAMES: + raise ValueError( + f"sweep_source: variant must be one of {VARIANT_NAMES}, got {variant!r}" + ) + lines = source.splitlines() + anchors = _extract_anchors(lines) + findings: list[SweepFinding] = [] + + for lineno, claim in anchors: + lookahead_start = lineno # lines is 0-indexed, lineno is 1-indexed + lookahead_end = min(lineno + _LOOKAHEAD_LINES, len(lines)) + lookahead = lines[lookahead_start:lookahead_end] + contradiction = _check_claim_vs_lookahead(claim, lookahead) + if contradiction: + loc = f"{location_prefix}:L{lineno}" if location_prefix else f"L{lineno}" + findings.append( + SweepFinding( + severity="critical", + location=loc, + detail=f"Contract '{claim}' — {contradiction}", + variant=variant, + ) + ) + + return findings + + +def sweep_variant_sources( + variant_sources: dict[str, Any], + *, + variant: str, +) -> list[SweepFinding]: + """Sweep all source files in a variant dict. + + ``variant_sources`` maps ``location_prefix`` → ``source_text`` (str). + Returns the union of findings across all files. + """ + all_findings: list[SweepFinding] = [] + for location, source in variant_sources.items(): + if not isinstance(source, str): + continue + all_findings.extend( + sweep_source(source, variant=variant, location_prefix=location) + ) + return all_findings diff --git a/tests/test_grace_eval_schema.py b/tests/test_grace_eval_schema.py new file mode 100644 index 00000000..74eac9a4 --- /dev/null +++ b/tests/test_grace_eval_schema.py @@ -0,0 +1,530 @@ +"""Tests for grace_eval.schema data contracts (#339). + +Covers: + GE1 — variant enumeration is complete and ordered + GE2 — GraceFixture round-trip and validation + GE3 — VariantRunRecord round-trip and validation + GE4 — VariantAggregate round-trip and validation + GE5 — SweepFinding round-trip and validation + GE6 — GraceReport round-trip and helpers + GE7 — make_run_id is deterministic and unique + GE8 — aggregate_runs computes correct stats + GE9 — aggregate_runs baseline delta computation +""" + +from __future__ import annotations + +import json +import pytest + +from mapify_cli.grace_eval.schema import ( + VARIANT_NAMES, + CODE_LOCAL_VARIANTS, + PROMPT_INJECTED_VARIANTS, + NO_ANCHOR_VARIANTS, + GraceFixture, + GraceReport, + SweepFinding, + VariantAggregate, + VariantRunRecord, + aggregate_runs, + make_run_id, +) + + +# --------------------------------------------------------------------------- +# GE1 — variant enumeration +# --------------------------------------------------------------------------- + + +class TestGe1VariantEnumeration: + def test_all_six_variants_present(self) -> None: + assert set(VARIANT_NAMES) == {"baseline", "inline", "lex", "min", "inj", "lie"} + + def test_variant_names_ordered(self) -> None: + assert VARIANT_NAMES == ("baseline", "inline", "lex", "min", "inj", "lie") + + def test_variant_sets_partition(self) -> None: + # baseline, inline, lex, min, inj, lie — all covered + assert "baseline" in NO_ANCHOR_VARIANTS + assert "inline" in CODE_LOCAL_VARIANTS + assert "lex" in CODE_LOCAL_VARIANTS + assert "min" in CODE_LOCAL_VARIANTS + assert "inj" in PROMPT_INJECTED_VARIANTS + assert "lie" in CODE_LOCAL_VARIANTS + # no variant is in two sets at once + assert not (CODE_LOCAL_VARIANTS & PROMPT_INJECTED_VARIANTS) + assert not (CODE_LOCAL_VARIANTS & NO_ANCHOR_VARIANTS) + assert not (PROMPT_INJECTED_VARIANTS & NO_ANCHOR_VARIANTS) + + def test_lie_in_code_local(self) -> None: + assert "lie" in CODE_LOCAL_VARIANTS + + def test_inj_in_prompt_injected(self) -> None: + assert "inj" in PROMPT_INJECTED_VARIANTS + + +# --------------------------------------------------------------------------- +# GE2 — GraceFixture +# --------------------------------------------------------------------------- + + +class TestGe2GraceFixture: + def _make(self) -> GraceFixture: + return GraceFixture( + fixture_id="off-by-one", + title="Off-by-one in range check", + description="The range bound is exclusive but code treats it as inclusive.", + bug_summary="fence-post error in validate_range()", + expected_changed_files=["src/utils.py"], + tags=["off-by-one", "boundary"], + ) + + def test_round_trip(self) -> None: + f = self._make() + rt = GraceFixture.from_dict(f.to_dict()) + assert rt.fixture_id == f.fixture_id + assert rt.title == f.title + assert rt.description == f.description + assert rt.bug_summary == f.bug_summary + assert rt.expected_changed_files == f.expected_changed_files + assert rt.tags == f.tags + + def test_json_serialisable(self) -> None: + f = self._make() + raw = json.dumps(f.to_dict()) + data = json.loads(raw) + rt = GraceFixture.from_dict(data) + assert rt.fixture_id == "off-by-one" + + def test_schema_version_present(self) -> None: + f = self._make() + assert "schema_version" in f.to_dict() + + def test_invalid_empty_fixture_id(self) -> None: + with pytest.raises(ValueError, match="fixture_id"): + GraceFixture(fixture_id="", title="T", description="", bug_summary="") + + def test_invalid_empty_title(self) -> None: + with pytest.raises(ValueError, match="title"): + GraceFixture(fixture_id="x", title="", description="", bug_summary="") + + def test_optional_fields_default(self) -> None: + f = GraceFixture(fixture_id="x", title="T", description="", bug_summary="") + assert f.expected_changed_files == [] + assert f.tags == [] + + def test_from_dict_tolerates_missing_optional(self) -> None: + f = GraceFixture.from_dict({"fixture_id": "x", "title": "T"}) + assert f.description == "" + assert f.expected_changed_files == [] + + +# --------------------------------------------------------------------------- +# GE3 — VariantRunRecord +# --------------------------------------------------------------------------- + + +class TestGe3VariantRunRecord: + def _make(self, variant: str = "baseline") -> VariantRunRecord: + return VariantRunRecord( + run_id="off-by-one-baseline-r0", + fixture_id="off-by-one", + variant=variant, + success=True, + retry_count=1, + total_tokens=4500, + output_tokens=800, + repeated_reads=2, + stale_detection=False, + ) + + def test_round_trip(self) -> None: + r = self._make() + rt = VariantRunRecord.from_dict(r.to_dict()) + assert rt.run_id == r.run_id + assert rt.fixture_id == r.fixture_id + assert rt.variant == r.variant + assert rt.success == r.success + assert rt.retry_count == r.retry_count + assert rt.total_tokens == r.total_tokens + assert rt.output_tokens == r.output_tokens + assert rt.repeated_reads == r.repeated_reads + assert rt.stale_detection == r.stale_detection + assert rt.error is None + + def test_json_serialisable(self) -> None: + r = self._make() + data = json.loads(json.dumps(r.to_dict())) + rt = VariantRunRecord.from_dict(data) + assert rt.total_tokens == 4500 + + def test_stale_detection_lie_variant(self) -> None: + r2 = VariantRunRecord( + run_id="x-lie-r0", fixture_id="x", variant="lie", + success=False, stale_detection=True, + ) + assert r2.stale_detection is True + d = r2.to_dict() + rt = VariantRunRecord.from_dict(d) + assert rt.stale_detection is True + + def test_invalid_variant(self) -> None: + with pytest.raises(ValueError, match="variant"): + VariantRunRecord(run_id="x", fixture_id="f", variant="unknown", success=True) + + def test_invalid_negative_retry(self) -> None: + with pytest.raises(ValueError, match="retry_count"): + VariantRunRecord( + run_id="x", fixture_id="f", variant="baseline", + success=True, retry_count=-1, + ) + + def test_error_field_round_trip(self) -> None: + r = VariantRunRecord( + run_id="x", fixture_id="f", variant="lex", + success=False, error="timeout after 30s", + ) + rt = VariantRunRecord.from_dict(r.to_dict()) + assert rt.error == "timeout after 30s" + + @pytest.mark.parametrize("variant", VARIANT_NAMES) + def test_all_variants_accepted(self, variant: str) -> None: + r = VariantRunRecord(run_id=f"x-{variant}-r0", fixture_id="x", variant=variant, success=True) + assert r.variant == variant + + +# --------------------------------------------------------------------------- +# GE4 — VariantAggregate +# --------------------------------------------------------------------------- + + +class TestGe4VariantAggregate: + def _make(self, variant: str = "baseline") -> VariantAggregate: + return VariantAggregate( + variant=variant, + n=5, + success_rate=0.8, + mean_retries=0.4, + mean_total_tokens=5000.0, + mean_repeated_reads=1.2, + stale_detections=0, + ) + + def test_round_trip(self) -> None: + a = self._make() + rt = VariantAggregate.from_dict(a.to_dict()) + assert rt.variant == a.variant + assert rt.n == a.n + assert abs(rt.success_rate - a.success_rate) < 1e-6 + assert rt.trajectory_delta_note == "no_baseline" + + def test_json_serialisable(self) -> None: + a = self._make("lex") + data = json.loads(json.dumps(a.to_dict())) + rt = VariantAggregate.from_dict(data) + assert rt.variant == "lex" + + def test_invalid_trajectory_note(self) -> None: + with pytest.raises(ValueError, match="trajectory_delta_note"): + VariantAggregate( + variant="lex", n=1, success_rate=1.0, mean_retries=0.0, + mean_total_tokens=0.0, mean_repeated_reads=0.0, stale_detections=0, + trajectory_delta_note="unknown", + ) + + def test_delta_fields_nullable(self) -> None: + a = self._make() + d = a.to_dict() + assert d["vs_baseline_correctness_delta"] is None + assert d["vs_baseline_tokens_delta"] is None + + def test_delta_fields_non_null_for_non_baseline(self) -> None: + a = VariantAggregate( + variant="lex", n=3, success_rate=0.9, mean_retries=0.2, + mean_total_tokens=4000.0, mean_repeated_reads=0.8, stale_detections=0, + vs_baseline_correctness_delta=0.1, vs_baseline_tokens_delta=-500.0, + trajectory_delta_note="tie", + ) + d = a.to_dict() + rt = VariantAggregate.from_dict(d) + assert rt.vs_baseline_correctness_delta is not None + assert abs(rt.vs_baseline_correctness_delta - 0.1) < 1e-6 + + +# --------------------------------------------------------------------------- +# GE5 — SweepFinding +# --------------------------------------------------------------------------- + + +class TestGe5SweepFinding: + def test_round_trip(self) -> None: + f = SweepFinding( + severity="critical", + location="src/utils.py:L42", + detail="Contract 'never returns None' — return None found", + variant="lie", + ) + rt = SweepFinding.from_dict(f.to_dict()) + assert rt.severity == "critical" + assert rt.location == "src/utils.py:L42" + assert rt.variant == "lie" + + def test_warning_severity_accepted(self) -> None: + f = SweepFinding(severity="warning", location="L1", detail="d", variant="inline") + assert f.severity == "warning" + + def test_invalid_severity(self) -> None: + with pytest.raises(ValueError, match="severity"): + SweepFinding(severity="error", location="L1", detail="d", variant="lex") + + def test_invalid_variant(self) -> None: + with pytest.raises(ValueError, match="variant"): + SweepFinding(severity="warning", location="L1", detail="d", variant="unknown") + + def test_frozen(self) -> None: + f = SweepFinding(severity="warning", location="L1", detail="d", variant="min") + with pytest.raises((AttributeError, TypeError)): + f.severity = "critical" # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# GE6 — GraceReport +# --------------------------------------------------------------------------- + + +class TestGe6GraceReport: + def _make(self) -> GraceReport: + agg = VariantAggregate( + variant="baseline", n=3, success_rate=0.67, mean_retries=0.5, + mean_total_tokens=5000.0, mean_repeated_reads=1.0, stale_detections=0, + ) + finding = SweepFinding( + severity="critical", location="src/x.py:L10", + detail="stale anchor", variant="lie", + ) + return GraceReport( + fixture_id="off-by-one", + generated_at="2026-07-18T12:00:00Z", + aggregates=[agg], + sweep_findings=[finding], + ) + + def test_round_trip(self) -> None: + r = self._make() + rt = GraceReport.from_dict(r.to_dict()) + assert rt.fixture_id == "off-by-one" + assert rt.generated_at == "2026-07-18T12:00:00Z" + assert len(rt.aggregates) == 1 + assert len(rt.sweep_findings) == 1 + + def test_json_serialisable(self) -> None: + r = self._make() + data = json.loads(json.dumps(r.to_dict())) + rt = GraceReport.from_dict(data) + assert rt.fixture_id == "off-by-one" + + def test_aggregate_for_helper(self) -> None: + r = self._make() + a = r.aggregate_for("baseline") + assert a is not None + assert a.variant == "baseline" + assert r.aggregate_for("lex") is None + + def test_n_sweep_findings(self) -> None: + r = self._make() + assert r.n_sweep_findings == 1 + + def test_n_stale_detections(self) -> None: + r = self._make() + assert r.n_stale_detections == 0 + + def test_schema_version_preserved(self) -> None: + r = self._make() + d = r.to_dict() + assert d["schema_version"] == "1.0" + rt = GraceReport.from_dict(d) + assert rt.schema_version == "1.0" + + def test_empty_aggregates_and_findings(self) -> None: + r = GraceReport( + fixture_id="x", generated_at="2026-01-01T00:00:00Z", aggregates=[] + ) + assert r.n_sweep_findings == 0 + assert r.n_stale_detections == 0 + rt = GraceReport.from_dict(r.to_dict()) + assert rt.aggregates == [] + + +# --------------------------------------------------------------------------- +# GE7 — make_run_id +# --------------------------------------------------------------------------- + + +class TestGe7MakeRunId: + def test_format(self) -> None: + rid = make_run_id("off-by-one", "lex", 2) + assert rid == "off-by-one-lex-r2" + + def test_deterministic(self) -> None: + assert make_run_id("f", "baseline", 0) == make_run_id("f", "baseline", 0) + + def test_unique_across_variants(self) -> None: + ids = {make_run_id("fix", v, 0) for v in VARIANT_NAMES} + assert len(ids) == len(VARIANT_NAMES) + + def test_unique_across_runs(self) -> None: + ids = {make_run_id("fix", "lex", r) for r in range(5)} + assert len(ids) == 5 + + def test_special_chars_slugified(self) -> None: + rid = make_run_id("my fixture/test", "min", 1) + assert "/" not in rid + assert "min" in rid + assert "r1" in rid + + +# --------------------------------------------------------------------------- +# GE8 — aggregate_runs basic stats +# --------------------------------------------------------------------------- + + +class TestGe8AggregateRuns: + def _records(self, successes: list[bool], variant: str = "baseline") -> list[VariantRunRecord]: + return [ + VariantRunRecord( + run_id=make_run_id("f", variant, i), + fixture_id="f", + variant=variant, + success=s, + retry_count=1 if not s else 0, + total_tokens=4000 + i * 100, + output_tokens=500, + repeated_reads=1, + ) + for i, s in enumerate(successes) + ] + + def test_success_rate(self) -> None: + recs = self._records([True, True, False, True, False]) + agg = aggregate_runs(recs) + assert abs(agg.success_rate - 0.6) < 1e-6 + + def test_mean_retries(self) -> None: + recs = self._records([True, False]) + agg = aggregate_runs(recs) + assert abs(agg.mean_retries - 0.5) < 1e-6 + + def test_n(self) -> None: + recs = self._records([True, True, True]) + agg = aggregate_runs(recs) + assert agg.n == 3 + + def test_mean_total_tokens(self) -> None: + recs = self._records([True, True]) + # tokens: 4000, 4100 + agg = aggregate_runs(recs) + assert abs(agg.mean_total_tokens - 4050.0) < 1e-6 + + def test_all_success(self) -> None: + recs = self._records([True, True, True]) + agg = aggregate_runs(recs) + assert agg.success_rate == 1.0 + + def test_all_failure(self) -> None: + recs = self._records([False, False]) + agg = aggregate_runs(recs) + assert agg.success_rate == 0.0 + + def test_empty_raises(self) -> None: + with pytest.raises(ValueError, match="at least one"): + aggregate_runs([]) + + def test_mixed_variants_raises(self) -> None: + r1 = VariantRunRecord(run_id="a", fixture_id="f", variant="baseline", success=True) + r2 = VariantRunRecord(run_id="b", fixture_id="f", variant="lex", success=True) + with pytest.raises(ValueError, match="mixed variants"): + aggregate_runs([r1, r2]) + + def test_stale_detections_counted(self) -> None: + recs = [ + VariantRunRecord(run_id=f"x-lie-r{i}", fixture_id="x", variant="lie", + success=False, stale_detection=(i % 2 == 0)) + for i in range(4) + ] + agg = aggregate_runs(recs) + assert agg.stale_detections == 2 + + def test_baseline_note_is_no_baseline(self) -> None: + recs = self._records([True, True]) + agg = aggregate_runs(recs) + assert agg.trajectory_delta_note == "no_baseline" + assert agg.vs_baseline_correctness_delta is None + + +# --------------------------------------------------------------------------- +# GE9 — aggregate_runs baseline delta +# --------------------------------------------------------------------------- + + +class TestGe9BaselineDelta: + def _baseline(self, success_rate: float, tokens: float) -> VariantAggregate: + return VariantAggregate( + variant="baseline", n=4, + success_rate=success_rate, mean_retries=0.0, + mean_total_tokens=tokens, mean_repeated_reads=0.0, + stale_detections=0, + ) + + def _records(self, variant: str, successes: list[bool], token_base: int = 4000) -> list[VariantRunRecord]: + return [ + VariantRunRecord( + run_id=make_run_id("f", variant, i), + fixture_id="f", + variant=variant, + success=s, + total_tokens=token_base, + ) + for i, s in enumerate(successes) + ] + + def test_correctness_delta_positive(self) -> None: + base = self._baseline(0.5, 4000.0) + recs = self._records("lex", [True, True, True, True]) + agg = aggregate_runs(recs, baseline_agg=base) + assert agg.vs_baseline_correctness_delta is not None + assert abs(agg.vs_baseline_correctness_delta - 0.5) < 1e-4 + + def test_correctness_delta_negative(self) -> None: + base = self._baseline(1.0, 4000.0) + recs = self._records("lie", [True, False, False, False]) + agg = aggregate_runs(recs, baseline_agg=base) + assert agg.vs_baseline_correctness_delta is not None + assert agg.vs_baseline_correctness_delta < 0 + + def test_token_tie_note(self) -> None: + base = self._baseline(0.8, 4000.0) + recs = self._records("inline", [True, True, True, True], token_base=4200) + agg = aggregate_runs(recs, baseline_agg=base) + # delta is 200 tokens -> tie band (< 500) + assert agg.trajectory_delta_note == "tie" + + def test_token_improvement_note(self) -> None: + base = self._baseline(0.8, 6500.0) + recs = self._records("min", [True, True, True, True], token_base=4000) + agg = aggregate_runs(recs, baseline_agg=base) + # delta is -2500 tokens -> improvement (saved tokens) + assert agg.trajectory_delta_note == "improvement" + + def test_token_regression_note(self) -> None: + base = self._baseline(0.8, 4000.0) + recs = self._records("inj", [True, True, True, True], token_base=7000) + agg = aggregate_runs(recs, baseline_agg=base) + # delta is +3000 tokens -> regression + assert agg.trajectory_delta_note == "regression" + + def test_no_baseline_passed_baseline_variant(self) -> None: + recs = self._records("baseline", [True, True]) + agg = aggregate_runs(recs, baseline_agg=None) + assert agg.trajectory_delta_note == "no_baseline" + assert agg.vs_baseline_correctness_delta is None diff --git a/tests/test_grace_eval_sweep.py b/tests/test_grace_eval_sweep.py new file mode 100644 index 00000000..a0aa0461 --- /dev/null +++ b/tests/test_grace_eval_sweep.py @@ -0,0 +1,271 @@ +"""Tests for grace_eval.sweep contract sweep / stale-anchor audit (#339). + +Covers: + GS1 — anchor extraction + GS2 — contradiction detection for known signal pairs + GS3 — no false positives on clean code + GS4 — lie-variant fixture detection + GS5 — sweep_variant_sources aggregates across multiple files + GS6 — SweepFinding severity and location format + GS7 — invalid variant rejected +""" + +from __future__ import annotations + +import pytest + +from mapify_cli.grace_eval.sweep import sweep_source, sweep_variant_sources +from mapify_cli.grace_eval.schema import VARIANT_NAMES + + +# --------------------------------------------------------------------------- +# GS1 — anchor extraction (indirect: checks findings are produced for anchors) +# --------------------------------------------------------------------------- + + +class TestGs1AnchorExtraction: + def test_contract_keyword_detected(self) -> None: + source = ( + "def get_value(x):\n" + " # CONTRACT: never returns none\n" + " return None\n" + ) + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_anchor_keyword_detected(self) -> None: + source = ( + "def check(x):\n" + " # ANCHOR: never returns none\n" + " return None\n" + ) + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_case_insensitive_keyword(self) -> None: + source = ( + "# contract: never returns none\n" + "return None\n" + ) + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_no_anchors_no_findings(self) -> None: + source = "def f(x):\n return x + 1\n" + findings = sweep_source(source, variant="baseline") + assert findings == [] + + def test_multiple_anchors_in_file(self) -> None: + source = ( + "# CONTRACT: never returns none\n" + "return None\n" + "# CONTRACT: never returns none\n" + "return None\n" + ) + findings = sweep_source(source, variant="lie") + assert len(findings) == 2 + + +# --------------------------------------------------------------------------- +# GS2 — contradiction detection for known signal pairs +# --------------------------------------------------------------------------- + + +class TestGs2ContradictionDetection: + def test_returns_none_contradicts_never_returns_none(self) -> None: + source = "# CONTRACT: never returns none\nreturn None\n" + findings = sweep_source(source, variant="lie") + assert any("never returns none" in f.detail.lower() for f in findings) + + def test_returns_true_contradicts_returns_false(self) -> None: + source = "# CONTRACT: returns false\nreturn True\n" + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_returns_false_contradicts_returns_true(self) -> None: + source = "# CONTRACT: returns true\nreturn False\n" + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_list_append_contradicts_idempotent(self) -> None: + source = "# CONTRACT: idempotent\nresults.append(x)\n" + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_self_assignment_contradicts_no_side_effect(self) -> None: + source = "# CONTRACT: no side effect\nself.count = 0\n" + findings = sweep_source(source, variant="lie") + assert len(findings) == 1 + + def test_contradiction_beyond_lookahead_not_flagged(self) -> None: + # Contradiction is placed 20 lines after the anchor, outside the 8-line window. + filler = " pass\n" * 15 + source = f"# CONTRACT: never returns none\n{filler}return None\n" + findings = sweep_source(source, variant="lie") + assert len(findings) == 0 + + +# --------------------------------------------------------------------------- +# GS3 — no false positives on clean code +# --------------------------------------------------------------------------- + + +class TestGs3NoFalsePositives: + def test_correct_never_returns_none(self) -> None: + source = ( + "# CONTRACT: never returns none\n" + "def f(x):\n" + " if x > 0:\n" + " return x\n" + " return 0\n" + ) + findings = sweep_source(source, variant="inline") + assert findings == [] + + def test_correct_idempotent(self) -> None: + source = ( + "# CONTRACT: idempotent\n" + "def write(path, data):\n" + " path.write_text(data)\n" + ) + findings = sweep_source(source, variant="min") + assert findings == [] + + def test_clean_code_no_anchor(self) -> None: + source = "x = 1\ny = x + 2\n" + findings = sweep_source(source, variant="lex") + assert findings == [] + + +# --------------------------------------------------------------------------- +# GS4 — lie-variant fixture detection +# --------------------------------------------------------------------------- + + +class TestGs4LieVariant: + def _lie_source(self) -> str: + return ( + "def validate(value):\n" + " # CONTRACT: returns false when value is negative\n" + " if value < 0:\n" + " return True # lie: says False but code returns True\n" + ) + + def test_lie_variant_flagged(self) -> None: + findings = sweep_source(self._lie_source(), variant="lie") + assert len(findings) == 1 + assert findings[0].variant == "lie" + assert findings[0].severity == "critical" + + def test_same_source_inline_also_flagged(self) -> None: + # The sweep does not distinguish lie from other variants in detection logic; + # lie just represents a source known to contain stale contracts. + findings = sweep_source(self._lie_source(), variant="inline") + assert len(findings) == 1 + assert findings[0].variant == "inline" + + def test_location_prefix_included(self) -> None: + source = "# CONTRACT: returns false when value is negative\nreturn True\n" + findings = sweep_source(source, variant="lie", location_prefix="src/utils.py") + assert len(findings) == 1 + assert "src/utils.py" in findings[0].location + + def test_location_has_line_number(self) -> None: + source = "# CONTRACT: returns false when value is negative\nreturn True\n" + findings = sweep_source(source, variant="lie", location_prefix="f.py") + assert "L1" in findings[0].location + + +# --------------------------------------------------------------------------- +# GS5 — sweep_variant_sources aggregates across multiple files +# --------------------------------------------------------------------------- + + +class TestGs5SweepVariantSources: + def test_empty_sources(self) -> None: + findings = sweep_variant_sources({}, variant="lie") + assert findings == [] + + def test_single_clean_file(self) -> None: + findings = sweep_variant_sources({"src/a.py": "x = 1\n"}, variant="lex") + assert findings == [] + + def test_single_file_with_finding(self) -> None: + sources = { + "src/a.py": "# CONTRACT: never returns none\nreturn None\n", + } + findings = sweep_variant_sources(sources, variant="lie") + assert len(findings) == 1 + assert "src/a.py" in findings[0].location + + def test_multiple_files_findings_aggregated(self) -> None: + sources = { + "src/a.py": "# CONTRACT: never returns none\nreturn None\n", + "src/b.py": "# CONTRACT: returns false\nreturn True\n", + } + findings = sweep_variant_sources(sources, variant="lie") + assert len(findings) == 2 + locs = {f.location for f in findings} + assert any("src/a.py" in loc for loc in locs) + assert any("src/b.py" in loc for loc in locs) + + def test_non_string_values_skipped(self) -> None: + sources = { + "src/a.py": "# CONTRACT: never returns none\nreturn None\n", + "src/b.py": 42, # non-str, should be skipped + } + findings = sweep_variant_sources(sources, variant="lie") # type: ignore[arg-type] + assert len(findings) == 1 + + def test_clean_and_stale_mix(self) -> None: + sources = { + "src/clean.py": "def f(x):\n return x\n", + "src/stale.py": "# CONTRACT: never returns none\nreturn None\n", + } + findings = sweep_variant_sources(sources, variant="lie") + assert len(findings) == 1 + assert "src/stale.py" in findings[0].location + + +# --------------------------------------------------------------------------- +# GS6 — SweepFinding severity and location format +# --------------------------------------------------------------------------- + + +class TestGs6FindingFormat: + def test_finding_is_critical_for_contradiction(self) -> None: + source = "# CONTRACT: never returns none\nreturn None\n" + findings = sweep_source(source, variant="lie") + assert all(f.severity == "critical" for f in findings) + + def test_detail_is_non_empty(self) -> None: + source = "# CONTRACT: never returns none\nreturn None\n" + findings = sweep_source(source, variant="lie") + assert all(f.detail for f in findings) + + def test_variant_matches_input(self) -> None: + source = "# CONTRACT: never returns none\nreturn None\n" + for variant in ("lie", "inline", "lex"): + findings = sweep_source(source, variant=variant) + for f in findings: + assert f.variant == variant + + +# --------------------------------------------------------------------------- +# GS7 — invalid variant rejected +# --------------------------------------------------------------------------- + + +class TestGs7InvalidVariant: + def test_sweep_source_invalid_variant(self) -> None: + with pytest.raises(ValueError, match="variant"): + sweep_source("x = 1\n", variant="unknown") + + def test_sweep_variant_sources_invalid_variant(self) -> None: + with pytest.raises(ValueError, match="variant"): + sweep_variant_sources({"f.py": "x = 1\n"}, variant="bad") + + @pytest.mark.parametrize("variant", VARIANT_NAMES) + def test_all_valid_variants_accepted(self, variant: str) -> None: + findings = sweep_source("def f(): pass\n", variant=variant) + assert isinstance(findings, list)