fix(turn): validate original capture before retry preview

docs/reference/effect-interpreter-packet.md

@@ -120,6 +120,12 @@ by `turn run-once`. When the failed Host recorded `resume_session`, inspection
 performs the current read-only Session Binding check; without an explicit retry
 request, the decision reports `failed_retry_not_requested`.
 
+When continuation would invoke Host again, inspection also uses the executor's
+read-only original decision capture check. A missing, altered, or invalid capture
+exits non-zero without effects; restore the original capture before retrying the
+same Turn. Recovery of an already saved Host result and effect-free terminal
+replay do not require that capture, including historical v0 journals.
+
 It resolves only the canonical journal location. There is no arbitrary
 `--journal-path` input. The command validates selectors, takes the existing
 journal lock, schema-checks the stored JSON, and returns the versioned

loopx/cli_commands/turn_inspection.py

@@ -81,6 +81,7 @@ def handle_turn_journal_inspection(
                         agent_id=args.agent_id,
                         turn_key=args.turn_key,
                         retry_failed=bool(args.retry_failed_turn),
+                        require_decision_capture=True,
                         session_binding_resolver=(
                             lambda turn_envelope: codex_cli_session_binding(
                                 runtime_root,
@@ -155,6 +156,7 @@ def handle_turn_journal_inspection(
             agent_id=args.agent_id,
             turn_key=args.turn_key,
             retry_failed=bool(args.retry_failed_turn),
+            require_decision_capture=True,
             session_binding_resolver=(
                 lambda turn_envelope: codex_cli_session_binding(
                     runtime_root,

loopx/control_plane/turn_driver/executor.py

@@ -717,6 +717,7 @@ def inspect_loopx_turn_journal(
     turn_key: str,
     retry_failed: bool = False,
     session_binding_resolver: SessionBindingResolver | None = None,
+    require_decision_capture: bool = False,
 ) -> dict[str, object]:
     """Inspect one canonical Turn journal without granting execution authority."""
 
@@ -731,46 +732,61 @@ def inspect_loopx_turn_journal(
     try:
         with exclusive_file_lock(path):
             journal = _load_journal(path)
+            if journal is None:
+                raise ValueError("LoopX Turn journal does not exist")
+
+            request: dict[str, Any] = {}
+            session_check = None
+            recovery_error = None
+            assess_session = True
+            if retry_failed and journal.get("status") == "failed":
+                plan = journal.get("plan")
+                if isinstance(plan, Mapping):
+                    try:
+                        request = build_loopx_turn_host_request(plan)
+                    except (TypeError, ValueError) as exc:
+                        assess_session = False
+                        recovery_error = ValueError("failed-Turn Host request is invalid")
+                        session_check = {
+                            "kind": "host_session_binding",
+                            "outcome": "failed",
+                            "reason": "host_request_invalid",
+                        }
+                        recovery_error.__cause__ = exc
+
+            assessment = assess_existing_turn_recovery(
+                journal,
+                request,
+                goal_id=safe_goal_id,
+                agent_id=agent_id,
+                turn_key=turn_key,
+                retry_failed=retry_failed,
+                session_binding_resolver=session_binding_resolver,
+                assess_session=assess_session,
+                session_recovery_check=session_check,
+                recovery_error=recovery_error,
+            )
+            if (require_decision_capture
+                    and assessment.decision.get("action") == "continue"
+                    and assessment.decision.get("reinvoke_host") is True):
+                # Use dispatch's read-only check only when recovery needs Host
+                # input. Stored results and terminal replay need no new input.
+                try:
+                    stage_turn_decision_capture(
+                        journal["plan"], journal_path=path,
+                        decision=None, existing_journal=True,
+                    )
+                except json.JSONDecodeError:
+                    raise ValueError("Turn decision capture contains malformed JSON") from None
+                except OSError:
+                    raise ValueError("Turn decision capture could not be read") from None
+            return assessment.inspection
     except json.JSONDecodeError:
         raise ValueError("LoopX Turn journal contains malformed JSON") from None
     except LockAcquireTimeoutError:
         raise
     except OSError:
         raise ValueError("LoopX Turn journal could not be read") from None
-    if journal is None:
-        raise ValueError("LoopX Turn journal does not exist")
-
-    request: dict[str, Any] = {}
-    session_check = None
-    recovery_error = None
-    assess_session = True
-    if retry_failed and journal.get("status") == "failed":
-        plan = journal.get("plan")
-        if isinstance(plan, Mapping):
-            try:
-                request = build_loopx_turn_host_request(plan)
-            except (TypeError, ValueError) as exc:
-                assess_session = False
-                recovery_error = ValueError("failed-Turn Host request is invalid")
-                session_check = {
-                    "kind": "host_session_binding",
-                    "outcome": "failed",
-                    "reason": "host_request_invalid",
-                }
-                recovery_error.__cause__ = exc
-
-    return assess_existing_turn_recovery(
-        journal,
-        request,
-        goal_id=safe_goal_id,
-        agent_id=agent_id,
-        turn_key=turn_key,
-        retry_failed=retry_failed,
-        session_binding_resolver=session_binding_resolver,
-        assess_session=assess_session,
-        session_recovery_check=session_check,
-        recovery_error=recovery_error,
-    ).inspection
 
 
 def _receipt(

tests/test_loopx_turn_journal_inspection.py

@@ -595,10 +595,7 @@ def test_inspect_journal_cli_branches_before_live_or_write_paths(
     def unexpected_call(*args: object, **kwargs: object) -> None:
         raise AssertionError("inspect-journal reached a live or write path")
 
-    for name in (
-        "build_loopx_turn_plan",
-        "build_turn_envelope",
-    ):
+    for name in ("build_loopx_turn_plan",):
         monkeypatch.setattr(turn_command, name, unexpected_call)
     monkeypatch.setattr(turn_run_once, "execute_turn_run_once", unexpected_call)
     # The shared decision owner now performs the live reads this command used to

tests/test_turn_inspection_session_host.py

@@ -11,40 +11,40 @@
 
 from loopx.cli import main
 from loopx.control_plane import turn_driver
+from loopx.control_plane.quota.turn_envelope import build_turn_envelope
 from loopx.control_plane.turn_driver.codex_sessions import (
     _session_path,
     _write_codex_cli_session,
 )
 from loopx.control_plane.turn_driver.executor import BuiltInHostError, turn_journal_path
+from tests.test_turn_envelope import _full_decision
 
 
-def _failed_turn(tmp_path: Path, *, host: str, scope: str) -> tuple[dict, Path, Path]:
-    envelope = {
-        "ok": True,
-        "schema_version": "loopx_turn_envelope_v0",
-        "goal_id": "fixture-goal",
-        "agent_id": "fixture-agent",
-        "should_run": True,
-        "effective_action": "normal_run",
-        "action": {
-            "must_attempt": True,
-            "delivery_allowed": True,
-            "quiet_noop_allowed": False,
-            "selected_todo": {
-                "todo_id": "todo_fixture0001", "text": "Advance one public fixture",
-            },
-        },
-        "user": {"action_required": False, "open_count": 0, "notify": "DONT_NOTIFY"},
-        "writeback": {"spend_after_validation": True},
-        "scheduler": {"action": "run_now"},
-        "action_signature": {
-            "matches": True, "source_hash": "sha256:fixture",
-            "envelope_hash": "sha256:fixture",
-        },
-        "compaction": {"within_budget": True},
-    }
+def _complete_source(tmp_path: Path) -> dict:
+    from tests.test_loopx_turn_driver import _write_live_fixture
+
+    _, runtime, registry = _write_live_fixture(tmp_path / "source")
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        code = main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "turn", "plan", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture",
+            "--host", "generic-cli", "--execution-mode", "isolated-headless",
+            "--include-transaction-detail",
+        ])
+    plan = json.loads(output.getvalue())
+    assert code == 0 and plan["ok"], plan
+    return plan["work_contract"]
+
+
+def _failed_turn(
+    tmp_path: Path, *, host: str, scope: str, legacy: bool = True,
+) -> tuple[dict, Path, Path]:
+    decision = _full_decision() if legacy else _complete_source(tmp_path)
+    goal_id = decision["goal_id"]
     plan = turn_driver.build_loopx_turn_plan(
-        envelope, host=host, execution_mode="isolated-headless", session_scope=scope,
+        build_turn_envelope(decision) if legacy else decision,
+        host=host, execution_mode="isolated-headless", session_scope=scope,
     )
 
     def interrupted(_request):
@@ -55,12 +55,13 @@ def no_effect(*_args, **_kwargs):
 
     result = turn_driver.run_loopx_turn_once(
         plan, host_runner=interrupted, project=tmp_path, runtime_root=tmp_path,
-        goal_id="fixture-goal", timeout_seconds=5, execute=True,
+        goal_id=goal_id, timeout_seconds=5, execute=True,
         writeback=no_effect, spend=no_effect, scheduler=no_effect,
+        require_decision_capture=True, decision_capture_payload=decision,
     )
     assert result["status"] == "failed"
     lineage = {
-        "goal_id": "fixture-goal", "agent_id": "fixture-agent",
+        "goal_id": goal_id, "agent_id": "codex-fixture",
         "todo_id": "todo_fixture0001",
     }
     _write_codex_cli_session(
@@ -70,19 +71,20 @@ def no_effect(*_args, **_kwargs):
     return (
         plan,
         turn_journal_path(
-            tmp_path, goal_id="fixture-goal", turn_key=plan["transaction"]["turn_key"],
+            tmp_path, goal_id=goal_id, turn_key=plan["transaction"]["turn_key"],
         ),
         _session_path(tmp_path, lineage, scope),
     )
 
 
-def _inspect(tmp_path: Path, plan: dict, *, watch: bool, agent: str = "fixture-agent"):
+def _inspect(tmp_path: Path, plan: dict, *, watch: bool, agent: str = "codex-fixture"):
+    source = plan.get("work_contract") or plan["turn_envelope"]
     output = io.StringIO()
     with contextlib.redirect_stdout(output):
         code = main([
             "--registry", str(tmp_path / "registry.json"),
             "--runtime-root", str(tmp_path), "turn", "inspect-journal",
-            "--goal-id", "fixture-goal", "--agent-id", agent,
+            "--goal-id", source["goal_id"], "--agent-id", agent,
             "--turn-key", plan["transaction"]["turn_key"], "--retry-failed-turn",
             *(["--watch", "--watch-interval", "0.01"] if watch else []),
             "--format", "json",
@@ -131,7 +133,7 @@ def test_inspection_does_not_fallback_to_an_unrelated_binding(
         # A valid Codex binding at the same scope must not stand in for DSH.
         _write_codex_cli_session(
             tmp_path, lineage={
-                "goal_id": "fixture-goal", "agent_id": "fixture-agent",
+                "goal_id": "fixture-goal", "agent_id": "codex-fixture",
                 "todo_id": "todo_fixture0001",
             }, host="codex-cli", session_id="other-host-session",
         )
@@ -158,3 +160,55 @@ def capture(*args, **kwargs):
     assert decision["reinvoke_host"] is False
     assert observed[0]["effects"] == []
     assert journal.read_bytes() == before
+
+
+@pytest.mark.parametrize("watch", [False, True])
+@pytest.mark.parametrize("legacy", [False, True])
+@pytest.mark.parametrize(("fault", "message"), [
+    ("missing", "Original Turn decision capture is missing"),
+    ("tampered", "Turn decision capture does not match the original observation"),
+    ("symlink", "Turn decision capture must not be a symlink"),
+    ("malformed", "Turn decision capture contains malformed JSON"),
+    ("directory", "capture must be a regular file"),
+])
+def test_inspection_refuses_to_retry_with_unusable_capture(
+    tmp_path: Path, watch: bool, legacy: bool, fault: str, message: str,
+) -> None:
+    plan, journal, session = _failed_turn(tmp_path, host="codex-cli", scope="todo", legacy=legacy)
+    capture = journal.parent / "decisions" / journal.name
+    original = capture.read_bytes()
+    if fault == "missing":
+        capture.unlink()
+    elif fault == "tampered":
+        changed = {**json.loads(original), "reason": "altered after the failed Turn"}
+        capture.write_text(json.dumps(changed), encoding="utf-8")
+    elif fault == "malformed":
+        capture.write_text("{", encoding="utf-8")
+    elif fault == "directory":
+        capture.unlink()
+        capture.mkdir()
+    else:
+        target = capture.rename(capture.with_name("redirected.json"))
+        try:
+            capture.symlink_to(target)
+        except OSError:
+            pytest.skip("creating a symlink requires permission on this host")
+    before = journal.read_bytes(), session.read_bytes()
+
+    code, (result,) = _inspect(tmp_path, plan, watch=watch)
+
+    assert code == 1
+    assert result["ok"] is False and message in result["error"]
+    assert result["effects"] == []
+    assert (journal.read_bytes(), session.read_bytes()) == before
+
+    if capture.is_dir():
+        capture.rmdir()
+    else:
+        capture.unlink(missing_ok=True)
+    capture.write_bytes(original)
+    restored_code, restored_output = _inspect(tmp_path, plan, watch=watch)
+    assert restored_code == 0
+    assert all(row["effects"] == [] for row in restored_output)
+    assert (journal.read_bytes(), session.read_bytes()) == before
+    assert capture.read_bytes() == original