feat: add bounded mutation validation

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}"
+        )

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),

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()