refactor(replan): localize refresh settlement qualification

Signed-off-by: huangruiteng <huangrt01@163.com>

Author: huangruiteng

loopx/control_plane/work_items/semantic_replan_writeback.py

@@ -4,6 +4,7 @@
 
 import shlex
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
 from ...agent_registry import registered_agent_ids_for_goal
@@ -22,17 +23,36 @@
     select_quota_todo_summary,
 )
 from ..todos.succession_warning import todo_succession_gap_items
+from .autonomous_replan_ack import (
+    latest_monitor_replan_frontier_identity,
+    watch_lane_continuation_todo_ids,
+)
 from .autonomous_replan_obligation import (
     build_autonomous_replan_cli_actions,
     project_todo_lifecycle_settlement_reentry,
 )
+from .delivery_outcome import ACCOUNTABLE_DELIVERY_OUTCOMES
 from .progress_observation import semantic_delta_from_writeback
+from .repair_delta import (
+    build_repair_delta_contract,
+    repair_delta_kinds_have_accountable_progress,
+)
 from .work_lane_context import build_work_lane_context_contract
 
 
 REPLAN_WRITEBACK_REJECTION_SCHEMA_VERSION = "replan_writeback_rejection_v0"
 
 
+@dataclass(frozen=True)
+class RefreshReplanQualification:
+    repair_delta_contract: dict[str, Any] | None
+    semantic_delta: dict[str, Any] | None
+    frontier_identity: str | None
+    classification: str
+    delivery_outcome: str | None
+    autonomous_replan_recorded: bool
+
+
 class ReplanWritebackRejected(ValueError):
     """A strict writeback rejection with its actionable obligation attached."""
 
@@ -344,3 +364,102 @@ def enforce_open_replan_writeback(
         obligation=obligation,
         semantic_delta=semantic_delta,
     )
+
+
+def qualify_refresh_replan_writeback(
+    *,
+    autonomous_replan_recorded: bool,
+    requested_delta_kinds: list[str],
+    active_state_next_action_update: dict[str, Any] | None,
+    agent_vision: dict[str, Any] | None,
+    existing_agent_vision: dict[str, Any] | None,
+    agent_id: str,
+    dry_run: bool,
+    settlement_todo_id: str | None,
+    settlement_bound: bool,
+    newest_first_runs: list[dict[str, Any]],
+    state_text: str,
+    goal_id: str,
+    progress_observation: dict[str, Any] | None,
+    registry_goal: dict[str, Any] | None,
+    completion_todo_id: str | None,
+    completion_turn_key: str | None,
+    classification: str,
+    delivery_outcome: str | None,
+) -> RefreshReplanQualification:
+    """Qualify one refresh's replan delta and accountable settlement outcome."""
+
+    requested_classification = classification
+    requested_delivery_outcome = delivery_outcome
+    repair_delta_contract = None
+    frontier_identity = None
+    effective_recorded = bool(autonomous_replan_recorded)
+    if autonomous_replan_recorded:
+        repair_delta_contract = build_repair_delta_contract(
+            requested_delta_kinds=requested_delta_kinds,
+            active_state_next_action_update=active_state_next_action_update,
+            agent_vision=agent_vision,
+            existing_agent_vision=existing_agent_vision,
+            agent_todo_summary=parse_active_state_todos(
+                state_text, item_limit=None
+            ).get("agent_todos"),
+            agent_id=agent_id or None,
+            dry_run=dry_run,
+            selected_todo_id=settlement_todo_id,
+            newest_first_runs=newest_first_runs,
+        )
+        validated_delta_kinds = list(repair_delta_contract.get("delta_kinds") or [])
+        if not repair_delta_contract["delta_present"]:
+            normalized = str(classification or "").strip().lower()
+            classification = (
+                "repair_noop"
+                if "repair" in normalized and "replan" not in normalized
+                else "replan_noop"
+            )
+            effective_recorded = False
+            if delivery_outcome in {"outcome_progress", "primary_goal_outcome"}:
+                delivery_outcome = "outcome_gap"
+        elif (
+            delivery_outcome in {"outcome_progress", "primary_goal_outcome"}
+            and not repair_delta_kinds_have_accountable_progress(
+                validated_delta_kinds
+            )
+        ):
+            delivery_outcome = "outcome_gap"
+        if effective_recorded and agent_id:
+            frontier_identity = latest_monitor_replan_frontier_identity(
+                newest_first_runs,
+                agent_id=agent_id,
+                watch_todo_ids=watch_lane_continuation_todo_ids(
+                    repair_delta_contract
+                ),
+            )
+
+    semantic_delta = enforce_open_replan_writeback(
+        newest_first_runs=newest_first_runs,
+        state_text=state_text,
+        agent_id=agent_id,
+        goal_id=goal_id,
+        progress_observation=progress_observation,
+        registry_goal=registry_goal,
+        agent_vision=agent_vision,
+        completion_todo_id=completion_todo_id,
+        completion_turn_key=completion_turn_key,
+    )
+    if semantic_delta:
+        effective_recorded = True
+        classification = requested_classification
+        delivery_outcome = requested_delivery_outcome
+    if settlement_bound and delivery_outcome not in ACCOUNTABLE_DELIVERY_OUTCOMES:
+        raise ValueError(
+            "turn-scoped refresh-state produced no accountable semantic delta; "
+            "no state or settlement receipt was written"
+        )
+    return RefreshReplanQualification(
+        repair_delta_contract=repair_delta_contract,
+        semantic_delta=semantic_delta,
+        frontier_identity=frontier_identity,
+        classification=classification,
+        delivery_outcome=delivery_outcome,
+        autonomous_replan_recorded=effective_recorded,
+    )

