From 31287f14494f2005c942327be71c721a1a299de6 Mon Sep 17 00:00:00 2001 From: sladro Date: Thu, 2 Apr 2026 12:32:32 +0800 Subject: [PATCH] feat: add bounded mutation validation --- engine/mutation_engine.py | 53 +++++++++++++++++++++ scripts/run_task.py | 14 ++++++ tests/test_mutation_engine.py | 86 +++++++++++++++++++++++++++++++++++ 3 files changed, 153 insertions(+) create mode 100644 engine/mutation_engine.py create mode 100644 tests/test_mutation_engine.py diff --git a/engine/mutation_engine.py b/engine/mutation_engine.py new file mode 100644 index 0000000..d0d0c8e --- /dev/null +++ b/engine/mutation_engine.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from difflib import unified_diff +from pathlib import Path + +from engine.models import BaselineSnapshot, TaskSpec + + +class MutationValidationError(ValueError): + pass + + +def _count_changed_lines(before: str, after: str, path: Path) -> int: + diff = unified_diff( + before.splitlines(keepends=True), + after.splitlines(keepends=True), + fromfile=f"{path.as_posix()} (before)", + tofile=f"{path.as_posix()} (after)", + ) + changed_lines = 0 + for line in diff: + if line.startswith(("---", "+++", "@@")): + continue + if line.startswith(("+", "-")): + changed_lines += 1 + return changed_lines + + +def validate_candidate_changes(task: TaskSpec, snapshot: BaselineSnapshot) -> None: + changed_files = 0 + changed_lines = 0 + allowed_file_types = set(task.mutation.allowed_file_types) + + for path, baseline_text in snapshot.file_contents.items(): + current_text = path.read_text(encoding="utf-8") if path.exists() else "" + if current_text == baseline_text: + continue + + changed_files += 1 + if path.suffix not in allowed_file_types: + raise MutationValidationError(f"disallowed file type: {path.suffix}") + + changed_lines += _count_changed_lines(baseline_text, current_text, path) + + if changed_files > task.artifacts.max_files_per_iteration: + raise MutationValidationError( + f"too many changed files: {changed_files} > {task.artifacts.max_files_per_iteration}" + ) + + if changed_lines > task.mutation.max_changed_lines: + raise MutationValidationError( + f"too many changed lines: {changed_lines} > {task.mutation.max_changed_lines}" + ) diff --git a/scripts/run_task.py b/scripts/run_task.py index 638715b..0155211 100644 --- a/scripts/run_task.py +++ b/scripts/run_task.py @@ -11,6 +11,7 @@ if str(ROOT_DIR) not in sys.path: from engine.artifact_manager import ArtifactManager from engine.decision_engine import decide_candidate +from engine.mutation_engine import MutationValidationError, validate_candidate_changes from engine.runner import run_command from engine.scorer import parse_score_output from engine.task_loader import load_task @@ -58,6 +59,19 @@ def main() -> int: artifact_manager = ArtifactManager(task) snapshot = artifact_manager.snapshot() + try: + validate_candidate_changes(task, snapshot) + except MutationValidationError as exc: + return _emit_record( + repo_root=repo_root, + task_id=task.id, + results_file=task.logging.results_file, + status="discard", + reason=str(exc), + candidate_score=None, + diff_summary=artifact_manager.diff_summary(snapshot), + ) + run_result = run_command( task.runner.command, _resolve_repo_path(repo_root, task.runner.cwd), diff --git a/tests/test_mutation_engine.py b/tests/test_mutation_engine.py new file mode 100644 index 0000000..2700b23 --- /dev/null +++ b/tests/test_mutation_engine.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from engine.artifact_manager import ArtifactManager +from engine.models import ( + ArtifactSpec, + BudgetSpec, + ConstraintSpec, + LoggingSpec, + MutationSpec, + ObjectiveSpec, + PolicySpec, + RunnerSpec, + ScorerParseSpec, + ScorerSpec, + TaskSpec, +) +from engine.mutation_engine import MutationValidationError, validate_candidate_changes + + +def _make_task(root_dir: Path, allowed_file_types: list[str], max_changed_lines: int) -> TaskSpec: + return TaskSpec( + id="mutation-test", + description="Mutation validation fixture.", + artifacts=ArtifactSpec(include=["fixtures/*"], exclude=[], max_files_per_iteration=10), + mutation=MutationSpec( + mode="direct_edit", + allowed_file_types=allowed_file_types, + max_changed_lines=max_changed_lines, + ), + runner=RunnerSpec(command="python -c \"print('runner ok')\"", cwd=".", timeout_seconds=30), + scorer=ScorerSpec( + type="command", + command="python -c \"print('{\\\"score\\\": 1.0, \\\"metrics\\\": {}}')\"", + parse=ScorerParseSpec(format="json", score_field="score", metrics_field="metrics"), + ), + objective=ObjectiveSpec(primary_metric="score", direction="maximize"), + constraints=[], + policy=PolicySpec(keep_if="better_primary", tie_breakers=[], on_failure="discard"), + budget=BudgetSpec(max_iterations=1, max_failures=1), + logging=LoggingSpec(results_file="work/results.jsonl", candidate_dir="work/candidates"), + root_dir=root_dir, + ) + + +class MutationEngineTest(unittest.TestCase): + def test_rejects_too_many_changed_lines(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root_dir = Path(tmp) + fixture_dir = root_dir / "fixtures" + fixture_dir.mkdir(parents=True) + target = fixture_dir / "note.md" + target.write_text("line 1\nline 2\n", encoding="utf-8") + + task = _make_task(root_dir, allowed_file_types=[".md"], max_changed_lines=1) + snapshot = ArtifactManager(task).snapshot() + target.write_text("line 1\nline 2\nline 3\n", encoding="utf-8") + + with self.assertRaises(MutationValidationError) as ctx: + validate_candidate_changes(task, snapshot) + + self.assertIn("changed lines", str(ctx.exception)) + + def test_rejects_disallowed_extension(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root_dir = Path(tmp) + fixture_dir = root_dir / "fixtures" + fixture_dir.mkdir(parents=True) + target = fixture_dir / "note.txt" + target.write_text("line 1\n", encoding="utf-8") + + task = _make_task(root_dir, allowed_file_types=[".md"], max_changed_lines=10) + snapshot = ArtifactManager(task).snapshot() + target.write_text("line 1 changed\n", encoding="utf-8") + + with self.assertRaises(MutationValidationError) as ctx: + validate_candidate_changes(task, snapshot) + + self.assertIn("disallowed file type", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()