Harden task validation and repair across the remaining FlyDSL tasks

agents/quality_loop/README.md

@@ -1,7 +1,7 @@
 # quality_loop agent
 
 `quality_loop` is a repository-level task curator. It audits every selected task,
-attempts one repair for blocking validator failures, runs exactly one Codex
+attempts bounded repairs for blocking validator failures, runs exactly one Codex
 optimization iteration, sends the result to an independent Codex reviewer, and
 optionally hardens an easy baseline or task cases behind fail-closed correctness
 gates. It records unrepairable task defects locally and bundles all accepted task
@@ -72,6 +72,32 @@ push and PR creation. `--plan` is the only intentionally offline mode.
 
 ## Task contract and budgets
 
+Use `quality_loop.mode: repair_only` to audit and repair task quality without
+optimizing candidates, promoting baselines or claiming a speedup. That mode has
+zero optimization iterations. Normal `optimize` mode retains exactly one.
+`max_repair_attempts` defaults to 1 and permits 1--3 attempts. Every attempt needs
+a changed, previously unseen proposal; unchanged validator reruns cannot erase
+an existing finding. All proposals and rejected evidence remain in the artifacts.
+
+WARN-only findings remain recorded by default. `repair_warnings` can explicitly
+select validator check names to repair, for example:
+
+```yaml
+quality_loop:
+  mode: repair_only
+  max_repair_attempts: 2
+  repair_warnings: [correctness_implementation_review, benchmark_integrity]
+```
+
+Repairs preserve the original TaskSpec, all existing case identities/checks and
+captured timing/sample/cache policy. Additional correctness-only cases are
+allowed; changing scored workloads requires a separate reviewed migration.
+An independent repair reviewer compares the source with the original package,
+checks reference/tolerance/input-generation semantics not captured by those
+deterministic checks, and must close every retained finding. A new validator PASS
+alone cannot authorize weakening a task. GPU validation and the read-only review
+both precede acceptance or publication of repaired files.
+
 Tasks use the common [v2 task contract](../../docs/how-to/add-task.md).
 Discovery preserves the full path below `tasks/` as the stable ID, including when
 copying a task to a scratch directory. Nested dependency configs are not separate
@@ -133,8 +159,11 @@ drain timeout is identified separately from the configured role deadline.
 1. Run the shared task validator in a fresh workspace. Require the framework's
    completion marker, matching report hash, task ID, fresh timestamp and successful
    framework status. Missing or operationally failed reports stop the attempt.
-2. Record WARN findings without repairing them. For task FAIL, allow one repair
-   and re-run the validator in a new workspace; a repair needs clean PASS.
+2. Record WARN findings; repair only explicitly selected warning checks. For task
+   FAIL, attempt the configured bounded repairs with original-policy and finding
+   checks, then re-run the validator in a new workspace. A repair needs clean
+   PASS and an accepting independent repair review. In `repair_only` mode stop
+   here and publish only accepted task repairs through the same host finalizer.
 3. Use the shared `TaskSession` to validate the initial package and freeze its
    baseline before the optimizer edits anything. Unimplemented candidates remain
    legal only during initial validation. The declared baseline correctness policy

agents/quality_loop/agent_config.yaml

@@ -21,8 +21,10 @@ quality_loop:
     effort: medium
     timeout_seconds: 3600
 
-  max_repair_attempts: 1
-  optimization_iterations: 1  # enforced; any other value is rejected
+  mode: optimize             # repair_only skips optimization/promotion
+  repair_warnings: []        # opt in by validator check name
+  max_repair_attempts: 1      # bounded 1..3; each attempt needs changed source
+  optimization_iterations: 1 # 1 in optimize mode; 0 in repair_only mode
   easy_speedup_threshold: 5.0
   easy_confirmation_runs: 3
   case_enhancement: true

agents/quality_loop/config.py

@@ -74,6 +74,8 @@ class QualityLoopConfig:
     validator: BackendConfig = field(default_factory=BackendConfig)
     github: GitHubConfig = field(default_factory=GitHubConfig)
     max_repair_attempts: int = 1
