Enforce PaperFit repair candidate approval gates

Author: xxllovemkm

scripts/orchestrator_runtime.py

@@ -12,6 +12,8 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from runtime_repair_risk import annotate_repair_candidates
+
 
 def _load_json(path: Optional[str]) -> Dict[str, Any]:
     if not path:
@@ -982,6 +984,7 @@ def generate_repair_plan(
         + visual_space_candidates
         + log_candidates
     )
+    candidates = annotate_repair_candidates(candidates)
     candidates.sort(
         key=lambda item: (
             -int(item.get("priority_score") or 0),

scripts/repair_plan_generator.py

@@ -5,6 +5,11 @@
 
 from typing import Any, Dict
 
+try:
+    from runtime_repair_risk import classify_repair_candidate_risk
+except ModuleNotFoundError:  # package import during unit tests
+    from .runtime_repair_risk import classify_repair_candidate_risk
+
 
 SOURCE_CHANGING_TASK_TYPES = {
     "full_vto",
@@ -146,6 +151,11 @@ def build_approval_object(
         requires_approval = False
         approval_granted = True
         reason = str(action.get("status"))
+    elif action_skipped and action_reason == "approval_scope_blocked":
+        status = "approval_required"
+        requires_approval = True
+        approval_granted = False
+        reason = "approval_scope_blocked"
     elif action_skipped:
         status = "not_required"
         requires_approval = False
@@ -256,3 +266,77 @@ def build_approval_scope_carry_forward_check(
         "high_risk_operations": high_risk_operations,
         "checks": checks,
     }
+
+
+def build_candidate_approval_scope_gate(
+    *,
+    task: Dict[str, Any],
+    approval: Dict[str, Any],
+    repair_plan: Dict[str, Any],
+    candidate_limit: int = 1,
+) -> Dict[str, Any]:
+    """Evaluate whether selected repair candidates fit the approval scope."""
+
+    task_type = _task_type(task)
+    if task_type not in SOURCE_CHANGING_TASK_TYPES:
+        return {
+            "schema_version": "1.0",
+            "status": "not_applicable",
+            "reason": "task_does_not_change_source",
+            "selected_candidate_count": 0,
+            "candidate_risks": [],
+            "blocked_candidates": [],
+        }
+
+    policy = approval.get("policy") if isinstance(approval.get("policy"), dict) else {}
+    allowed_surface = set(str(item) for item in (policy.get("mutation_surface") or []))
+    high_risk_operations = set(str(item) for item in (policy.get("high_risk_operations") or []))
+    candidates = [
+        item
+        for item in (repair_plan.get("candidates") or [])[: max(0, int(candidate_limit or 0))]
+        if isinstance(item, dict)
+    ]
+    candidate_risks = []
+    blocked_candidates = []
+    for index, candidate in enumerate(candidates, start=1):
+        risk = candidate.get("risk") if isinstance(candidate.get("risk"), dict) else classify_repair_candidate_risk(candidate)
+        surfaces = set(str(item) for item in (risk.get("mutation_surface") or []))
+        operation = str(risk.get("operation") or "")
+        surface_within_scope = bool(surfaces) and surfaces.issubset(allowed_surface)
+        high_risk_operation = operation in high_risk_operations or str(risk.get("risk_level") or "") == "high"
+        allowed = surface_within_scope and not high_risk_operation
+        entry = {
+            "index": index,
+            "defect_family": candidate.get("defect_family"),
+            "candidate_type": candidate.get("candidate_type"),
+            "proposed_action": candidate.get("proposed_action"),
+            "target": candidate.get("target"),
+            "risk": risk,
+            "checks": {
+                "mutation_surface_within_scope": surface_within_scope,
+                "high_risk_operation_requires_fresh_approval": high_risk_operation,
+            },
+            "allowed_under_current_scope": allowed,
+        }
+        candidate_risks.append(entry)
+        if not allowed:
+            blocked_candidates.append(entry)
+
+    status = "pass" if candidates and not blocked_candidates else "blocked" if candidates else "not_evaluated"
+    reason = (
+        "selected_candidates_within_approval_scope"
+        if status == "pass"
+        else "selected_candidate_exceeds_approval_scope"
+        if blocked_candidates
+        else "no_selected_candidates"
+    )
+    return {
+        "schema_version": "1.0",
+        "status": status,
+        "reason": reason,
+        "approval_scope": policy.get("approval_scope"),
+        "candidate_limit": candidate_limit,
+        "selected_candidate_count": len(candidates),
+        "candidate_risks": candidate_risks,
+        "blocked_candidates": blocked_candidates,
+    }

scripts/runtime_approval.py

@@ -88,6 +88,9 @@ def build_repair_loop_policy(
     repair_action = (runtime_actions.get("repair_plan_executor") or {}) if isinstance(runtime_actions, dict) else {}
     freshness = (artifact_manifest.get("freshness") or {}) if isinstance(artifact_manifest, dict) else {}
     approval_policy = approval.get("policy") or {}
+    approval_scope_gate = repair_action.get("approval_scope_gate") if isinstance(repair_action, dict) else None
+    if not isinstance(approval_scope_gate, dict):
+        approval_scope_gate = None
 
     max_rounds = max(1, _as_int(task.get("max_rounds"), 1))
     current_round = max(1, _as_int(state.get("current_round"), 1))
@@ -96,7 +99,9 @@ def build_repair_loop_policy(
     dry_run = bool(task.get("dry_run_source_mutation")) or repair_action.get("reason") == "dry_run_source_mutation"
 
     stop_condition = "continue"
-    if approval.get("status") == "approval_required":
+    if approval_scope_gate and approval_scope_gate.get("status") != "pass":
+        stop_condition = "approval_scope_blocked"
+    elif approval.get("status") == "approval_required":
         stop_condition = "approval_required"
     elif str(status or "").lower() == "done" or str(gatekeeper_decision or "").upper() == "DONE":
         stop_condition = "done"
@@ -110,6 +115,8 @@ def build_repair_loop_policy(
     next_round_reason = "multi_round_apply_not_enabled_in_current_runtime"
     if dry_run:
         next_round_reason = "dry_run_source_mutation"
+    elif stop_condition == "approval_scope_blocked":
+        next_round_reason = "approval_scope_blocked"
     elif approval.get("status") == "approval_required":
         next_round_reason = "approval_required"
     elif stop_condition == "done":
@@ -132,6 +139,7 @@ def build_repair_loop_policy(
         "artifact_freshness_pass": freshness.get("status") == "pass",
         "mutation_integrity_available": bool(content_integrity.get("validation_status")),
         "source_mutation_executed": applied_count > 0,
+        "candidate_approval_scope_gate_pass": approval_scope_gate is None or approval_scope_gate.get("status") == "pass",
         "within_round_limit": current_round < max_rounds,
         "runtime_execution_mode_can_auto_apply": False,
     }
@@ -159,6 +167,7 @@ def build_repair_loop_policy(
         "next_round_allowed": False,
         "next_round_reason": next_round_reason,
         "approval_scope_carry_forward": approval_scope_carry_forward,
+        "candidate_approval_scope_gate": approval_scope_gate,
         "round_artifact_lineage": round_lineage,
         "second_round_apply_readiness": {
             "schema_version": "1.0",

scripts/runtime_repair_loop.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Candidate-level risk classification for source-changing repair plans."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable, List
+
+
+def _as_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def classify_repair_candidate_risk(candidate: Dict[str, Any]) -> Dict[str, Any]:
+    """Classify the mutation risk for one repair-plan candidate."""
+
+    defect_family = str(candidate.get("defect_family") or "")
+    candidate_type = str(candidate.get("candidate_type") or "")
+    proposed_action = str(candidate.get("proposed_action") or "")
+    target = candidate.get("target") if isinstance(candidate.get("target"), dict) else {}
+    target_kind = str(target.get("object_kind") or "")
+    target_float_type = str(target.get("float_type") or "")
+    section_distance = _as_int(candidate.get("section_distance"), 0)
+
+    operation = "layout_macros"
+    mutation_surface: List[str] = ["layout_macros"]
+    risk_level = "medium"
+    reason = "layout_macro_adjustment"
+
+    if defect_family.startswith("A") or defect_family.startswith("C"):
+        operation = "spacing"
+        mutation_surface = ["spacing"]
+        risk_level = "low"
+        reason = "spacing_or_consistency_adjustment"
+    elif defect_family == "B1":
+        operation = "float_placement"
+        mutation_surface = ["float_placement"]
+        risk_level = "medium"
+        reason = "float_placement_near_reference"
+        if section_distance >= 1:
+            operation = "float_movement_across_section_boundary"
+            risk_level = "high"
+            reason = "float_movement_may_cross_section_boundary"
+    elif defect_family in {"B2", "B3"}:
+        operation = "float_placement" if defect_family == "B3" else "layout_macros"
+        mutation_surface = ["float_placement"] if defect_family == "B3" else ["layout_macros"]
+        risk_level = "medium"
+        reason = "float_layout_adjustment"
+        if target_kind == "table_like" or target_float_type == "table" or candidate.get("source_table_env"):
+            operation = "table_reconstruction"
+            mutation_surface = ["table_environment", "table_placement"]
+            risk_level = "high"
+            reason = "table_layout_reconstruction_risk"
+    elif defect_family.startswith("D"):
+        operation = "layout_macros"
+        mutation_surface = ["layout_macros"]
+        risk_level = "medium"
+        reason = "overflow_layout_repair"
+
+    if candidate_type == "global" and proposed_action == "review_paragraph_spacing_and_looseness":
+        operation = "semantic_text_edit"
+        mutation_surface = ["text_spans"]
+        risk_level = "high"
+        reason = "semantic_text_edit_requires_fresh_approval"
+    if proposed_action == "template_migration":
+        operation = "template_migration"
+        mutation_surface = ["documentclass", "preamble", "template_macros"]
+        risk_level = "high"
+        reason = "template_migration_requires_fresh_approval"
+
+    return {
+        "schema_version": "1.0",
+        "risk_level": risk_level,
+        "operation": operation,
+        "mutation_surface": mutation_surface,
+        "requires_fresh_approval": risk_level == "high",
+        "reason": reason,
+    }
+
+
+def annotate_repair_candidates(candidates: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Return candidates with a stable risk object attached."""
+
+    annotated: List[Dict[str, Any]] = []
+    for candidate in candidates:
+        item = dict(candidate)
+        if not isinstance(item.get("risk"), dict):
+            item["risk"] = classify_repair_candidate_risk(item)
+        annotated.append(item)
+    return annotated

scripts/runtime_repair_risk.py

@@ -2,7 +2,12 @@
 
 import unittest
 
-from scripts.runtime_approval import build_approval_object, build_approval_scope_carry_forward_check
+from scripts.runtime_approval import (
+    build_approval_object,
+    build_approval_scope_carry_forward_check,
+    build_candidate_approval_scope_gate,
+)
+from scripts.runtime_repair_risk import classify_repair_candidate_risk
 from scripts.runtime_state_machine import (
     IllegalTransitionError,
     SOURCE_CHANGING_STATE_MACHINE,
@@ -183,6 +188,119 @@ def test_approval_scope_carry_forward_reports_contract_match(self) -> None:
         self.assertTrue(check["checks"]["approval_scope_matches"])
         self.assertTrue(check["checks"]["fresh_approval_required_for_high_risk_operations"])
 
+    def test_repair_candidate_risk_classifies_same_section_float_as_medium(self) -> None:
+        risk = classify_repair_candidate_risk(
+            {
+                "candidate_type": "source_anchor",
+                "defect_family": "B1",
+                "proposed_action": "move_float_closer_to_first_reference",
+                "section_distance": 0,
+                "target": {"float_type": "figure"},
+            }
+        )
+
+        self.assertEqual(risk["risk_level"], "medium")
+        self.assertEqual(risk["operation"], "float_placement")
+        self.assertFalse(risk["requires_fresh_approval"])
+
+    def test_repair_candidate_risk_classifies_cross_section_float_as_high(self) -> None:
+        risk = classify_repair_candidate_risk(
+            {
+                "candidate_type": "source_anchor",
+                "defect_family": "B1",
+                "proposed_action": "move_float_closer_to_first_reference",
+                "section_distance": 2,
+                "target": {"float_type": "figure"},
+            }
+        )
+
+        self.assertEqual(risk["risk_level"], "high")
+        self.assertEqual(risk["operation"], "float_movement_across_section_boundary")
+        self.assertTrue(risk["requires_fresh_approval"])
+
+    def test_repair_candidate_risk_classifies_table_reconstruction_as_high(self) -> None:
+        risk = classify_repair_candidate_risk(
+            {
+                "candidate_type": "source_anchor",
+                "defect_family": "B2",
+                "proposed_action": "adjust_float_width",
+                "target": {"object_kind": "table_like", "float_type": "table"},
+            }
+        )
+
+        self.assertEqual(risk["risk_level"], "high")
+        self.assertEqual(risk["operation"], "table_reconstruction")
+        self.assertTrue(risk["requires_fresh_approval"])
+
+    def test_repair_candidate_risk_classifies_global_paragraph_review_as_high(self) -> None:
+        risk = classify_repair_candidate_risk(
+            {
+                "candidate_type": "global",
+                "defect_family": "A1",
+                "proposed_action": "review_paragraph_spacing_and_looseness",
+            }
+        )
+
+        self.assertEqual(risk["risk_level"], "high")
+        self.assertEqual(risk["operation"], "semantic_text_edit")
+        self.assertTrue(risk["requires_fresh_approval"])
+
+    def test_candidate_approval_scope_gate_blocks_high_risk_selected_candidate(self) -> None:
+        approval = build_approval_object(
+            task={
+                "task_type": "full_vto",
+                "rollback_policy": "required",
+                "pre_repair_snapshot_required": True,
+            },
+            state={"repair_plan_summary": {"total_candidates": 1}},
+            runtime_actions={},
+        )
+        gate = build_candidate_approval_scope_gate(
+            task={"task_type": "full_vto"},
+            approval=approval,
+            repair_plan={
+                "candidates": [
+                    {
+                        "candidate_type": "source_anchor",
+                        "defect_family": "B1",
+                        "proposed_action": "move_float_closer_to_first_reference",
+                        "section_distance": 1,
+                    }
+                ]
+            },
+        )
+
+        self.assertEqual(gate["status"], "blocked")
+        self.assertEqual(gate["blocked_candidates"][0]["risk"]["operation"], "float_movement_across_section_boundary")
+
+    def test_candidate_approval_scope_gate_allows_medium_same_section_float_candidate(self) -> None:
+        approval = build_approval_object(
+            task={
+                "task_type": "full_vto",
+                "rollback_policy": "required",
+                "pre_repair_snapshot_required": True,
+            },
+            state={"repair_plan_summary": {"total_candidates": 1}},
+            runtime_actions={},
+        )
+        gate = build_candidate_approval_scope_gate(
+            task={"task_type": "full_vto"},
+            approval=approval,
+            repair_plan={
+                "candidates": [
+                    {
+                        "candidate_type": "source_anchor",
+                        "defect_family": "B1",
+                        "proposed_action": "move_float_closer_to_first_reference",
+                        "section_distance": 0,
+                    }
+                ]
+            },
+        )
+
+        self.assertEqual(gate["status"], "pass")
+        self.assertTrue(gate["candidate_risks"][0]["allowed_under_current_scope"])
+
 
 if __name__ == "__main__":
     unittest.main()