fix: let terminal qualification correct effect-free vision input

docs/reference/protocols/model-behavior-qualification-v0.md

@@ -309,17 +309,26 @@ writeback and spend receipts. Later receipts advance the original Turn's plan;
 they do not discard its identity-bound command templates. This follows the shared
 [quota-bound completion contract](../canonical-terminal-review.md#quota-bound-cli-completion)
 without changing the portfolio, repeats, contrasts or settlement order.
-The refresh CLI's pure progress-input decoder identifies invalid argument
-combinations before dispatch or authority writes. Its structured rejection is
-returned to the same model, which may correct its input within the original
-ten-call budget. Rejected calls remain counted and commit no settlement step;
-the actor never retries a command or restarts a scenario automatically. Other
-non-zero CLI results and fixture identity, route or ordering violations still
-fail immediately; non-zero execution alone does not prove rollback.
+The terminal actor observes the complete disposable fixture before a refresh
+command. An exit-one JSON rejection with `ok=false` and `appended=false` is
+returned to the same model only when an independent file/metadata comparison
+also finds the fixture unchanged. This covers ordinary vision-input correction
+as well as the typed pure progress-input decoder; error wording is not an
+admission rule. Complete vision JSON is structurally checked by the existing
+typed owner before refresh acquires a history lock. Inline patches still use
+current history, and state-dependent validation remains in the normal writer.
+File replacement, same-byte rewrites, permissions, added or
+removed files, unsupported nodes and symlinks fail closed. These observations
+apply to this isolated synthetic fixture, not production rollback or external
+effect recovery.
+The model may correct its input within the original ten-call budget. Rejected
+calls remain counted and commit no settlement step; the actor never retries a
+command or restarts a scenario automatically. Other non-zero results, changed
+fixture state and identity, route or ordering violations still fail immediately.
 The shared CLI transport retains the process exit code and transient stdout;
-even a non-zero process that emitted valid committed-receipt JSON fails unless
-it is that exact pre-dispatch input rejection. Raw stdout is never copied into
-durable qualification receipts.
+even valid committed-receipt JSON cannot bypass a non-zero failure. A false
+`appended` field alone cannot authorize recovery or spend. Raw stdout, fixture
+contents and fingerprints are never copied into durable qualification receipts.
 After the proof is read, a terminal actor may author a project draft in
 `.scratch/` using the existing OS-isolated shell. Original fixture inputs and
 authority stores remain protected; private host data and external networking

loopx/cli_commands/project_lifecycle_refresh_state.py

@@ -516,10 +516,13 @@ def handle_refresh_state_command(
                 turn_instance_id=args.turn_instance_id,
             ),
         )
-        if outcome_update is not None and agent_vision_packet is not None:
-            # Reject malformed input before committing the independent outcome
-            # refinement. Reuse the typed owner; do not authorize a transition
-            # or replace the original packet with this context-free result.
+        if agent_vision_packet is not None and (
+            outcome_update is not None or not merge_agent_vision_patch
+        ):
+            # Reject malformed complete input before any history lock or outcome
+            # refinement. Inline patches without an outcome update still merge
+            # with current history first. Reuse the typed owner; this preflight
+            # cannot authorize a transition or replace the original packet.
             # Refresh still checks merged history and policy against current
             # state, including whether a terminal stride is actually enabled.
             prepare_vision_refresh(

loopx/control_plane/testing/model_tool_behavior.py

@@ -3,6 +3,7 @@
 import json
 import os
 import shlex
+import stat
 import subprocess
 import sys
 from collections.abc import Callable, Mapping, Sequence
@@ -569,3 +570,28 @@ def __init__(self, returncode: int, detail: str, *, stdout: str = "") -> None:
         # Transient tool feedback only. Durable receipts must never serialize
         # this raw output or infer rollback from the presence of valid JSON.
         self.stdout = stdout
+
+
+def snapshot_disposable_fixture(root: Path) -> dict[str, tuple[Any, ...]]:
+    """Observe all fixture files, never infer production rollback from stdout.
+
+    Contents remain transient hashes. Metadata catches replacement, permission
+    changes and same-byte rewrites; access times do not make reads into writes.
+    Unknown nodes and links fail closed rather than hide effects outside root.
+    """
+    root = root.resolve(strict=True)
+    observed: dict[str, tuple[Any, ...]] = {}
+    for path in (root, *root.rglob("*")):
+        metadata = path.lstat()
+        mode = metadata.st_mode
+        identity = (mode, metadata.st_uid, metadata.st_gid, metadata.st_ino,
+                    metadata.st_mtime_ns, metadata.st_ctime_ns)
+        value: tuple[Any, ...]
+        if stat.S_ISDIR(mode):
+            value = identity
+        elif stat.S_ISREG(mode):
+            value = (*identity, sha256(path.read_bytes()).hexdigest())
+        else:
+            raise ValueError("Disposable fixture contains an unsupported node")
+        observed[str(path.relative_to(root))] = value
+    return observed

loopx/control_plane/testing/terminal_settlement_tool_behavior.py

@@ -30,6 +30,7 @@
     digest_text,
     execute_loopx_cli,
     loopx_command_tokens,
+    snapshot_disposable_fixture,
 )
 from .selected_todo_tool_behavior import bounded_workspace_read_plan
 from .vision_shell_host import VisionShellHost