loopx/state_refresh.py

@@ -31,19 +31,13 @@
 from .control_plane.work_items.repair_delta import (
     REPAIR_DELTA_CONTRACT_SCHEMA_VERSION as REPAIR_DELTA_CONTRACT_SCHEMA_VERSION,
     REPAIR_DELTA_KIND_CHOICES as REPAIR_DELTA_KIND_CHOICES,
-    build_repair_delta_contract,
     normalize_repair_delta_kinds,
-    repair_delta_kinds_have_accountable_progress,
-)
-from .control_plane.work_items.autonomous_replan_ack import (
-    latest_monitor_replan_frontier_identity,
-    watch_lane_continuation_todo_ids,
 )
 from .control_plane.work_items.progress_observation import (
     normalize_progress_observation,
 )
 from .control_plane.work_items.semantic_replan_writeback import (
-    enforce_open_replan_writeback,
+    qualify_refresh_replan_writeback,
 )
 from .control_plane.runtime.shared_runtime_refresh_projection import (
     build_shared_runtime_projection,
@@ -84,7 +78,6 @@
     normalize_todo_claimed_by,
     normalize_todo_replan_obligation_id,
 )
-from .control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from .control_plane.todos.completion_validation_accountability import (
     require_accountable_completion_validation,
 )
@@ -232,13 +225,6 @@ def normalize_progress_scope(value: str | None) -> str:
     return normalized
 
 
-def _noop_classification_for(classification: str) -> str:
-    normalized = str(classification or "").strip().lower()
-    if "repair" in normalized and "replan" not in normalized:
-        return "repair_noop"
-    return "replan_noop"
-
-
 def next_action_section_bounds(lines: list[str]) -> tuple[int, int] | None:
     for index, line in enumerate(lines):
         if line.strip() != "## Next Action":
