From 0fb8a777a1b1c6a2cb7f7dd4776c59e888df4270 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 19:08:13 +0800 Subject: [PATCH 1/2] fix(turn): validate original capture before retry preview Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/effect-interpreter-packet.md | 6 + loopx/cli_commands/turn_inspection.py | 2 + loopx/control_plane/turn_driver/executor.py | 84 ++++++++------ tests/test_turn_inspection_session_host.py | 118 ++++++++++++++------ 4 files changed, 144 insertions(+), 66 deletions(-) diff --git a/docs/reference/effect-interpreter-packet.md b/docs/reference/effect-interpreter-packet.md index 5a3c7c2361..b16bebaf98 100644 --- a/docs/reference/effect-interpreter-packet.md +++ b/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 diff --git a/loopx/cli_commands/turn_inspection.py b/loopx/cli_commands/turn_inspection.py index 1b6dca1fef..af0b4ad936 100644 --- a/loopx/cli_commands/turn_inspection.py +++ b/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, diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index a45a754f45..3ca076c661 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/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( diff --git a/tests/test_turn_inspection_session_host.py b/tests/test_turn_inspection_session_host.py index 042e301c64..12d4aab6a6 100644 --- a/tests/test_turn_inspection_session_host.py +++ b/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 From 9732c420c6852d6c7ed482044c1955a5af42277d Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 10 Nov 2026 12:51:30 +0800 Subject: [PATCH 2/2] test: retain current inspection live-path guard Signed-off-by: huangruiteng --- tests/test_loopx_turn_journal_inspection.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/tests/test_loopx_turn_journal_inspection.py b/tests/test_loopx_turn_journal_inspection.py index e1e0faea7e..bcfb37bc4e 100644 --- a/tests/test_loopx_turn_journal_inspection.py +++ b/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