@@ -73,6 +74,10 @@ class _TerminalToolViolation(ValueError):
     """A fixture-owned invariant code, never provider or CLI error text."""
 
 
+class _UnchangedFixtureRejection(dict[str, Any]):
+    """Real CLI rejection independently observed to leave this fixture intact."""
+
+
 def _terminal_failure_code(error: Exception) -> str:
     return str(error) if isinstance(error, _TerminalToolViolation) else "tool_execution_failed"
 
@@ -711,6 +716,8 @@ def _execute_cli(command: str, state: _QualificationState) -> dict[str, Any]:
             raise _TerminalToolViolation("vision_draft_outside_workspace")
     if tokens and "--format" not in tokens:
         command = shlex.join([tokens[0], "--format", "json", *tokens[1:]])
+    before = (snapshot_disposable_fixture(fixture.project_root.parent)
+              if _command_path(tokens or [])[0] == "refresh_state" else None)
     try:
         output = execute_loopx_cli(
             command,
@@ -730,9 +737,10 @@ def _execute_cli(command: str, state: _QualificationState) -> dict[str, Any]:
             raise exc from None
         if not (isinstance(rejected, dict) and rejected.get("ok") is False
                 and rejected.get("appended") is False
-                and rejected.get("error_code") == "progress_observation_invalid_arguments"):
+                and before is not None
+                and snapshot_disposable_fixture(fixture.project_root.parent) == before):
             raise
-        return rejected
+        return _UnchangedFixtureRejection(rejected)
     payload = json.loads(output)
     if not isinstance(payload, dict):
         raise ValueError("LoopX command output must be an object")