@@ -1148,78 +1134,35 @@ def refresh_state_run(
     else:
         action, recommended_action_source = derive_recommended_action_with_source(state_text)
     validate_local_control_text("recommended_action", action)
-    repair_delta_contract: dict[str, Any] | None = None
-    validated_repair_delta_kinds: list[str] = []
     requested_classification = classification
-    requested_delivery_outcome = normalized_delivery_outcome
-    effective_autonomous_replan_recorded = bool(autonomous_replan_recorded)
-    if autonomous_replan_recorded:
-        repair_delta_contract = build_repair_delta_contract(
-            requested_delta_kinds=normalized_repair_delta_kinds,
-            active_state_next_action_update=active_state_next_action_update,
-            agent_vision=agent_vision,
-            existing_agent_vision=existing_agent_vision,
-            agent_todo_summary=parse_active_state_todos(
-                state_text, item_limit=None
-            ).get("agent_todos"),
-            agent_id=normalized_agent_id or None,
-            dry_run=dry_run,
-            selected_todo_id=(settlement_identity.todo_id if settlement_identity else None),
-            newest_first_runs=newest_first_runs,
-        )
-        validated_repair_delta_kinds = list(
-            repair_delta_contract.get("delta_kinds") or []
-        )
-        if not repair_delta_contract["delta_present"]:
-            classification = _noop_classification_for(classification)
-            effective_autonomous_replan_recorded = False
-            if normalized_delivery_outcome in {"outcome_progress", "primary_goal_outcome"}:
-                normalized_delivery_outcome = "outcome_gap"
-        elif (
-            normalized_delivery_outcome in {"outcome_progress", "primary_goal_outcome"}
-            and not repair_delta_kinds_have_accountable_progress(
-                validated_repair_delta_kinds
-            )
-        ):
-            normalized_delivery_outcome = "outcome_gap"
-        if effective_autonomous_replan_recorded and normalized_agent_id:
-            autonomous_replan_frontier_identity = (
-                latest_monitor_replan_frontier_identity(
-                    newest_first_runs,
-                    agent_id=normalized_agent_id,
-                    watch_todo_ids=watch_lane_continuation_todo_ids(
-                        repair_delta_contract
-                    ),
-                )
-            )
-    replan_semantic_delta = enforce_open_replan_writeback(
+    replan_qualification = qualify_refresh_replan_writeback(
+        autonomous_replan_recorded=autonomous_replan_recorded,
+        requested_delta_kinds=normalized_repair_delta_kinds,
+        active_state_next_action_update=active_state_next_action_update,
+        agent_vision=agent_vision,
+        existing_agent_vision=existing_agent_vision,
+        agent_id=normalized_agent_id,
+        dry_run=dry_run,
+        settlement_todo_id=(settlement_identity.todo_id if settlement_identity else None),
+        settlement_bound=settlement_identity is not None,
         newest_first_runs=newest_first_runs,
         state_text=state_text,
-        agent_id=normalized_agent_id,
         goal_id=safe_goal_id,
         progress_observation=normalized_progress_observation,
         registry_goal=registry_goal,
-        agent_vision=agent_vision,
         completion_todo_id=completion_todo_id,
         completion_turn_key=completion_turn_key,
+        classification=classification,
+        delivery_outcome=normalized_delivery_outcome,
+    )
+    repair_delta_contract = replan_qualification.repair_delta_contract
+    replan_semantic_delta = replan_qualification.semantic_delta
+    autonomous_replan_frontier_identity = replan_qualification.frontier_identity
+    classification = replan_qualification.classification
+    normalized_delivery_outcome = replan_qualification.delivery_outcome
+    effective_autonomous_replan_recorded = (
+        replan_qualification.autonomous_replan_recorded
     )
-    if replan_semantic_delta:
-        effective_autonomous_replan_recorded = True
-        # The semantic gate is authoritative for an open replan obligation.
-        # A repair-delta label may fail to describe the state change while the
-        # same writeback still carries a fresh accepted typed observation. In
-        # that case this is not a noop: preserve the caller's accountable
-        # settlement outcome so the durable run and receipt remain one chain.
-        classification = requested_classification
-        normalized_delivery_outcome = requested_delivery_outcome
-    if (
-        settlement_identity is not None
-        and normalized_delivery_outcome not in ACCOUNTABLE_DELIVERY_OUTCOMES
-    ):
-        raise ValueError(
-            "turn-scoped refresh-state produced no accountable semantic delta; "
-            "no state or settlement receipt was written"
-        )
     vision_checkpoint = build_vision_checkpoint(
         agent_id=normalized_agent_id or None,
         agent_vision=agent_vision,

tests/control_plane/test_refresh_state_replan_gate.py

@@ -16,13 +16,11 @@
 from loopx.control_plane.work_items.semantic_replan_writeback import (
     ReplanWritebackRejected,
     _obligation_was_created_by_current_completion,
+    enforce_open_replan_writeback,
     qualify_replan_writeback,
     project_replan_writeback_rejection,
 )
-from loopx.state_refresh import (
-    enforce_open_replan_writeback,
-    refresh_state_run,
-)
+from loopx.state_refresh import refresh_state_run
 from loopx.todos import add_goal_todo
 
 GOAL_ID = "replan-gate-fixture"