+    mode: str = "optimize"
+    repair_warnings: tuple[str, ...] = ()
     optimization_iterations: int = 1
     easy_speedup_threshold: float = 5.0
     easy_confirmation_runs: int = 3
@@ -96,15 +98,25 @@ def from_dict(cls, raw: dict[str, Any]) -> "QualityLoopConfig":
         audit = raw.get("quality_loop", {}) or {}
         if not isinstance(audit, dict):
             raise ValueError("quality_loop must be a mapping")
-        iterations = int(audit.get("optimization_iterations", 1))
-        if iterations != 1:
+        mode = audit.get("mode", "optimize")
+        if mode not in ("optimize", "repair_only"):
+            raise ValueError("quality_loop.mode must be optimize or repair_only")
+        expected_iterations = 0 if mode == "repair_only" else 1
+        iterations = audit.get("optimization_iterations", expected_iterations)
+        if type(iterations) is not int or iterations != expected_iterations:
             raise ValueError(
-                "quality_loop enforces exactly one optimization iteration; "
-                "optimization_iterations must be 1"
+                "quality_loop enforces exactly one optimization iteration in optimize mode "
+                "and zero in repair_only mode"
             )
-        repair_attempts = int(audit.get("max_repair_attempts", 1))
-        if repair_attempts != 1:
-            raise ValueError("quality_loop requires exactly one repair attempt")
+        repair_attempts = audit.get("max_repair_attempts", 1)
+        if type(repair_attempts) is not int or not 1 <= repair_attempts <= 3:
+            raise ValueError("quality_loop.max_repair_attempts must be an integer from 1 to 3")
+        from agents.task_validator.report_schema import CHECK_NAMES
+        repair_warnings = audit.get("repair_warnings", [])
+        if (not isinstance(repair_warnings, list)
+                or any(not isinstance(name, str) or name not in CHECK_NAMES for name in repair_warnings)
+                or len(set(repair_warnings)) != len(repair_warnings)):
+            raise ValueError("quality_loop.repair_warnings must list distinct validator check names")
         confirmations = int(audit.get("easy_confirmation_runs", 3))
         if confirmations < 1:
             raise ValueError("easy_confirmation_runs must be at least 1")
@@ -128,6 +140,8 @@ def from_dict(cls, raw: dict[str, Any]) -> "QualityLoopConfig":
             validator=BackendConfig.from_dict(audit.get("validator")),
             github=GitHubConfig.from_dict(audit.get("github")),
             max_repair_attempts=repair_attempts,
+            mode=mode,
+            repair_warnings=tuple(repair_warnings),
             optimization_iterations=iterations,
             easy_speedup_threshold=threshold,
             easy_confirmation_runs=confirmations,

agents/quality_loop/filesystem.py

@@ -22,6 +22,7 @@
     "baseline_perf.yaml", "optimized_perf.yaml", "quality_loop_review.yaml",
     "performance_report.json", "perf_report.json", "compile_report.json",
     ".validation_complete", "correctness_report.json", "benchmark_results.json",
+    "quality_loop_repair_review.yaml",
     "eval_result.yaml", AKA_HELPER_FILE_NAME, NATIVE_HIP_MATERIALIZED.name,
 }
 GENERATED_SUFFIXES = {

agents/quality_loop/orchestrator.py

@@ -31,6 +31,7 @@
     case_enhancement_prompt,
     optimizer_prompt,
     repair_prompt,
+    repair_review_prompt,
     reviewer_prompt,
 )
 from agents.quality_loop.state import (
@@ -40,6 +41,7 @@
     validate_run_id,
 )
 from agents.quality_loop.runtime import InitialValidationRejected, create_session