@@ -935,12 +943,13 @@ def _execute_terminal_cli(
         if not tokens or not _matches_identity(tokens, state):
             raise _TerminalToolViolation("writeback_identity_mismatch")
         payload = _execute_cli(command, state)
-        if (payload.get("ok") is False and payload.get("appended") is False
-                and payload.get("error_code") == "progress_observation_invalid_arguments"):
-            # Only the CLI's pure input decoder emits this code, before refresh
-            # dispatch. Give the same model its real tool error within the
-            # original call budget; do not retry or count a committed effect.
-            return json.dumps(payload, ensure_ascii=False, sort_keys=True), "progress_observation_invalid_arguments"
+        if isinstance(payload, _UnchangedFixtureRejection):
+            # A real error plus independently unchanged fixture files permits
+            # correction, not settlement. Keep every call in the same budget.
+            kind = ("progress_observation_invalid_arguments"
+                    if payload.get("error_code") == "progress_observation_invalid_arguments"
+                    else "fixture_unchanged_cli_rejection")
+            return json.dumps(payload, ensure_ascii=False, sort_keys=True), kind
         receipt_steps = _settlement_receipt_steps(payload)
         if receipt_steps != _EXPECTED_RECEIPT_STEPS[:2]:
             raise _TerminalToolViolation("writeback_receipt_mismatch")

tests/control_plane/test_outcome_replan_composition_cli.py

@@ -18,6 +18,42 @@
 )
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_invalid_complete_vision_leaves_fixture_untouched_before_history_lock(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
+) -> None:
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    vision = project / "invalid-vision.json"
+    vision.write_text(json.dumps({"state": "vision_closed", "vision_patch": {}}))
+
+    def files() -> dict:
+        return {
+            (str(root), str(path.relative_to(root))): (
+                path.stat().st_mode, path.stat().st_ino,
+                path.stat().st_mtime_ns, path.stat().st_ctime_ns,
+                path.read_bytes() if path.is_file() else None,
+            )
+            for root in (project, runtime)
+            for path in (root, *root.rglob("*"))
+        }
+
+    before = files()
+    code, result = _run_cli(
+        registry, runtime, "refresh-state", "--goal-id", GOAL_ID,
+        "--agent-id", AGENT_ID, "--classification", "validated_delivery",
+        "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress",
+        "--agent-vision-json", str(vision), "--no-global-sync", "--suppress-external-sinks",
+        cwd=project,
+    )
+    assert code == 1 and result["ok"] is False and result["appended"] is False
+    assert "at least one bounded vision field" in result["error"]
+    assert files() == before
+    assert not (runtime / "goals" / GOAL_ID / "runs").exists()
+
+
 def _review_packet(control: dict | None) -> dict:
     packet = {
         "schema_version": "goal_vision_replan_contract_v0",

tests/control_plane/test_terminal_settlement_tool_behavior.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import os
 import shlex
 import subprocess
 from collections.abc import Mapping
@@ -432,6 +433,59 @@ def completion(request):
         assert receipt["observed_tool_sequence"].count("progress_observation_invalid_arguments") == 1
 
 
+@pytest.mark.skipif(not shell_isolation_available(), reason="Draft authoring requires native OS isolation")
+def test_terminal_corrects_rejected_vision_within_original_budget(tmp_path: Path) -> None:
+    fixture = _build_fixture(tmp_path / "oracle")
+    before_rejection: dict[str, bytes] = {}
+
+    def authority_files() -> dict[str, bytes]:
+        root = tmp_path / "actor"
+        return {str(path.relative_to(root)): path.read_bytes()
+                for tree in (root / "runtime", root / "project/.codex", root / "project/.loopx", root / "home")
+                for path in tree.rglob("*") if path.is_file()}
+
+    def draft(request, *, corrected=False):
+        if corrected:
+            rejected = json.loads(request["messages"][-1]["content"])
+            assert rejected["ok"] is False and rejected["appended"] is False
+            assert authority_files() == before_rejection
+        vision = {
+            "schema_version": "goal_vision_replan_contract_v0", "state": "vision_closed",
+            "vision_patch": {},
+        }
+        if corrected:
+            vision["vision_patch"] = {"acceptance_summary": "Original proof and Todo validation are retained."}
+            vision["path_delta"] = {
+                "outcome": "no_change", "prior_assumption": "Final delivery requires validation.",
+                "observed_reality": "Original proof and Todo validation were checked.",
+                "retained": ["Original validated delivery"], "evidence_refs": ["fixture-settlement-proof"],
+            }
+        return ScriptedExecToolAction(command="cat > .scratch/vision.json <<'JSON'\n" +
+                                      json.dumps(vision) + "\nJSON")
+
+    def writeback(request):
+        before_rejection.clear()
+        before_rejection.update(authority_files())
+        return ScriptedExecToolAction(command=_writeback_action(request).command +
+                                      " --agent-vision-json .scratch/vision.json")
+
+    transport = ScriptedDoubaoExecTransport([
+        ScriptedExecToolAction(command=fixture.quota_guard_command),
+        ScriptedExecToolAction(command="cat fixture/settlement-proof.json"),
+        partial(_completion_action, kind="validation"), draft, writeback,
+        partial(draft, corrected=True), writeback, _spend_action, _completion_action,
+    ])
+    receipt = DoubaoTerminalSettlementToolBehaviorActor(
+        api_key="test-only-placeholder", transport=transport,
+    ).qualify(qualification_id="native-vision-input-correction", fixture_root=tmp_path / "actor")
+    assert receipt["qualification_passed"] is True, receipt
+    assert receipt["tool_call_count"] == len(transport.requests) == 9
+    assert receipt["settlement_receipt_steps"] == [
+        "validation", "durable_writeback", "quota_spend", "terminal_closeout",
+    ]
+    assert (tmp_path / "actor/project/fixture/settlement-proof.json").read_bytes() == fixture.proof_target.read_bytes()
+
+
 def test_terminal_input_rejections_do_not_reset_the_call_budget(tmp_path: Path) -> None:
     fixture = _build_fixture(tmp_path / "oracle")
 
@@ -455,9 +509,9 @@ def rejected_writeback(request):
     assert receipt["settlement_receipt_steps"] == ["validation"]
 
 
-@pytest.mark.parametrize("payload_ok", [False, True])
+@pytest.mark.parametrize(("payload_ok", "payload_appended"), [(False, True), (True, True), (False, False)])
 def test_nonzero_after_actual_writeback_cannot_advance_to_spend(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, payload_ok: bool,
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, payload_ok: bool, payload_appended: bool,
 ) -> None:
     from loopx.control_plane.quota.settlement import read_heartbeat_settlement
     fixture = _build_fixture(tmp_path / "oracle")
@@ -474,6 +528,7 @@ def fail_after_writeback(argv, *args, **kwargs):
             # Inject only a post-effect process failure; the actual CLI above
             # produced the independent writeback and committed receipt.
             payload["ok"] = payload_ok
+            payload["appended"] = payload_appended
             return subprocess.CompletedProcess(
                 completed.args, 1, json.dumps(payload), "synthetic post-effect failure",
             )
@@ -505,6 +560,73 @@ def fail_after_writeback(argv, *args, **kwargs):
     assert readback.progress["next_step"] == "quota_spend"
 
 
+@pytest.mark.parametrize("change", [
+    "same_bytes", "replacement", "added_file", "transient_file",
+    pytest.param("permissions", marks=pytest.mark.skipif(
+        os.name == "nt", reason="Windows chmod cannot toggle executable permission",
+    )),
+    pytest.param("symlink", marks=pytest.mark.skipif(
+        os.name == "nt", reason="Windows symlink creation needs separate privileges",
+    )),
+])
+def test_input_rejection_with_changed_fixture_cannot_continue(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, change: str,
+) -> None:
+    from loopx.control_plane.testing import terminal_settlement_tool_behavior as terminal
+    from loopx.control_plane.testing.model_tool_behavior import LoopxCliExecutionError
+
+    fixture = _build_fixture(tmp_path / "oracle")
+    execute = terminal.execute_loopx_cli
+
+    def mutate_after_rejection(command, **kwargs):
+        try:
+            return execute(command, **kwargs)
+        except LoopxCliExecutionError as exc:
+            assert "refresh-state" in command
+            rejected = json.loads(exc.stdout)
+            assert rejected["ok"] is False and rejected["appended"] is False
+            proof = kwargs["project_root"] / "fixture/settlement-proof.json"
+            metadata = proof.stat()
+            if change == "same_bytes":
+                proof.write_bytes(proof.read_bytes())
+                os.utime(proof, ns=(metadata.st_atime_ns, metadata.st_mtime_ns - 1_000_000_000))
+            elif change == "replacement":
+                replacement = proof.with_suffix(".replacement")
+                replacement.write_bytes(proof.read_bytes())
+                replacement.replace(proof)
+            elif change == "permissions":
+                proof.chmod(metadata.st_mode ^ 0o100)
+            elif change == "added_file":
+                proof.with_suffix(".unexpected").write_text("unexpected fixture write")
+            elif change == "transient_file":
+                transient = proof.with_suffix(".temporary")
+                transient.write_text("unexpected fixture write")
+                transient.unlink()
+            else:
+                proof.unlink()
+                proof.symlink_to(fixture.proof_target)
+            raise
+
+    def reject(request):
+        return ScriptedExecToolAction(command=_writeback_action(request).command +
+                                      " --progress-result-class no_followup")
+
+    monkeypatch.setattr(terminal, "execute_loopx_cli", mutate_after_rejection)
+    transport = ScriptedDoubaoExecTransport([
+        ScriptedExecToolAction(command=fixture.quota_guard_command),
+        ScriptedExecToolAction(command="cat fixture/settlement-proof.json"),
+        partial(_completion_action, kind="validation"), reject, _spend_action,
+    ])
+    receipt = DoubaoTerminalSettlementToolBehaviorActor(
+        api_key="test-only-placeholder", transport=transport,
+    ).qualify(qualification_id="rejection-with-mutation", fixture_root=tmp_path / "actor")
+    assert receipt["qualification_passed"] is False
+    assert receipt["failure_code"] == "tool_execution_failed"
+    assert receipt["tool_call_count"] == len(transport.requests) == 4
+    assert receipt["settlement_receipt_steps"] == ["validation"]
+    assert "unexpected fixture write" not in json.dumps(receipt)
+
+
 def test_ordinary_completion_requires_its_actual_validation_receipt(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
 ) -> None: