refactor(control-plane): resolve turn result kind through effect turn (#2956)

Author: huangruiteng

loopx/control_plane/turn_driver/executor.py

@@ -13,6 +13,7 @@
 
 from ...authority import validate_public_safe_text
 from ...file_lock import exclusive_file_lock
+from ..effect_program import interpret_turn_result_packet
 from ..goals.goal_vision import normalize_goal_vision_packet
 from ..work_items.delivery_batch_scale import require_delivery_batch_scale
 from ..work_items.delivery_outcome import require_delivery_outcome
@@ -851,7 +852,8 @@ def _task_validation_stage(
     journal_path: Path,
     effects: dict[str, bool],
 ) -> tuple[list[str], dict[str, Any] | None]:
-    kind = LoopXTurnResultKind(str(result["result_kind"]))
+    turn = interpret_turn_result_packet(result)
+    kind = LoopXTurnResultKind(turn.observation.decision)
     if kind in STOP_HOST_RESULT_KINDS:
         completed_phases = list(TRANSACTION_PHASES[:3])
         journal.update(

tests/test_loopx_turn_executor.py

@@ -13,7 +13,11 @@
     run_loopx_turn_once,
     validate_loopx_turn_host_result,
 )
-from loopx.control_plane.turn_driver.executor import BuiltInHostError
+from loopx.control_plane.turn_driver.executor import (
+    BuiltInHostError,
+    _task_validation_stage,
+)
+from loopx.control_plane.turn_driver.transaction import TRANSACTION_PHASES
 
 
 def _plan() -> dict[str, object]:
@@ -87,6 +91,32 @@ def _host_result(plan: dict[str, object], *, kind: str = "validated_progress") -
     return result
 
 
+def test_task_validation_stage_reads_result_kind_through_effect_turn(
+    tmp_path: Path,
+) -> None:
+    plan = _plan()
+    result = _host_result(plan, kind="wait")
+    journal = {
+        "status": "in_progress",
+        "completed_phases": list(TRANSACTION_PHASES[:2]),
+    }
+
+    completed, payload = _task_validation_stage(
+        plan,
+        result,
+        task_validator=None,
+        completed_phases=list(TRANSACTION_PHASES[:2]),
+        journal=journal,
+        journal_path=tmp_path / "journal.json",
+        effects={},
+    )
+
+    assert completed == list(TRANSACTION_PHASES[:3])
+    assert journal["status"] == "stopped"
+    assert payload is not None
+    assert payload["status"] == "stopped"
+
+
 def _host_argv(result_path: Path, count_path: Path) -> list[str]:
     script = """
 import json