+from agents.quality_loop.repair_policy import RepairPolicy, RepairRejected, blocking_findings
 from agents.task_validator.launch_agent import launch_agent as launch_validator
 from agents.task_validator.report_schema import (
     COMPLETION_MARKER_FILENAME, validation_report_is_complete,
@@ -172,6 +174,7 @@ def __init__(
         self._sessions: dict[Path, Any] = {}
         self._session_task_ids: dict[Path, str] = {}
         self._validation_evidence: list[dict[str, Any]] = []
+        self._validation_manifests: dict[Path, tuple[dict, ...]] = {}
         self.state: AuditState | None = None
         self.artifact_dir: Path | None = None
         self.worktree: Path | None = None
@@ -367,31 +370,43 @@ def _process_task(self, task_id: str, canonical_task: Path) -> None:
         original_validation_status = str(validation.get("overall_status", "FAIL")).upper()
         warnings = _validation_warnings(validation)
 
-        if original_validation_status == "FAIL":
+        repair_review = None
+        if original_validation_status == "FAIL" or blocking_findings(validation, self.config.repair_warnings):
             self.state.transition(task_id, "repairing", warnings=warnings)
-            spec = load_task_spec(candidate_task / "config.yaml", task_id=task_id)
-            before = snapshot_tree(validation_workspace)
-            self.backend.run(
-                repair_prompt(validation, task_id), validation_workspace, role="repair"
-            )
-            after = snapshot_tree(validation_workspace)
-            changes = _filtered_changes(
-                before, after, materialized=_materialized_destinations(spec),
-                source_root=validation_workspace
-            )
-            if changes.empty:
+            try:
+                repaired = self._repair_task(
+                    task_id, original_task, candidate_task, task_artifacts,
+                    validation_workspace, validation,
+                )
+            except RepairRejected as exc:
+                self.state.transition(task_id, "repair_rejected", repair_rejection=str(exc))
                 self._handle_unrepairable(task_id, validation)
                 return
-            apply_changes(validation_workspace, candidate_task, changes)
-            load_task_spec(candidate_task / "config.yaml", task_id=task_id)
-            restore_committed_perf_stubs(candidate_task)
-            _, validation = self._validate(
-                task_id, candidate_task, task_artifacts / "validation_repaired"
-            )
-            warnings = _validation_warnings(validation)
-            if validation.get("overall_status") != "PASS":
-                self._handle_unrepairable(task_id, validation)
+            if repaired is None:
                 return
+            validation, repair_review = repaired
+            warnings = _validation_warnings(validation)
+
+        if self.config.mode == "repair_only":
+            changes = _filtered_changes(original_tree, snapshot_tree(candidate_task),
+                                        source_root=candidate_task)
+            if not changes.empty:
+                if validation.get("overall_status") != "PASS" or repair_review is None:
+                    self._handle_unrepairable(task_id, validation)
+                    return
+                apply_changes(candidate_task, canonical_task, changes)
+                restore_committed_perf_stubs(canonical_task)
+            commit = (self.publisher.commit_task(self.worktree, task_id)
+                      if not changes.empty and not self.defer_github else None)
+            self.state.transition(
+                task_id, "completed", mode="repair_only", warnings=warnings,
+                changes=list(changes.paths), commit=commit,
+                commit_pending=not changes.empty and self.defer_github,
+                speedups=[], reviewer=repair_review, baseline_hardened=False,
+                cases_enhanced=False, validation_evidence=list(self._validation_evidence),
+                accepted_fingerprint=stable_fingerprint(snapshot_tree(canonical_task)),
+            )
+            return
 
         self.state.transition(task_id, "optimizing", warnings=warnings)
         optimization_workspace, baseline_cases, result = self._optimize_once(
@@ -519,6 +534,7 @@ def launch(*, eval_config, task_config_dir, workspace):
             return self.validator_launcher(eval_config, task_config_dir, workspace)
         validate_task_session(session, eval_config=settings, task_config_dir=str(package / "config.yaml"),
                               agent_launcher=launch)
+        self._validation_manifests[workspace] = tuple(session.manifest.cases) if session.manifest else ()
         if not validation_report_is_complete(workspace):
             raise RuntimeError(f"validator did not finalize a complete report for {task_id}")
         if report_path.is_symlink() or marker_path.is_symlink():
@@ -548,6 +564,79 @@ def launch(*, eval_config, task_config_dir, workspace):
         })
         return workspace, report
 
+    def _repair_task(self, task_id, original_task, candidate_task, artifacts, workspace, report):
+        spec = load_task_spec(candidate_task / "config.yaml", task_id=task_id)
+        policy = RepairPolicy.capture(spec, self._validation_manifests.get(workspace), report)
+        retained = blocking_findings(report, self.config.repair_warnings)
+        attempted = {stable_fingerprint(snapshot_tree(candidate_task))}
+        for attempt in range(1, self.config.max_repair_attempts + 1):
+            stage = artifacts / f"repair_{attempt:02d}"
+            editor = self._make_workspace(task_id, candidate_task, stage / "editor")
+            before = snapshot_tree(editor)
+            self.backend.run(repair_prompt(report, task_id, findings=retained,
+                                           warning_checks=self.config.repair_warnings),
+                             editor, role="repair")
+            changes = _filtered_changes(before, snapshot_tree(editor),
+                                        materialized=_materialized_destinations(spec), source_root=editor)
+            if changes.empty:
+                break
+            proposal = stage / "proposal"
+            self._copy_task(candidate_task, proposal)
+            apply_changes(editor, proposal, changes)
+            restore_committed_perf_stubs(proposal)
+            try:
+                proposed_spec = load_task_spec(proposal / "config.yaml", task_id=task_id)
+            except (ValueError, OSError) as exc:
+                raise RepairRejected(f"Repair proposed an invalid task specification: {exc}") from exc
+            policy.verify_specification(proposed_spec)
+            fingerprint = stable_fingerprint(snapshot_tree(proposal))
+            if fingerprint in attempted:
+                self.logger.warning("Stopping repeated repair proposal for %s", task_id)
+                break
+            attempted.add(fingerprint)
+            validated, next_report = self._validate(task_id, proposal, stage / "validation")
+            policy.verify(proposed_spec, self._validation_manifests.get(validated), next_report)
+            retained.update(blocking_findings(next_report, self.config.repair_warnings))
+            report = next_report
+            if report.get("overall_status") == "PASS":
+                decision = self._review_repair(task_id, original_task, validated, retained)
+                if decision.get("accepted") is True:
+                    self._replace_directory(candidate_task, proposal)
+                    return report, decision
+                self.state.transition(task_id, "repair_review_rejected", repair_review=decision)
+                break
+            # Keep each proposal/evidence tree; the next editor starts from this
+            # changed package, never from an unchanged validator rerun.
+            self._replace_directory(candidate_task, proposal)
+        self._handle_unrepairable(task_id, report)
+        return None
+
+    def _review_repair(self, task_id, original, workspace, findings):
+        output_name = "quality_loop_repair_review.yaml"
+        output = workspace / output_name
+        if output.exists() or output.is_symlink():
+            raise RuntimeError("Repair review requires a fresh decision")
+        before, original_before = snapshot_tree(workspace), snapshot_tree(original)
+        self.reviewer_backend.run(repair_review_prompt(task_id, original, findings, output_name),
+                                  workspace, role="repair_reviewer")
+        changes = diff_trees(before, snapshot_tree(workspace))
+        if (snapshot_tree(original) != original_before
+                or set(changes.paths) != {output_name} or changes.deleted
+                or not output.is_file() or output.is_symlink()):
+            raise RuntimeError("Repair reviewer changed protected files or omitted its decision")
+        result = yaml.safe_load(output.read_text())
+        if (not isinstance(result, dict) or not isinstance(result.get("summary"), str)
+                or any(type(result.get(k)) is not bool for k in
+                       ("accepted", "contract_preserved", "scoring_preserved"))
+                or not isinstance(result.get("resolved_findings"), list)
+                or any(not isinstance(k, str) for k in result["resolved_findings"])
+                or len(set(result["resolved_findings"])) != len(result["resolved_findings"])):
+            raise RuntimeError("Invalid independent repair review")
+        if (result["contract_preserved"] is not True or result["scoring_preserved"] is not True
+                or set(result["resolved_findings"]) != set(findings)):
+            result["accepted"] = False
+        return result
+
     def _new_session(self, task_id: str, workspace: Path, spec: TaskSpec, stage_dir: Path):
         session = self.session_factory(
             spec=spec, workspace=workspace, artifact_dir=stage_dir,