diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md
index ffb7a6d524..5dababc6c3 100644
--- a/docs/development/testing-and-quality.md
+++ b/docs/development/testing-and-quality.md
@@ -817,6 +817,8 @@ vision replan 未关闭时不能提前结束 Goal；这不是完整 Claude/Codex
 ```bash
 # No provider call by default. Explicit release opt-in uses ARK_API_KEY from the environment.
 python3 scripts/qualify-claude-goal-release.py --release-live
+# A completed inventory stage must not hide unfinished integrity acceptance.
+python3 scripts/qualify-claude-goal-release.py --release-live --scenario replan
 ```
 
 This arm uses the same ledger specification, independent oracle and durable
@@ -856,6 +858,77 @@ The same delivery class must pass failed-validation rejection and committed
 response-loss recovery without duplicate spending or premature terminal closure.
 Do not relabel delivery work or weaken the independent oracle to pass a host test.
 
+The `replan` scenario starts from a real, settled filename-inventory Todo and its
+valid `vision_closed` stage decision, not a fabricated missing writeback. The
+business specification still requires file sizes, checksums and a read-only
+integrity verifier. It requires an explicit successor vision/path decision,
+completed concrete successor work and terminal readback. The independent oracle
+checks actual hashes and sizes, then changes, removes and adds files in disposable
+copies; a verifier that silently regenerates its evidence fails. The original
+inventory must retain exactly one spend. This complements the finite delivery
+scenario: a model that simply closes every vision cannot pass both.
+The oracle does not require the literal final disposition `replan`: after the
+new successor has actually delivered, `no_followup` + `stop` is a valid scoped
+decision. It must still pass independent artifact, new successor, durable receipt
+and fresh terminal checks; `vision_closed` + `stop` is not Goal closure.
+
+`replan` 场景从真实完成并结算、具有有效 `vision_closed` 判断的“文件名清单”阶段启动，
+但完整验收仍缺少大小、校验和与只读校验器。测试要求后继 vision/显式路径调整、
+具体后继交付及最终终态；独立验收在一次性副本里篡改、删除、新增文件，拒绝通过
+自动重建清单掩盖错误。每个 Todo 必须恰好结算一次。后继真实交付后，最终路径可为
+`no_followup` + `stop`，不强求字面值 `replan`；`vision_closed` + `stop` 仍不是 Goal
+完成。两种场景都仅 release 前显式运行，普通 CI 不调用模型。
+
+### MCP vision authoring and recovery / MCP vision 写入与恢复
+
+The MCP guard projects `interaction_contract.mcp_channel`. For admitted normal
+Todo delivery, it replaces the raw CLI writeback/spend instructions with
+`complete_task` ownership; those are alternate transports, not two obligations.
+Replan-only and blocked lanes preserve their live CLI actions and binding.
+Vision field and total limits come from the same TS validator, not copied prompt
+constants. Quota admission, permission and workspace facts are unchanged.
+
+MCP 的普通 Todo 交付不再同时要求模型执行 CLI 记账和 MCP 结算两套流程；独立 replan
+仍使用动态 CLI 契约。vision 字段与预算直接来自 TS 校验器，不要求模型猜格式或翻测试。
+
+`complete_task` accepts either `agent_vision` (the existing bounded
+`goal_vision_replan_contract_v0` JSON packet) or `vision_unchanged_reason`.
+An unchanged decision needs a persisted valid baseline. The TypeScript host
+plan forwards that authored decision to its ordinary writeback; v1 requests
+fail closed against old runtimes instead of silently dropping the fields.
+Syntax and vision-budget preflight reuse the TS validator before lifecycle writes;
+baseline-dependent checks still run at writeback. Oversized authoring is a
+correctable input failure, not a terminal Goal failure. If an older/interrupted
+host already completed the Todo but failed writeback, retry `complete_task` with
+the same completion intent and a corrected uncommitted vision. Checkpoint-only
+recovery is not a substitute for unfinished settlement.
+
+If a previously completed MCP Todo omitted its decision, call
+`review_task_vision(todo_id, agent_id, agent_vision=...)` with that same Todo.
+It uses the original host Turn and the same writeback command constructor,
+delegating to the existing typed checkpoint recovery. It neither repeats Todo
+completion nor spends again. Exact replay is idempotent; a conflicting committed
+decision or a later superseding vision is rejected. It does not change Next
+Action or erase other work, gates, or permissions. A genuinely new replan follows
+the current interaction contract under a fresh admitted binding, not an edit to
+an already committed decision. Claude Todo-less replan now projects that identity
+re-entry before any refresh/spend instructions; ordinary MCP Todo delivery is
+unchanged.
+
+Todo acceptance, settled accounting, checkpoint satisfaction and Goal termination
+are separate facts. `vision_closed` closes a stage and still requires a successor
+vision for an active Goal. `no_followup` is an authored scoped closure assertion,
+not a substitute for evidence; remaining acceptance gaps or gates still prevent
+terminal quota. Kernel validation does not independently prove arbitrary prose
+true, so behavior qualification must also inspect the delivered artifacts.
+
+MCP 可随完成操作携带 vision 判断，也可用 `review_task_vision` 在原 Turn 补齐遗漏。
+复用 TS 的既有恢复规则，不新增结算引擎、不重扣额度；已提交的判断不能偷偷改写。
+格式和预算预检在 Todo 完成前拒绝非法输入；若旧宿主已部分完成，则修正未提交的
+vision 并重试原 `complete_task`，不能用仅补 checkpoint 的操作替代未完成结算。
+“checkpoint 满足”不等于“Goal 完成”，`vision_closed` 只结束阶段，真实缺口仍须规划。
+外层任务只描述业务验收，LoopX 协议由宿主内层指令和工具承接。
+
 ## Exact Release Commit Gate / 精确发布 Commit 门
 
 The final release gate does not rerun tests through a second orchestration
diff --git a/loopx/claude_goal_mode/scripts/goalmode_cmd.py b/loopx/claude_goal_mode/scripts/goalmode_cmd.py
index 5dff5ea4fd..dfc58ee369 100644
--- a/loopx/claude_goal_mode/scripts/goalmode_cmd.py
+++ b/loopx/claude_goal_mode/scripts/goalmode_cmd.py
@@ -79,7 +79,16 @@ def loop_execution_content(goal_id, agent_id) -> str:
         "Honor claim/lease and user/repository authority; claim only when required.\n"
         "Run real acceptance checks before `complete_task`; supply truthful evidence\n"
         f"and the bound agent_id=\"{agent_id}\". Complete only finished Todos, not partial work.\n"
-        "That MCP operation owns writeback/spend; do not repeat its accounting via CLI.\n"
+        "That MCP operation owns writeback/spend: use it INSTEAD OF the raw CLI sequence.\n"
+        "Use interaction_contract.mcp_channel for tool ownership and vision input limits.\n"
+        "At material delivery, compare the Goal's vision/acceptance with actual evidence.\n"
+        "Pass the resulting agent_vision or a justified vision_unchanged_reason to\n"
+        "complete_task. If omitted, use review_task_vision on that completed Todo to\n"
+        "repair its missing checkpoint without another spend. This is not a Goal-stop\n"
+        "shortcut: open acceptance needs replan; vision_closed closes a stage and needs\n"
+        "a successor vision; no_followup requires evidence of no remaining scoped work.\n"
+        "For new replan work not covered by these tools, use the exact live\n"
+        "interaction_contract CLI actions, preserving its binding and settlement order.\n"
         "Link already planned follow-up via successor_todo_ids; next_agent_todo creates\n"
         "new work, not a reference to an existing id. Do not duplicate the current plan.\n"
         "After a lost response, read back or retry the same completion intent; do not\n"
diff --git a/loopx/control_plane/goals/vision_checkpoint.ts b/loopx/control_plane/goals/vision_checkpoint.ts
index 3cf2d16867..9e3a15172b 100644
--- a/loopx/control_plane/goals/vision_checkpoint.ts
+++ b/loopx/control_plane/goals/vision_checkpoint.ts
@@ -21,6 +21,7 @@ const GOAL_PATH_DELTA_SCHEMA_VERSION = "goal_path_delta_v0";
 const GOAL_VISION_BUDGET_ERROR = "vision_budget_exceeded";
 // Direction and evidence-linked path changes share one bounded packet.
 const GOAL_VISION_TOTAL_LIMIT = 1_800;
+const GOAL_VISION_ADVANCEMENT_POLICIES = ["as_needed", "repeat_until_closed"] as const;
 const VISION_UNCHANGED_REASON_LIMIT = 240;
 const VISION_BUDGET_SUGGESTION_LIMIT = 96;
 
@@ -59,6 +60,33 @@ const GOAL_PATH_DELTA_LIST_LIMITS = {
   unresolved_questions: [2, 140],
   evidence_refs: [4, 140],
 } as const;
+
+/** Authoring hints share the validator's limits; they grant no transition authority. */
+export function visionAuthoringContract(): JsonObject {
+  return {
+    schema_version: GOAL_VISION_REPLAN_SCHEMA_VERSION,
+    fields: {state: "lifecycle token", vision_patch: {...GOAL_VISION_FIELD_LIMITS}},
+    common_states: ["vision_patch_proposed", "vision_closed", "no_followup"],
+    advancement_policies: [...GOAL_VISION_ADVANCEMENT_POLICIES],
+    minimal_example: {schema_version: GOAL_VISION_REPLAN_SCHEMA_VERSION, state: "vision_patch_proposed", vision_patch: {
+      vision_summary: "Scoped outcome", acceptance_summary: "Verified evidence and remaining gap",
+    }},
+    authoring_hint: "Fields are optional, not a checklist. Keep the whole decision compact; total includes path_delta. Do not copy the delivery evidence report into every field.",
+    total_text_limit: GOAL_VISION_TOTAL_LIMIT,
+    unchanged_reason_limit: VISION_UNCHANGED_REASON_LIMIT,
+    path_delta: {
+      schema_version: GOAL_PATH_DELTA_SCHEMA_VERSION,
+      outcomes: [...GOAL_PATH_DELTA_OUTCOMES],
+      required: ["outcome", "prior_assumption", "observed_reality"],
+      require_any: ["retained", "changed", "stopped"],
+      scalar_limits: {...GOAL_PATH_DELTA_SCALAR_LIMITS},
+      list_limits: Object.fromEntries(Object.entries(GOAL_PATH_DELTA_LIST_LIMITS).map(
+        ([field, [maxItems, maxChars]]) => [field, {item_type: "string", max_items: maxItems, max_item_chars: maxChars}],
+      )),
+    },
+    rule: "Compare acceptance with evidence. vision_closed closes a stage, not the Goal; no_followup requires no remaining scoped work. A changed mainline needs path_delta; respect the live replan contract.",
+  };
+}
 // Bounded typed fallback declarations survive prepare unchanged so the
 // declared direction cannot disappear behind later read-model compaction.
 const VISION_FALLBACK_DECLARATION_ENTRY_LIMIT = 4;
@@ -299,7 +327,7 @@ function normalizeGoalVisionState(value: unknown): string {
 
 function normalizeAdvancementPolicy(value: unknown): string {
   const candidate = compactText(value).toLowerCase().replaceAll("-", "_");
-  if (candidate !== "as_needed" && candidate !== "repeat_until_closed") {
+  if (!GOAL_VISION_ADVANCEMENT_POLICIES.some(policy => policy === candidate)) {
     throw new EffectRuntimeRequestError(
       "agent_vision.advancement_policy must be one of: as_needed, repeat_until_closed",
     );
@@ -624,7 +652,7 @@ function deliveryBoundary(value: unknown): DeliveryBoundary {
   throw new EffectRuntimeRequestError("delivery_boundary is unsupported");
 }
 
-function normalizeVisionUnchangedReason(value: unknown): string | null {
+export function normalizeVisionUnchangedReason(value: unknown): string | null {
   const unchanged = compactText(value);
   if (!unchanged) return null;
   validatePublicSafeText("vision_unchanged_reason", unchanged);
diff --git a/loopx/control_plane/host_adapter_settlement.py b/loopx/control_plane/host_adapter_settlement.py
index e9e2e6c20b..d3e6f6ed38 100644
--- a/loopx/control_plane/host_adapter_settlement.py
+++ b/loopx/control_plane/host_adapter_settlement.py
@@ -3,13 +3,17 @@
 from __future__ import annotations
 
 import json
+import tempfile
 from collections.abc import Mapping
-from dataclasses import dataclass
+from contextlib import contextmanager
+from dataclasses import dataclass, replace
+from pathlib import Path
 from enum import StrEnum
 from typing import Any, Protocol
 
 from .effect_program import SettlementIdentity
 from .effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from .goals.vision_checkpoint import prepare_vision_refresh
 
 
 HOST_ADAPTER_SETTLEMENT_SCHEMA_VERSION = "host_adapter_todo_settlement_v0"
@@ -55,6 +59,8 @@ class HostTodoSettlementRequest:
     execution_mode: str
     completion_args: tuple[str, ...]
     no_follow_up: bool = False
+    vision_path: str | None = None
+    vision_unchanged_reason: str | None = None
 
 
 class HostCliRunner(Protocol):
@@ -95,9 +101,63 @@ def _request_payload(
     }
     if provider_outcomes is not None:
         payload["provider_outcomes"] = provider_outcomes
+    if request.vision_path or request.vision_unchanged_reason or phase == "vision_refresh":
+        payload.update(
+            schema_version="loopx_host_todo_completion_transaction_v1",
+            vision_path=request.vision_path,
+            vision_unchanged_reason=request.vision_unchanged_reason,
+        )
     return payload
 
 
+@contextmanager
+def host_vision_request(request: HostTodoSettlementRequest, vision: dict | None, unchanged: str):
+    """Materialize authored JSON for the existing CLI codec, never as public state."""
+    if vision is not None and unchanged:
+        raise ValueError("choose a vision patch or an unchanged reason, not both")
+    if vision is not None and not isinstance(vision, dict):
+        raise ValueError("agent_vision must be a JSON object")
+    if vision is None:
+        yield replace(request, vision_unchanged_reason=unchanged or None)
+        return
+    # Reject malformed, misbound or oversized authoring before lifecycle writes.
+    # This is syntax/budget preflight only: refresh-state still validates against
+    # the real baseline and current replan/settlement state at writeback time.
+    prepare_vision_refresh(vision, goal_id=request.goal_id, agent_id=request.agent_id,
+        existing_agent_vision=None, merge_patch=False, require_path_delta_for_durable_change=False)
+    with tempfile.TemporaryDirectory(prefix="loopx-host-vision-") as directory:
+        path = Path(directory) / "vision.json"
+        path.write_text(json.dumps(vision, ensure_ascii=False, allow_nan=False), encoding="utf-8")
+        yield replace(request, vision_path=str(path))
+
+
+def refresh_host_todo_vision(request: HostTodoSettlementRequest, *, run_cli: HostCliRunner) -> str:
+    """Repair the original checkpoint; no lifecycle operation, new Turn or spend."""
+    plan = _runtime_reduction(_request_payload(request, phase="vision_refresh"), phase="vision_refresh")
+    _runtime_identity(plan.get("identity"))
+    args = plan.get("args")
+    if not isinstance(args, list) or any(not isinstance(arg, str) for arg in args):
+        raise RuntimeError("TypeScript host vision command shape mismatch")
+    return run_cli(args)
+
+
+def project_host_interaction(output: str) -> str:
+    """Keep admission facts; let the typed host lens choose the transport instructions."""
+    try:
+        packet = json.loads(output)
+    except ValueError:
+        return output
+    if not isinstance(packet, dict):
+        return output
+    projection = _runtime_reduction({
+        "schema_version": "loopx_host_todo_completion_transaction_v1",
+        "phase": "project_guard", "packet": packet,
+    }, phase="project_guard")
+    if not isinstance(projection.get("packet"), dict):
+        raise RuntimeError("TypeScript host interaction projection shape mismatch")
+    return json.dumps(projection["packet"], ensure_ascii=False)
+
+
 def _runtime_reduction(params: Mapping[str, Any], *, phase: str) -> dict[str, Any]:
     try:
         value = effect_runtime_result(_RUNTIME_METHOD, params)
diff --git a/loopx/control_plane/quota/spend_sources.py b/loopx/control_plane/quota/spend_sources.py
index e102199908..fbea522c56 100644
--- a/loopx/control_plane/quota/spend_sources.py
+++ b/loopx/control_plane/quota/spend_sources.py
@@ -5,9 +5,9 @@
 from typing import Any
 
 from ..scheduler.execution_context import (
-    NATIVE_GOAL_RUNTIME_PROFILES,
     SchedulerExecutionContextResolution,
     SchedulerRuntimeProfile,
+    VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES,
     scheduler_runtime_profile_for_execution_context,
 )
 from ..todos.contract import normalize_todo_id, normalize_todo_replan_obligation_id
@@ -41,12 +41,12 @@ def quota_spend_source_for_execution_context(
     value: Mapping[str, Any] | SchedulerExecutionContextResolution | None,
 ) -> str:
     profile = scheduler_runtime_profile_for_execution_context(value)
-    if profile in NATIVE_GOAL_RUNTIME_PROFILES:
+    if profile in VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES:
         return VISIBLE_GOAL_SLOT_SPEND_SOURCE
     return DEFAULT_SLOT_SPEND_SOURCE
 
 
-def visible_goal_turn_reentry_action(
+def host_goal_turn_reentry_action(
     payload: Mapping[str, Any],
     settlement_plan: Mapping[str, Any] | None,
     scheduler_execution_context: (
@@ -62,12 +62,16 @@ def visible_goal_turn_reentry_action(
     selected = selected_value if isinstance(selected_value, Mapping) else {}
     replan_value = payload.get("replan_action_packet")
     replan = replan_value if isinstance(replan_value, Mapping) else {}
+    replan_obligation_id = normalize_todo_replan_obligation_id(
+        replan.get("obligation_id")
+    )
     has_settlement_binding = bool(
         normalize_todo_id(selected.get("todo_id"))
-        or normalize_todo_replan_obligation_id(replan.get("obligation_id"))
+        or replan_obligation_id
     )
+    requires_turn_reentry = profile in VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES
     if (
-        profile in NATIVE_GOAL_RUNTIME_PROFILES
+        requires_turn_reentry
         and has_settlement_binding
         and settlement_plan is None
         and turn_instance_id is None
diff --git a/loopx/control_plane/scheduler/execution_context.py b/loopx/control_plane/scheduler/execution_context.py
index 8f4d8d3045..340f87ad9f 100644
--- a/loopx/control_plane/scheduler/execution_context.py
+++ b/loopx/control_plane/scheduler/execution_context.py
@@ -73,6 +73,17 @@ class GoalRuntimeContinuationDisposition(str, Enum):
     }
 )
 
+# Interactive hosts in this set can carry one explicit Turn identity through
+# accountable writeback and visible-Goal quota settlement. Claude's ordinary
+# Todo delivery remains MCP-owned; this shared contract also permits its
+# Todo-less replan re-entry to finish once a Turn has been supplied.
+VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES = frozenset(
+    {
+        *NATIVE_GOAL_RUNTIME_PROFILES,
+        SchedulerRuntimeProfile.CLAUDE_CODE_VISIBLE,
+    }
+)
+
 GUIDED_START_TURN_RUNTIME_PROFILES = frozenset(
     {
         SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT,
diff --git a/loopx/control_plane/turn_driver/host_interaction.ts b/loopx/control_plane/turn_driver/host_interaction.ts
new file mode 100644
index 0000000000..7887d61d19
--- /dev/null
+++ b/loopx/control_plane/turn_driver/host_interaction.ts
@@ -0,0 +1,35 @@
+import type { JsonObject } from "../effect_program.ts";
+import { visionAuthoringContract } from "../goals/vision_checkpoint.ts";
+
+function object(value: unknown): JsonObject | null {
+  return value !== null && typeof value === "object" && !Array.isArray(value)
+    ? value as JsonObject : null;
+}
+
+/** A transport lens, never a replacement for quota admission or vision rules. */
+export function projectMcpInteraction(guard: JsonObject): JsonObject {
+  const contract = object(guard.interaction_contract);
+  if (guard.ok !== true || !contract) return guard;
+  const selected = object(guard.selected_todo);
+  const ownsDelivery = guard.normal_delivery_allowed === true &&
+    typeof selected?.todo_id === "string";
+  const cli = object(contract.cli_channel);
+  // The raw CLI sequence is the implementation of complete_task, not a second
+  // obligation. Replan-only and blocked lanes retain their actual CLI actions.
+  let channel = cli;
+  if (ownsDelivery && cli) {
+    const {settlement_plan: _plan, next_cli_actions: _actions, ...facts} = cli;
+    channel = {...facts, next_cli_actions: [], executor: "mcp_complete_task"};
+  }
+  return {...guard, interaction_contract: {...contract,
+    ...(channel ? {cli_channel: channel} : {}),
+    mcp_channel: {
+      schema_version: "host_mcp_interaction_v0",
+      delivery_executor: "complete_task",
+      delivery_todo_id: ownsDelivery ? selected!.todo_id : null,
+      rule: "For verified Todo delivery, call complete_task INSTEAD OF manual refresh/spend. It owns lifecycle, writeback and accounting. For independent replan, follow the live CLI binding/actions; do not replay an old completion.",
+      vision_authoring: visionAuthoringContract(),
+      checkpoint_recovery_tool: "review_task_vision",
+    },
+  }};
+}
diff --git a/loopx/control_plane/turn_driver/host_todo_completion.ts b/loopx/control_plane/turn_driver/host_todo_completion.ts
index 8831e0a89b..3e8d0dfa5d 100644
--- a/loopx/control_plane/turn_driver/host_todo_completion.ts
+++ b/loopx/control_plane/turn_driver/host_todo_completion.ts
@@ -7,6 +7,8 @@ import {
   type SettlementIdentity,
 } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { normalizeVisionUnchangedReason } from "../goals/vision_checkpoint.ts";
+import { projectMcpInteraction } from "./host_interaction.ts";
 import {
   requireBoolean,
   requireJsonObject,
@@ -17,12 +19,14 @@ import {
 
 export const HOST_TODO_COMPLETION_TRANSACTION_SCHEMA_VERSION =
   "loopx_host_todo_completion_transaction_v0";
+export const HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION =
+  "loopx_host_todo_completion_transaction_v1";
 export const HOST_TODO_COMPLETION_REDUCTION_SCHEMA_VERSION =
   "loopx_host_todo_completion_reduction_v0";
 export const HOST_ADAPTER_SETTLEMENT_SCHEMA_VERSION =
   "host_adapter_todo_settlement_v0";
 
-const PHASES = ["prepare", "finalize", "classify_guard"] as const;
+const PHASES = ["prepare", "finalize", "classify_guard", "vision_refresh", "project_guard"] as const;
 const STEP_KINDS = [
   "guard",
   "lifecycle_completion",
@@ -37,7 +41,7 @@ type HostTodoCompletionStepKind = (typeof STEP_KINDS)[number];
 type HostGuardState = "selected" | "terminal_no_selection" | "invalid";
 
 interface HostTodoCompletionRequest {
-  phase: Exclude<HostTodoCompletionPhase, "classify_guard">;
+  phase: Exclude<HostTodoCompletionPhase, "classify_guard" | "project_guard">;
   goal_id: string;
   agent_id: string;
   todo_id: string;
@@ -47,6 +51,8 @@ interface HostTodoCompletionRequest {
   execution_mode: string;
   completion_args: readonly string[];
   no_follow_up: boolean;
+  vision_path: string | null;
+  vision_unchanged_reason: string | null;
   provider_outcomes: readonly ProviderOutcome[];
 }
 
@@ -72,7 +78,7 @@ interface GuardSelection extends JsonObject {
 function decodePhase(value: JsonObject): HostTodoCompletionPhase {
   requireStringLiteral(
     value.schema_version,
-    [HOST_TODO_COMPLETION_TRANSACTION_SCHEMA_VERSION] as const,
+    [HOST_TODO_COMPLETION_TRANSACTION_SCHEMA_VERSION, HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION] as const,
     "schema_version",
     "schema_version is unsupported",
   );
@@ -116,9 +122,23 @@ function decodeProviderOutcomes(value: unknown): ProviderOutcome[] {
 
 function decodeRequest(
   value: JsonObject,
-  phase: Exclude<HostTodoCompletionPhase, "classify_guard">,
+  phase: Exclude<HostTodoCompletionPhase, "classify_guard" | "project_guard">,
 ): HostTodoCompletionRequest {
   const todoId = typedTodoId(value.todo_id, "todo_id");
+  const optionalText = (field: string): string | null => value[field] == null
+    ? null : requireNonEmptyString(value[field], field);
+  const visionPath = optionalText("vision_path");
+  const unchanged = normalizeVisionUnchangedReason(optionalText("vision_unchanged_reason"));
+  if (visionPath && unchanged) {
+    throw new EffectRuntimeRequestError("choose a vision patch or an unchanged reason, not both");
+  }
+  if ((visionPath || unchanged || phase === "vision_refresh") &&
+      value.schema_version !== HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION) {
+    throw new EffectRuntimeRequestError("host vision authoring requires v1");
+  }
+  if (phase === "vision_refresh" && !visionPath && !unchanged) {
+    throw new EffectRuntimeRequestError("vision refresh requires an authored decision");
+  }
   const request: HostTodoCompletionRequest = {
     phase,
     goal_id: requireNonEmptyString(value.goal_id, "goal_id"),
@@ -145,6 +165,8 @@ function decodeRequest(
       "completion_args",
     ),
     no_follow_up: requireBoolean(value.no_follow_up, "no_follow_up"),
+    vision_path: visionPath,
+    vision_unchanged_reason: unchanged,
     provider_outcomes: [],
   };
   if (phase === "finalize") {
@@ -303,6 +325,22 @@ function spendContinueWhen(identity: JsonObject): JsonObject {
   );
 }
 
+// One command owner for first delivery and checkpoint-only recovery. The CLI's
+// typed refresh recovery validates the original intent, baseline and revision;
+// this projection neither invents a vision nor changes Todo/Goal terminal state.
+function writebackArgs(request: HostTodoCompletionRequest, identity: JsonObject): string[] {
+  return [
+    "refresh-state", "--goal-id", request.goal_id, "--agent-id", request.agent_id,
+    "--classification", "mcp_completed_turn_writeback",
+    "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress",
+    "--todo-id", request.todo_id, "--turn-instance-id", String(identity.turn_instance_id),
+    "--completion-todo-id", request.todo_id, "--completion-turn-key", String(identity.effect_id),
+    "--no-global-sync", "--suppress-external-sinks",
+    ...(request.vision_path ? ["--agent-vision-json", request.vision_path] : []),
+    ...(request.vision_unchanged_reason ? ["--vision-unchanged-reason", request.vision_unchanged_reason] : []),
+  ];
+}
+
 function providerSteps(
   request: HostTodoCompletionRequest,
   identity: JsonObject,
@@ -346,29 +384,7 @@ function providerSteps(
     },
     {
       step_kind: "durable_writeback",
-      args: [
-        "refresh-state",
-        "--goal-id",
-        request.goal_id,
-        "--agent-id",
-        request.agent_id,
-        "--classification",
-        "mcp_completed_turn_writeback",
-        "--delivery-batch-scale",
-        "single_surface",
-        "--delivery-outcome",
-        "outcome_progress",
-        "--todo-id",
-        request.todo_id,
-        "--turn-instance-id",
-        turnId,
-        "--completion-todo-id",
-        request.todo_id,
-        "--completion-turn-key",
-        String(identity.effect_id),
-        "--no-global-sync",
-        "--suppress-external-sinks",
-      ],
+      args: writebackArgs(request, identity),
       legacy_args: null,
       continue_when: writebackContinueWhen(identity),
     },
@@ -627,6 +643,14 @@ function blockedResult(
     settlement,
   };
   if (options.completion) result.completion = options.completion;
+  if (options.completion?.completed === true &&
+      (request.vision_path !== null || request.vision_unchanged_reason !== null)) {
+    result.recovery = {
+      tool: "complete_task", todo_id: request.todo_id,
+      settlement_identity: expectedIdentity(request).payload,
+      instruction: "After correcting the reported error, retry complete_task with the same Todo, evidence and successor/no-follow-up intent. Correct only an uncommitted vision decision; do not repeat work, create another successor, or manually spend. review_task_vision alone does not finish a pending settlement. Committed conflicts and authority failures must not be bypassed.",
+    };
+  }
   return result;
 }
 
@@ -906,11 +930,27 @@ function finalize(request: HostTodoCompletionRequest): JsonObject {
     settlement_identity: identity,
     completion: finalCompletion,
     settlement,
+    ...(request.vision_path !== null || request.vision_unchanged_reason !== null ? {
+      completion_scope: "todo",
+      goal_terminal: {
+        assessed: false,
+        authority: "should_run.interaction_contract",
+        next_tool: "should_run",
+        instruction: "Todo terminal_closeout/no_followup is not Goal completion. Obey the fresh Goal contract, including vision replan. A vision evidence timestamp predating completion does not make that live decision stale. If all scoped acceptance is verified, use the admitted replan path to author no_followup; do not invent filler work or silently drop the obligation.",
+      },
+    } : {}),
   });
 }
 
 export function evaluateHostTodoCompletion(value: JsonObject): JsonObject {
   const phase = decodePhase(value);
+  if (phase === "project_guard") {
+    if (value.schema_version !== HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION) {
+      throw new EffectRuntimeRequestError("host interaction projection requires v1");
+    }
+    return {schema_version: HOST_TODO_COMPLETION_REDUCTION_SCHEMA_VERSION, phase,
+      packet: projectMcpInteraction(requireJsonObject(value.packet, "packet"))};
+  }
   if (phase === "classify_guard") {
     if (typeof value.guard_output !== "string") {
       throw new EffectRuntimeRequestError("guard_output must be a string");
@@ -925,6 +965,12 @@ export function evaluateHostTodoCompletion(value: JsonObject): JsonObject {
   const request = decodeRequest(value, phase);
   if (phase === "finalize") return finalize(request);
   const { payload: identity } = expectedIdentity(request);
+  if (phase === "vision_refresh") {
+    return {
+      schema_version: HOST_TODO_COMPLETION_REDUCTION_SCHEMA_VERSION,
+      phase, identity, args: writebackArgs(request, identity),
+    };
+  }
   const steps = providerSteps(request, identity);
   return reduction(
     "prepare",
diff --git a/loopx/control_plane/work_items/accountable_settlement.py b/loopx/control_plane/work_items/accountable_settlement.py
index 8c407a4f11..a0785f2709 100644
--- a/loopx/control_plane/work_items/accountable_settlement.py
+++ b/loopx/control_plane/work_items/accountable_settlement.py
@@ -7,8 +7,8 @@
     build_turn_scoped_cli_settlement_plan,
 )
 from ..scheduler.execution_context import (
-    NATIVE_GOAL_RUNTIME_PROFILES,
     SchedulerRuntimeProfile,
+    VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES,
 )
 
 
@@ -38,7 +38,7 @@ def build_accountable_work_item_settlement_plan(
             turn_instance_id_ref=normalized_turn_instance_id,
             delivery_boundary=delivery_boundary,
         )
-    if runtime_profile in NATIVE_GOAL_RUNTIME_PROFILES:
+    if runtime_profile in VISIBLE_GOAL_SETTLEMENT_RUNTIME_PROFILES:
         normalized_turn_instance_id = normalize_turn_instance_id(turn_instance_id)
         if normalized_turn_instance_id is None:
             return None
diff --git a/loopx/control_plane/work_items/autonomous_replan_obligation.py b/loopx/control_plane/work_items/autonomous_replan_obligation.py
index bd7a923fa2..2b19df8f59 100644
--- a/loopx/control_plane/work_items/autonomous_replan_obligation.py
+++ b/loopx/control_plane/work_items/autonomous_replan_obligation.py
@@ -147,13 +147,27 @@ def build_autonomous_replan_cli_actions(
             "execute replan_action_packet.writeback_contract.successor_command",
             "on host_action=end_current_heartbeat: stop",
         ]
-    typed_progress_args = (
-        "--progress-result-class "
-        "<advanced|blocked|exploration_exhausted|no_followup> "
-        "--progress-surface-id <surface-id> "
-        "--progress-hypothesis-id <hypothesis-id> "
-        "--progress-probe-kind <probe-kind> "
-        "--progress-evidence-id <evidence-id>"
+    raw_obligation = payload.get("autonomous_replan_obligation")
+    obligation: Mapping[str, Any] = (
+        raw_obligation if isinstance(raw_obligation, Mapping) else {}
+    )
+    vision_successor_required = any(
+        isinstance(trigger, Mapping)
+        and trigger.get("kind") == "vision_successor_required"
+        for trigger in obligation.get("triggers") or []
+    )
+    semantic_delta_args = (
+        "--agent-vision-json "
+        "'<path-to-evidence-linked-goal-vision-replan-contract-v0.json>'"
+        if vision_successor_required
+        else (
+            "--progress-result-class "
+            "<advanced|blocked|exploration_exhausted|no_followup> "
+            "--progress-surface-id <surface-id> "
+            "--progress-hypothesis-id <hypothesis-id> "
+            "--progress-probe-kind <probe-kind> "
+            "--progress-evidence-id <evidence-id>"
+        )
     )
     delivery_args = (
         "--delivery-batch-scale single_surface "
@@ -166,7 +180,7 @@ def build_autonomous_replan_cli_actions(
         f"{cli_prefix} --format json refresh-state --goal-id {goal_id} "
         "--progress-scope agent_lane "
         "--classification bounded_replan_progress "
-        f"{delivery_args}{typed_progress_args}"
+        f"{delivery_args}{semantic_delta_args}"
         f"{settlement_args}{scoped_cli_args}"
     )
     if not settlement_chain_ready:
diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py
index a661a1d337..79d38377ff 100644
--- a/loopx/control_plane/work_items/interaction_contract.py
+++ b/loopx/control_plane/work_items/interaction_contract.py
@@ -16,7 +16,7 @@
 )
 from ..quota.spend_sources import (
     build_quota_spend_action,
-    visible_goal_turn_reentry_action,
+    host_goal_turn_reentry_action,
 )
 from ..scheduler.execution_context import (
     SchedulerExecutionContextResolution,
@@ -713,7 +713,7 @@ def interaction_next_cli_actions(
         if scheduler_args
         else "rerun the typed quota_guard from the current host packet"
     )
-    if turn_reentry_action := visible_goal_turn_reentry_action(
+    if turn_reentry_action := host_goal_turn_reentry_action(
         payload, settlement_plan, scheduler_execution_context, turn_instance_id, typed_quota_guard
     ):
         return [turn_reentry_action]
diff --git a/loopx/goal_mode_mcp.py b/loopx/goal_mode_mcp.py
index ef17890127..98d21327b7 100644
--- a/loopx/goal_mode_mcp.py
+++ b/loopx/goal_mode_mcp.py
@@ -18,6 +18,9 @@ class Strict:  # type: ignore[no-redef]
 
 from .control_plane.host_adapter_settlement import (
     HostTodoSettlementRequest,
+    host_vision_request,
+    project_host_interaction,
+    refresh_host_todo_vision,
     settle_host_todo_completion,
 )
 
@@ -143,7 +146,7 @@ def should_run(self) -> str:
         if not goal_id:
             return self.no_goal_message()
         args, legacy_args = self.should_run_args(goal_id, self.bound_agent_id())
-        return self.run_cli(args, legacy_args=legacy_args)
+        return project_host_interaction(self.run_cli(args, legacy_args=legacy_args))
 
     def list_todos(self) -> str:
         return self.should_run()
@@ -189,6 +192,8 @@ def complete_task(
         task_lease_expected_version: ExpectedTaskLeaseVersion = None,
         no_follow_up: bool = False,
         successor_todo_ids: list[str] | None = None,
+        agent_vision: dict[str, Any] | None = None,
+        vision_unchanged_reason: str = "",
     ) -> str:
         goal_id, _ = self.context()
         if not goal_id:
@@ -237,20 +242,39 @@ def complete_task(
             ]
         if no_follow_up:
             args.append("--no-follow-up")
-        return settle_host_todo_completion(
-            HostTodoSettlementRequest(
-                goal_id=goal_id,
-                agent_id=agent_id,
-                todo_id=todo_id,
-                runtime_profile=self.config.runtime_profile,
-                legacy_host_surface=self.config.legacy_host_surface,
-                scheduler_owner=self.config.scheduler_owner,
-                execution_mode=self.config.execution_mode,
-                completion_args=tuple(args),
-                no_follow_up=no_follow_up,
-            ),
-            run_cli=self.run_cli,
+        request = HostTodoSettlementRequest(
+            goal_id=goal_id,
+            agent_id=agent_id,
+            todo_id=todo_id,
+            runtime_profile=self.config.runtime_profile,
+            legacy_host_surface=self.config.legacy_host_surface,
+            scheduler_owner=self.config.scheduler_owner,
+            execution_mode=self.config.execution_mode,
+            completion_args=tuple(args),
+            no_follow_up=no_follow_up,
+        )
+        with host_vision_request(request, agent_vision, vision_unchanged_reason) as authored:
+            return settle_host_todo_completion(authored, run_cli=self.run_cli)
+
+    def review_task_vision(
+        self, todo_id: str, agent_id: str, agent_vision: dict[str, Any] | None = None,
+        vision_unchanged_reason: str = "",
+    ) -> str:
+        goal_id, _ = self.context()
+        if not goal_id:
+            return self.no_goal_message()
+        identity_error = self._identity_error(agent_id)
+        if identity_error:
+            return identity_error
+        request = HostTodoSettlementRequest(
+            goal_id=goal_id, agent_id=agent_id, todo_id=todo_id,
+            runtime_profile=self.config.runtime_profile,
+            legacy_host_surface=self.config.legacy_host_surface,
+            scheduler_owner=self.config.scheduler_owner, execution_mode=self.config.execution_mode,
+            completion_args=(),
         )
+        with host_vision_request(request, agent_vision, vision_unchanged_reason) as authored:
+            return refresh_host_todo_vision(authored, run_cli=self.run_cli)
 
 
 def create_fastmcp_server(
@@ -289,6 +313,21 @@ def claim_task(todo_id: str, agent_id: str) -> str:
         """Claim one todo as the bound agent."""
         return control.claim_task(todo_id, agent_id)
 
+    @server.tool()
+    def review_task_vision(
+        todo_id: str, agent_id: str, agent_vision: dict[str, Any] | None = None,
+        vision_unchanged_reason: str = "",
+    ) -> str:
+        """Supply a missing vision decision for a previously completed MCP Todo.
+        Uses its original Turn, never repeats work or spends again. agent_vision is
+        a goal_vision_replan_contract_v0 packet with state and vision_patch fields.
+        Compare Goal acceptance with evidence; vision_closed closes a stage and
+        still requires a successor, no_followup asserts no remaining scoped work.
+        An unchanged reason requires an existing valid vision. Recheck should_run;
+        checkpoint success alone does not certify Goal completion or clear gates.
+        """
+        return control.review_task_vision(todo_id, agent_id, agent_vision, vision_unchanged_reason)
+
     @server.tool()
     def complete_task(
         todo_id: str,
@@ -299,10 +338,19 @@ def complete_task(
         task_lease_expected_version: ExpectedTaskLeaseVersion = None,
         no_follow_up: bool = False,
         successor_todo_ids: list[str] | None = None,
+        agent_vision: dict[str, Any] | None = None,
+        vision_unchanged_reason: str = "",
     ) -> str:
         """Complete verified work and settle once. Link existing planned successors
         with successor_todo_ids; next_agent_todo creates a NEW Todo, not an id link.
-        Use no_follow_up only for terminal intent. Do not duplicate existing work.
+        no_follow_up closes this Todo's continuation, NOT the Goal's vision.
+        Do not duplicate existing work; only the fresh should_run contract can
+        establish Goal terminal state, regardless of the Todo closeout receipt.
+        Include an authored agent_vision (goal_vision_replan_contract_v0 with state
+        and vision_patch), or an unchanged reason backed by an existing vision.
+        Omission keeps a required checkpoint open; repair with review_task_vision.
+        If settlement failed, correct uncommitted input and retry complete_task
+        with the same completion intent; checkpoint-only recovery cannot spend.
         """
         return control.complete_task(
             todo_id,
@@ -313,6 +361,8 @@ def complete_task(
             task_lease_expected_version=task_lease_expected_version,
             no_follow_up=no_follow_up,
             successor_todo_ids=successor_todo_ids,
+            agent_vision=agent_vision,
+            vision_unchanged_reason=vision_unchanged_reason,
         )
 
     return server, control
diff --git a/loopx/kunluncode_goal_mode/guards.py b/loopx/kunluncode_goal_mode/guards.py
index 15b8fd367e..7779fe96df 100644
--- a/loopx/kunluncode_goal_mode/guards.py
+++ b/loopx/kunluncode_goal_mode/guards.py
@@ -42,6 +42,8 @@ def blocked_complete(
         task_lease_expected_version: int | None = None,
         no_follow_up: bool = False,
         successor_todo_ids: list[str] | None = None,
+        agent_vision: dict | None = None,
+        vision_unchanged_reason: str = "",
     ) -> str:
         del (
             next_agent_todo,
@@ -49,12 +51,19 @@ def blocked_complete(
             task_lease_expected_version,
             no_follow_up,
             successor_todo_ids,
+            agent_vision,
+            vision_unchanged_reason,
         )
         return _native_controller_rejection("complete_task")
 
     control_plane.claim_task = blocked_claim
     control_plane.complete_task = blocked_complete
 
+    def blocked_vision(*_args: Any, **_kwargs: Any) -> str:
+        return _native_controller_rejection("review_task_vision")
+
+    control_plane.review_task_vision = blocked_vision
+
 
 def native_controller_cli_write_block(arguments: Any) -> dict[str, Any] | None:
     """Block direct model CLI writes to the Goal owned by the outer controller."""
diff --git a/scripts/qualify-claude-goal-release.py b/scripts/qualify-claude-goal-release.py
index 9cac3c7ee5..0f0d475a7d 100644
--- a/scripts/qualify-claude-goal-release.py
+++ b/scripts/qualify-claude-goal-release.py
@@ -25,6 +25,7 @@
 )
 
 ARK_ANTHROPIC_BASE = "https://ark.cn-beijing.volces.com/api/compatible"
+MANIFEST_ASSETS = {"assets/alpha.txt": b"alpha\n", "assets/nested/beta.bin": bytes(range(256))}
 
 
 def run_host(command: list[str], *, cwd: Path, env: dict, timeout: float) -> str:
@@ -80,7 +81,7 @@ def host_environment(root: Path, launcher: Path) -> dict[str, str]:
     return env
 
 
-def verify_mcp_completions(events: list[dict]) -> None:
+def verify_mcp_completions(events: list[dict], expected_todos: set[str] | None = None) -> None:
     """A tool invocation is not evidence that the MCP transaction succeeded."""
     pending: dict[str, str] = {}
     completed: set[str] = set()
@@ -105,13 +106,12 @@ def verify_mcp_completions(events: list[dict]) -> None:
             assert payload.get("todo_id") == todo_id and payload.get("completed") is True
             assert (payload.get("settlement") or {}).get("ok") is True
             completed.add(todo_id)
-    assert completed == shared.TODOS, "mcp_delivery_transactions_not_completed"
+    assert completed == (shared.TODOS if expected_todos is None else expected_todos), "mcp_delivery_transactions_not_completed"
 
 
-def qualify(root: Path, claude: str, timeout: int) -> dict:
+def exercise_host(root: Path, project: Path, launcher: Path, claude: str, timeout: int) -> list[dict]:
     from loopx.claude_goal_mode.scripts.goalmode_cmd import write_loop_md
 
-    project, runtime, launcher = shared.setup(root)
     write_loop_md(project, shared.GOAL, shared.AGENT)
     config = root / "mcp.json"
     config.write_text(json.dumps({"mcpServers": {"loopx": {
@@ -127,7 +127,8 @@ def qualify(root: Path, claude: str, timeout: int) -> dict:
         "--output-format", "stream-json", "--verbose", "-p",
         "Read .claude/loop.md and follow this project's active LoopX work contract. "
         "Read TASK.md for acceptance. Use the bound LoopX MCP tools; preserve the "
-        "isolated project binding. Do not create timers in this headless qualification.",
+        "isolated project binding. Do not create timers in this headless qualification. "
+        "Use product contracts, not qualification scripts or their external acceptance oracles.",
     ]
     # This executes the actual per-iteration adapter, not Claude's interactive
     # /loop timer. Never report headless delivery as scheduler qualification.
@@ -135,6 +136,12 @@ def qualify(root: Path, claude: str, timeout: int) -> dict:
     events = [json.loads(line) for line in output.splitlines() if line.strip()]
     results = [e for e in events if e.get("type") == "result"]
     assert len(results) == 1 and results[0].get("is_error") is False, "claude_turn_failed"
+    return events
+
+
+def qualify(root: Path, claude: str, timeout: int) -> dict:
+    project, runtime, launcher = shared.setup(root)
+    events = exercise_host(root, project, launcher, claude, timeout)
     calls = [block.get("name") for event in events if event.get("type") == "assistant"
              for block in (event.get("message") or {}).get("content", [])
              if block.get("type") == "tool_use"]
@@ -146,11 +153,130 @@ def qualify(root: Path, claude: str, timeout: int) -> dict:
             **shared.verify_delivery(project, runtime, launcher, "claude_code")}
 
 
+def setup_replan(root: Path) -> tuple[Path, Path, Path]:
+    """A legitimately finished inventory stage, not fabricated Goal acceptance."""
+    from loopx.goal_mode_mcp import GoalModeMCPConfig, GoalModeMCPControlPlane
+
+    project, runtime, launcher = shared.setup(root)
+    assets = project / "assets"
+    for name, content in MANIFEST_ASSETS.items():
+        path = project / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+    (project / "manifest.json").write_text(json.dumps(["assets/alpha.txt", "assets/nested/beta.bin"]))
+    shutil.copyfile(REPO / "tests/fixtures/host_vision_replan/TASK.md", project / "TASK.md")
+    (project / "ACTIVE_GOAL_STATE.md").write_text(
+        "---\nstatus: active\n---\n\n## Objective\n\nDeliver and validate TASK.md.\n\n"
+        "## Next Action\n\nVerify artifact integrity, beyond the completed filename inventory.\n\n"
+        "## User Todo\n\n## Agent Todo\n\n"
+        "- [ ] [P1] Inventory the regular asset filenames in manifest.json.\n"
+        "  <!-- loopx:todo todo_id=todo_inventory status=open task_class=advancement_task "
+        "action_kind=implement claimed_by=worker-a task_repository=git:example.com/ledger -->\n",
+    )
+    assert json.loads((project / "manifest.json").read_text()) == sorted(
+        str(path.relative_to(project)) for path in assets.rglob("*") if path.is_file())
+    control = GoalModeMCPControlPlane(GoalModeMCPConfig(server_name="fixture",
+        runtime_profile="claude_code", legacy_host_surface="claude_code"),
+        lambda: {"goal_id": shared.GOAL, "agent_id": shared.AGENT})
+    control.command_prefix = lambda: [str(launcher)]
+    previous = Path.cwd()
+    try:
+        os.chdir(project)
+        completed = json.loads(control.complete_task("todo_inventory", shared.AGENT,
+            "Filename inventory matches both regular asset files; integrity delivery remains unverified.",
+            no_follow_up=True, agent_vision={
+                "schema_version": "goal_vision_replan_contract_v0", "state": "vision_closed",
+                "vision_patch": {
+                    "vision_summary": "Inventory regular asset filenames.",
+                    "acceptance_summary": "Both filenames match the actual asset directory.",
+                    "last_patch_summary": "Filename inventory stage verified; full integrity acceptance remains.",
+                },
+            }))
+    finally:
+        os.chdir(previous)
+    assert completed["ok"] is True, "inventory_stage_not_settled"
+    return project, runtime, launcher
+
+
+def verify_manifest_delivery(project: Path) -> None:
+    """Independent acceptance, including destructive-input *copies* and no self-repair."""
+    import hashlib
+
+    observed = {str(path.relative_to(project)): path.read_bytes()
+        for path in (project / "assets").rglob("*") if path.is_file()}
+    assert observed == MANIFEST_ASSETS, "manifest_inputs_modified"
+    expected = {name: {"size": len(content), "sha256": hashlib.sha256(content).hexdigest()}
+        for name, content in MANIFEST_ASSETS.items()}
+    assert json.loads((project / "manifest.json").read_text()) == expected, "manifest_acceptance_failed"
+    assert (project / "README.md").is_file(), "manifest_readme_missing"
+    for mutation in ("none", "changed", "missing", "additional"):
+        with tempfile.TemporaryDirectory(prefix="loopx-manifest-oracle-") as raw:
+            copy = Path(raw)
+            shutil.copytree(project / "assets", copy / "assets")
+            for name in ("manifest.json", "verify_manifest.py"):
+                shutil.copyfile(project / name, copy / name)
+            if mutation == "changed":
+                (copy / "assets/alpha.txt").write_text("corrupted")
+            elif mutation == "missing":
+                (copy / "assets/alpha.txt").unlink()
+            elif mutation == "additional":
+                (copy / "assets/additional.txt").write_text("extra")
+            before = {str(path.relative_to(copy)): path.read_bytes() for path in copy.rglob("*") if path.is_file()}
+            result = subprocess.run([sys.executable, "-B", "verify_manifest.py"], cwd=copy,
+                                    capture_output=True, timeout=30)
+            assert (result.returncode == 0) is (mutation == "none"), "manifest_verifier_unsound"
+            after = {str(path.relative_to(copy)): path.read_bytes() for path in copy.rglob("*") if path.is_file()}
+            assert before == after, "manifest_verifier_mutated_evidence"
+
+
+def qualify_replan(root: Path, claude: str, timeout: int) -> dict:
+    project, runtime, launcher = setup_replan(root)
+    before = shared.cli(launcher, "quota", "should-run", "--goal-id", shared.GOAL,
+        "--agent-id", shared.AGENT, "--runtime-profile", "claude_code")
+    assert before["should_run"] is True, "finished_stage_must_not_hide_goal_gap"
+    events = exercise_host(root, project, launcher, claude, timeout)
+    return verify_replan_delivery(project, runtime, launcher, events)
+
+
+def verify_replan_delivery(project: Path, runtime: Path, launcher: Path, events: list[dict]) -> dict:
+    """Prove gap -> delivered successor -> scoped vision decision, not a magic label."""
+    verify_manifest_delivery(project)
+    assert (project / "TASK.md").read_bytes() == (REPO / "tests/fixtures/host_vision_replan/TASK.md").read_bytes()
+    todos = shared.cli(launcher, "todo", "list", "--goal-id", shared.GOAL, "--role", "agent")["todos"]
+    successors = [t for t in todos if t["todo_id"] != "todo_inventory"]
+    assert successors and all(t["status"] == "done" for t in todos), "replan_must_deliver_concrete_successor"
+    assert all(t["task_class"] == "advancement_task" for t in successors)
+    verify_mcp_completions(events, {t["todo_id"] for t in successors})
+    rows = [json.loads(line) for line in (runtime / "goals" / shared.GOAL / "runs/index.jsonl").read_text().splitlines()]
+    visions = [r["agent_vision"] for r in rows if isinstance(r.get("agent_vision"), dict)
+               and r.get("todo_id") != "todo_inventory"]
+    # Replan is an operation, not a mandatory final path disposition. After
+    # actual successor delivery, no_followup + stop is valid scoped closure.
+    # It cannot shortcut this oracle's independent artifact/successor checks.
+    assert any(v.get("path_delta", {}).get("outcome") == "replan" or
+               (v.get("state") == "no_followup" and v.get("path_delta", {}).get("outcome") == "stop")
+               for v in visions), "no_successor_vision_decision"
+    spent = [r for r in rows if r.get("classification") == "quota_slot_spent"]
+    effects = [r["settlement_identity"]["effect_id"] for r in spent]
+    assert len(effects) == len(set(effects)), "duplicate_replan_spend"
+    for todo in todos:
+        assert sum(r.get("todo_id") == todo["todo_id"] for r in spent) == 1, "todo_completion_spend_not_exactly_once"
+    shared.verify_spend_receipts(runtime, spent)
+    final = shared.cli(launcher, "quota", "should-run", "--goal-id", shared.GOAL,
+        "--agent-id", shared.AGENT, "--runtime-profile", "claude_code")
+    assert final["interaction_contract"]["mode"] == "terminal_no_followup"
+    return {"status": "passed", "host": "claude_code", "model_executed": True,
+        "model": DOUBAO_SEED_EVOLVING_MODEL, "scenario": "replan", "successor_count": len(successors),
+        "settled_spends": len(spent), "host_events": len(events), "independent_acceptance": "passed",
+        "scheduler_qualification": "not_run_headless"}
+
+
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--release-live", action="store_true")
     parser.add_argument("--claude-bin", default="claude")
     parser.add_argument("--timeout-seconds", type=int, default=1200)
+    parser.add_argument("--scenario", choices=("delivery", "replan"), default="delivery")
     args = parser.parse_args(argv)
     if args.timeout_seconds <= 0:
         parser.error("timeout must be positive")
@@ -163,7 +289,8 @@ def main(argv: list[str] | None = None) -> int:
                 result = {"status": "skipped", "reason": reason, "model_executed": False}
             else:
                 with tempfile.TemporaryDirectory(prefix="loopx-claude-release-") as raw:
-                    result = qualify(Path(raw), args.claude_bin, args.timeout_seconds)
+                    qualification = qualify_replan if args.scenario == "replan" else qualify
+                    result = qualification(Path(raw), args.claude_bin, args.timeout_seconds)
         except Exception as exc:
             result = {"status": "failed", "error_kind": type(exc).__name__}
     print(json.dumps(result, sort_keys=True))
diff --git a/scripts/qualify-native-goal-release.py b/scripts/qualify-native-goal-release.py
index 7dae7ae430..b729496ee4 100644
--- a/scripts/qualify-native-goal-release.py
+++ b/scripts/qualify-native-goal-release.py
@@ -144,9 +144,6 @@ def cli(launcher: Path, *args: str) -> dict:
 
 
 def verify_settlement(runtime: Path, todos: list[dict]) -> int:
-    from loopx.control_plane.effect_program import SettlementStepKind
-    from loopx.control_plane.quota.settlement import read_heartbeat_settlement
-
     assert len(todos) == len(TODOS) and {t["todo_id"] for t in todos} == TODOS
     assert all(t.get("status") == "done" for t in todos)
     rows = [json.loads(line) for line in
@@ -156,6 +153,15 @@ def verify_settlement(runtime: Path, todos: list[dict]) -> int:
     assert all(isinstance(r.get("settlement_identity"), dict) for r in spends), "unbound_spend"
     identities = [r["settlement_identity"]["effect_id"] for r in spends]
     assert len(identities) == len(set(identities)), "duplicate_spend"
+    verify_spend_receipts(runtime, spends)
+    return len(spends)
+
+
+def verify_spend_receipts(runtime: Path, spends: list[dict]) -> None:
+    """Read durable receipts for Todo delivery and independent replan Turns."""
+    from loopx.control_plane.effect_program import SettlementStepKind
+    from loopx.control_plane.quota.settlement import read_heartbeat_settlement
+
     for row in spends:
         readback = read_heartbeat_settlement(
             runtime, goal_id=GOAL, agent_id=AGENT, todo_id=row.get("todo_id"),
@@ -169,7 +175,6 @@ def verify_settlement(runtime: Path, todos: list[dict]) -> int:
             SettlementStepKind.QUOTA_SPEND,
         }, "missing_settlement_receipts"
         assert readback.writeback_run is not None and readback.spend_run is not None
-    return len(spends)
 
 
 def verify_delivery(project: Path, runtime: Path, launcher: Path, profile: str) -> dict:
diff --git a/tests/control_plane/test_quota_settlement.py b/tests/control_plane/test_quota_settlement.py
index c69bb82df3..75d4a4a1f5 100644
--- a/tests/control_plane/test_quota_settlement.py
+++ b/tests/control_plane/test_quota_settlement.py
@@ -732,6 +732,46 @@ def test_turn_bound_native_goal_preserves_visible_goal_settlement(profile) -> No
         assert f"--turn-instance-id {turn_instance_id}" in command
 
 
+def test_claude_visible_goal_reenters_before_exposing_bound_settlement() -> None:
+    payload = {
+        "goal_id": GOAL_ID,
+        "agent_identity": {"agent_id": AGENT_ID},
+        "selected_todo": {"todo_id": TODO_ID},
+    }
+    context = scheduler_execution_context_for_runtime_profile(
+        SchedulerRuntimeProfile.CLAUDE_CODE_VISIBLE
+    )
+
+    unbound = interaction_next_cli_actions(
+        payload,
+        mode="bounded_delivery",
+        scheduler_execution_context=context,
+    )
+
+    assert len(unbound) == 1
+    assert unbound[0].startswith("loopx --format json quota should-run")
+    assert "--runtime-profile claude_code" in unbound[0]
+    assert "--turn-instance-id" in unbound[0]
+    assert "refresh-state" not in unbound[0]
+    assert "spend-slot" not in unbound[0]
+
+    turn_instance_id = "claude-visible-goal-turn-1"
+    bound = interaction_next_cli_actions(
+        payload,
+        mode="bounded_delivery",
+        scheduler_execution_context=context,
+        turn_instance_id=turn_instance_id,
+    )
+
+    assert len(bound) == 2
+    assert bound[0].startswith("loopx refresh-state")
+    assert bound[1].startswith("loopx quota spend-slot")
+    assert "--source visible-goal" in bound[1]
+    for command in bound:
+        assert f"--todo-id {TODO_ID}" in command
+        assert f"--turn-instance-id {turn_instance_id}" in command
+
+
 def test_codex_app_external_observation_settles_only_substantive_writeback() -> None:
     todo_id = "todo_external_observation"
     actions = interaction_next_cli_actions(
diff --git a/tests/control_plane_ts/host_interaction.test.ts b/tests/control_plane_ts/host_interaction.test.ts
new file mode 100644
index 0000000000..6d068b983c
--- /dev/null
+++ b/tests/control_plane_ts/host_interaction.test.ts
@@ -0,0 +1,40 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { projectMcpInteraction } from "../../loopx/control_plane/turn_driver/host_interaction.ts";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+
+function guard(overrides: JsonObject = {}): JsonObject {
+  return {ok: true, normal_delivery_allowed: true, selected_todo: {todo_id: "todo_example"},
+    interaction_contract: {mode: "bounded_delivery", agent_channel: {must_attempt: true},
+      cli_channel: {next_cli_actions: ["refresh-state", "spend-slot"],
+        settlement_plan: {steps: ["validation", "durable_writeback", "quota_spend"]},
+        required_reads: ["acceptance"], delivery_workspace_causality: {requirement: "required"}}},
+    ...overrides};
+}
+
+test("MCP delivery replaces duplicate CLI procedure, preserving admission and scope facts", () => {
+  const source = guard();
+  const before = structuredClone(source);
+  const projected = projectMcpInteraction(source);
+  const contract = projected.interaction_contract as JsonObject;
+  const cli = contract.cli_channel as JsonObject;
+  assert.deepEqual(cli.next_cli_actions, []);
+  assert.equal(cli.settlement_plan, undefined);
+  assert.deepEqual(cli.delivery_workspace_causality, {requirement: "required"});
+  assert.deepEqual(cli.required_reads, ["acceptance"]);
+  assert.deepEqual(contract.agent_channel, {must_attempt: true});
+  assert.equal(projected.normal_delivery_allowed, true);
+  assert.equal((contract.mcp_channel as JsonObject).delivery_todo_id, "todo_example");
+  assert.deepEqual(source, before);
+});
+
+test("replan, blocked and unavailable guards never lose their actual actions", () => {
+  for (const overrides of [{normal_delivery_allowed: false}, {selected_todo: null}]) {
+    const source = guard(overrides);
+    assert.deepEqual((projectMcpInteraction(source).interaction_contract as JsonObject).cli_channel,
+      (source.interaction_contract as JsonObject).cli_channel);
+  }
+  const failed = guard({ok: false});
+  assert.deepEqual(projectMcpInteraction(failed), failed);
+  assert.deepEqual(projectMcpInteraction({ok: true}), {ok: true});
+});
diff --git a/tests/control_plane_ts/host_todo_completion.test.ts b/tests/control_plane_ts/host_todo_completion.test.ts
index 950d93cb7f..6074acd378 100644
--- a/tests/control_plane_ts/host_todo_completion.test.ts
+++ b/tests/control_plane_ts/host_todo_completion.test.ts
@@ -8,10 +8,64 @@ import {
   HOST_ADAPTER_SETTLEMENT_SCHEMA_VERSION,
   HOST_TODO_COMPLETION_REDUCTION_SCHEMA_VERSION,
   HOST_TODO_COMPLETION_TRANSACTION_SCHEMA_VERSION,
+  HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION,
 } from "../../loopx/control_plane/turn_driver/host_todo_completion.ts";
 
 const todoId = "todo_abc123";
 
+test("vision refresh shares the original delivery command and identity without a spend", () => {
+  const authored = {schema_version: HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION,
+    vision_path: "fixture-vision.json"};
+  const first = prepare(authored);
+  const recovery = evaluateHostTodoCompletion(request("prepare", {...authored, phase: "vision_refresh"}));
+  const steps = (first.provider_effect as {steps: {step_kind: string; args: string[]}[]}).steps;
+  assert.deepEqual(recovery.args, steps.find(step => step.step_kind === "durable_writeback")!.args);
+  assert.deepEqual(recovery.identity, first.identity);
+  assert.equal(recovery.provider_effect, undefined);
+  assert.equal((recovery.args as string[]).includes("spend-slot"), false);
+  assert.equal((recovery.args as string[]).includes("--next-action"), false);
+});
+
+test("vision decisions require v1 and cannot combine patch with unchanged", () => {
+  assert.throws(() => prepare({vision_path: "vision.json"}), /requires v1/);
+  assert.throws(() => prepare({schema_version: HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION,
+    vision_path: "vision.json", vision_unchanged_reason: "unchanged"}), /not both/);
+  assert.throws(() => evaluateHostTodoCompletion(request("prepare", {
+    schema_version: HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION, phase: "vision_refresh",
+  })), /authored decision/);
+  const base = prepare();
+  assert.deepEqual(prepare({schema_version: HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION}), base);
+});
+
+test("unchanged vision authoring is validated before completion or recovery effects", () => {
+  for (const phase of ["prepare", "vision_refresh"] as const) {
+    const authored = {
+      schema_version: HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION,
+      phase,
+      vision_unchanged_reason: "x".repeat(241),
+    };
+    assert.throws(() => evaluateHostTodoCompletion(request("prepare", authored)),
+      /vision_unchanged_reason exceeds 240 chars/);
+    const reduced = evaluateHostTodoCompletion(request("prepare", {
+      ...authored, vision_unchanged_reason: `  ${"x".repeat(240)}  `,
+    }));
+    const args = phase === "vision_refresh" ? reduced.args :
+      (reduced.provider_effect as {steps: {step_kind: string; args: string[]}[]})
+        .steps.find(step => step.step_kind === "durable_writeback")!.args;
+    const command = args as string[];
+    assert.equal(command[command.indexOf("--vision-unchanged-reason") + 1], "x".repeat(240));
+  }
+});
+
+test("vision-aware Todo closeout never certifies Goal termination", () => {
+  const reduced = finalize(providerOutcomes(identityFrom(prepare())), {schema_version: HOST_TODO_VISION_TRANSACTION_SCHEMA_VERSION,
+    vision_path: "vision.json"});
+  assert.equal((reduced.result as Record<string, unknown>).completion_scope, "todo");
+  const terminal = (reduced.result as Record<string, unknown>).goal_terminal as Record<string, unknown>;
+  assert.equal(terminal.assessed, false);
+  assert.equal(terminal.next_tool, "should_run");
+});
+
 function request(
   phase: "prepare" | "finalize" | "classify_guard",
   overrides: Record<string, unknown> = {},
diff --git a/tests/fixtures/host_vision_replan/TASK.md b/tests/fixtures/host_vision_replan/TASK.md
new file mode 100644
index 0000000000..4980a2800e
--- /dev/null
+++ b/tests/fixtures/host_vision_replan/TASK.md
@@ -0,0 +1,17 @@
+# Verified local artifact manifest
+
+The first stage inventoried the files under `assets/`; `manifest.json` currently
+contains only their names. The finished deliverable must also prove integrity.
+
+Replace the inventory with a JSON object mapping each project-relative file path to an
+object containing `size` (bytes) and `sha256` (lowercase hexadecimal). Include
+every regular file recursively under `assets/`, no others. Add
+`verify_manifest.py`, using only the Python standard library, which validates
+the current assets against that manifest. It must exit zero for an exact match
+and nonzero for a changed, missing, or additional file. Verification is read-only:
+never regenerate the manifest or change assets to conceal a mismatch. Include
+tests and a concise README explaining generation and verification.
+
+Do not alter this acceptance specification. Work is local to this disposable
+project; no network delivery, deployment, package installation or publication
+is requested.
diff --git a/tests/test_claude_goal_release_qualification.py b/tests/test_claude_goal_release_qualification.py
index 35c54e8951..9ead9fd017 100644
--- a/tests/test_claude_goal_release_qualification.py
+++ b/tests/test_claude_goal_release_qualification.py
@@ -4,6 +4,7 @@
 import importlib.util
 import json
 import os
+import shlex
 import subprocess
 from pathlib import Path
 import sys
@@ -23,6 +24,8 @@ def forbidden(*_):
     monkeypatch.setattr(runner, "prerequisite_failure", forbidden)
     assert runner.main([]) == 0
     assert json.loads(capsys.readouterr().out)["model_executed"] is False
+    assert runner.main(["--scenario", "replan"]) == 0
+    assert json.loads(capsys.readouterr().out)["model_executed"] is False
 
 
 def test_missing_environment_skips_but_attempted_failure_fails(monkeypatch, capsys):
@@ -105,9 +108,10 @@ def test_host_nonzero_is_not_reported_as_a_successful_model_turn(tmp_path):
 
 
 @pytest.mark.parametrize("failed", [False, True])
-def test_mcp_oracle_requires_successful_transactions_not_only_invocations(failed):
+@pytest.mark.parametrize("expected_todos", [None, {"todo_successor"}])
+def test_mcp_oracle_requires_successful_transactions_not_only_invocations(failed, expected_todos):
     events = []
-    for todo in sorted(runner.shared.TODOS):
+    for todo in sorted(runner.shared.TODOS if expected_todos is None else expected_todos):
         events.append({"message": {"content": [{"type": "tool_use", "id": todo,
             "name": "mcp__loopx__complete_task", "input": {"todo_id": todo}}]}})
         result = {"ok": not failed, "completed": True, "todo_id": todo,
@@ -116,9 +120,9 @@ def test_mcp_oracle_requires_successful_transactions_not_only_invocations(failed
             "content": json.dumps({"result": json.dumps(result)})}]}})
     if failed:
         with pytest.raises(AssertionError, match="mcp_delivery_transactions_not_completed"):
-            runner.verify_mcp_completions(events)
+            runner.verify_mcp_completions(events, expected_todos)
     else:
-        runner.verify_mcp_completions(events)
+        runner.verify_mcp_completions(events, expected_todos)
 
 
 def test_real_claude_stdio_mcp_binding_and_identity_gate(tmp_path):
@@ -150,15 +154,195 @@ async def exercise():
                 assert "call the bound LoopX `host_prompt`" not in current["task_body"]
                 complete = next(t for t in tools.tools if t.name == "complete_task")
                 assert "successor_todo_ids" in complete.inputSchema["properties"]
+                assert "agent_vision" in complete.inputSchema["properties"]
+                assert "review_task_vision" in {t.name for t in tools.tools}
                 guard = await session.call_tool("should_run", {})
                 payload = json.loads(guard.content[0].text)
                 assert payload["ok"] is True and payload["selected_todo"]["todo_id"] == "todo_reducer"
+                contract = payload["interaction_contract"]
+                assert contract["cli_channel"]["next_cli_actions"] == []
+                assert contract["mcp_channel"]["delivery_executor"] == "complete_task"
+                assert contract["mcp_channel"]["vision_authoring"]["fields"]["vision_patch"]["acceptance_summary"] == 420
                 rejected = await session.call_tool("claim_task", {"todo_id": "todo_reducer", "agent_id": "other-agent"})
                 assert json.loads(rejected.content[0].text)["ok"] is False
     asyncio.run(exercise())
     assert state.read_bytes() == before
 
 
+def _run_projected_action(
+    action: str,
+    *,
+    launcher: Path,
+    project: Path,
+    replacements: dict[str, str] | None = None,
+) -> dict:
+    rendered = action
+    for source, target in (replacements or {}).items():
+        rendered = rendered.replace(source, target)
+    argv = shlex.split(rendered)
+    assert argv[0] == "loopx"
+    argv[0] = str(launcher)
+    if "--format" not in argv:
+        command_index = next(
+            index
+            for index, token in enumerate(argv)
+            if token in {"quota", "refresh-state"}
+        )
+        argv[command_index:command_index] = ["--format", "json"]
+    result = subprocess.run(
+        argv,
+        cwd=project,
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    payload = json.loads(result.stdout)
+    assert result.returncode == 0, payload.get("error") or payload
+    assert payload["ok"] is True
+    return payload
+
+
+def test_replan_fixture_reentry_executes_bound_vision_settlement_once(tmp_path):
+    project, runtime, launcher = runner.setup_replan(tmp_path)
+    todos = runner.shared.cli(launcher, "todo", "list", "--goal-id", runner.shared.GOAL, "--role", "agent")["todos"]
+    assert len(todos) == 1
+    assert todos[0]["status"] == "done"
+    quota = runner.shared.cli(launcher, "quota", "should-run", "--goal-id", runner.shared.GOAL,
+        "--agent-id", runner.shared.AGENT, "--runtime-profile", "claude_code")
+    assert quota["should_run"] is True
+    assert quota["interaction_contract"]["mode"] != "terminal_no_followup"
+    rows = [json.loads(line) for line in (runtime / "goals" / runner.shared.GOAL / "runs/index.jsonl").read_text().splitlines()]
+    assert any(row.get("agent_vision", {}).get("state") == "vision_closed" for row in rows)
+    actions = quota["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    assert len(actions) == 1
+    assert "--turn-instance-id" in actions[0]
+    assert "spend-slot" not in actions[0]
+
+    turn_instance_id = "claude-replan-turn-1"
+    bound = runner.shared.cli(
+        launcher,
+        "quota",
+        "should-run",
+        "--goal-id",
+        runner.shared.GOAL,
+        "--agent-id",
+        runner.shared.AGENT,
+        "--runtime-profile",
+        "claude_code",
+        "--turn-instance-id",
+        turn_instance_id,
+    )
+    obligation_id = quota["replan_action_packet"]["obligation_id"]
+    assert bound["replan_action_packet"]["obligation_id"] == obligation_id
+    channel = bound["interaction_contract"]["cli_channel"]
+    identity = channel["settlement_plan"]["identity"]
+    assert identity["turn_instance_id"] == turn_instance_id
+    assert identity["replan_obligation_id"] == obligation_id
+    assert identity["binding_kind"] == "autonomous_replan"
+    bound_actions = channel["next_cli_actions"]
+    assert len(bound_actions) == 2
+    refresh_action = next(action for action in bound_actions if "refresh-state" in action)
+    spend_action = next(action for action in bound_actions if "spend-slot" in action)
+    for action in bound_actions:
+        assert f"--replan-obligation-id {obligation_id}" in action
+        assert f"--turn-instance-id {turn_instance_id}" in action
+    assert "--agent-vision-json" in refresh_action
+    assert "--progress-result-class" not in refresh_action
+    assert "--source visible-goal" in spend_action
+
+    vision_path = project / "next-vision.json"
+    vision_path.write_text(
+        json.dumps(
+            {
+                "schema_version": "goal_vision_replan_contract_v0",
+                "state": "vision_patch_proposed",
+                "vision_patch": {
+                    "vision_summary": "Verify asset contents and deliver a durable integrity manifest.",
+                    "role_scope": "Implement and independently validate the bounded manifest integrity stage.",
+                    "acceptance_summary": "Every expected asset has a verified byte size and SHA-256 digest.",
+                    "advancement_policy": "repeat_until_closed",
+                    "replan_trigger_summary": "Create runnable integrity work when the manifest is incomplete.",
+                    "last_patch_summary": "The filename inventory is retained while content integrity becomes the active stage.",
+                },
+                "path_delta": {
+                    "schema_version": "goal_path_delta_v0",
+                    "outcome": "replan",
+                    "prior_assumption": "A filename inventory was sufficient for the current stage.",
+                    "observed_reality": "The active acceptance also requires content size and digest verification.",
+                    "retained": ["Keep the verified list of expected asset paths."],
+                    "changed": ["Add byte-size and SHA-256 verification for every asset."],
+                    "evidence_refs": ["evidence:task-acceptance-review"],
+                },
+            },
+            sort_keys=True,
+        ),
+        encoding="utf-8",
+    )
+    refresh = _run_projected_action(
+        refresh_action,
+        launcher=launcher,
+        project=project,
+        replacements={
+            "<path-to-evidence-linked-goal-vision-replan-contract-v0.json>": str(
+                vision_path
+            )
+        },
+    )
+    assert refresh["settlement_result"]["ok"] is True
+    assert refresh["autonomous_replan_ack"]["semantic_delta"]["accepted"] is True
+    assert refresh["autonomous_replan_ack"]["semantic_delta"][
+        "satisfying_outcomes"
+    ] == ["fresh_vision_path_outcome"]
+
+    spend = _run_projected_action(
+        spend_action,
+        launcher=launcher,
+        project=project,
+    )
+    assert spend["settlement_result"]["ok"] is True
+    replay = _run_projected_action(
+        spend_action,
+        launcher=launcher,
+        project=project,
+    )
+    assert replay["settlement_result"]["ok"] is True
+    rows = [
+        json.loads(line)
+        for line in (
+            runtime / "goals" / runner.shared.GOAL / "runs/index.jsonl"
+        ).read_text().splitlines()
+    ]
+    assert sum(
+        row.get("classification") == "quota_slot_spent"
+        and row.get("replan_obligation_id") == obligation_id
+        for row in rows
+    ) == 1
+    with pytest.raises(AssertionError, match="manifest_acceptance_failed"):
+        runner.verify_manifest_delivery(project)
+
+
+@pytest.mark.parametrize("defect", ["ignores_corruption", "rewrites_inputs", "input_changed"])
+def test_manifest_oracle_rejects_false_acceptance(tmp_path, defect):
+    import hashlib
+
+    for name, content in runner.MANIFEST_ASSETS.items():
+        path = tmp_path / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+    (tmp_path / "manifest.json").write_text(json.dumps({
+        name: {"size": len(content), "sha256": hashlib.sha256(content).hexdigest()}
+        for name, content in runner.MANIFEST_ASSETS.items()
+    }))
+    (tmp_path / "README.md").write_text("Synthetic oracle fixture")
+    (tmp_path / "verify_manifest.py").write_text(
+        "from pathlib import Path\nPath('assets/alpha.txt').write_text('overwritten')\n"
+        if defect == "rewrites_inputs" else "raise SystemExit(0)\n")
+    if defect == "input_changed":
+        (tmp_path / "assets/alpha.txt").write_text("changed")
+    with pytest.raises(AssertionError, match="manifest_(inputs_modified|verifier_unsound|verifier_mutated_evidence)"):
+        runner.verify_manifest_delivery(tmp_path)
+
+
 def test_real_mcp_delivery_completes_and_settles_existing_plan(tmp_path):
     from loopx.goal_mode_mcp import GoalModeMCPConfig, GoalModeMCPControlPlane
 
diff --git a/tests/test_host_vision_recovery.py b/tests/test_host_vision_recovery.py
new file mode 100644
index 0000000000..941a62ff6a
--- /dev/null
+++ b/tests/test_host_vision_recovery.py
@@ -0,0 +1,240 @@
+"""Real CLI/MCP vision recovery: acceptance, accounting and terminal are distinct."""
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.goal_mode_mcp import GoalModeMCPConfig, GoalModeMCPControlPlane
+
+REPO = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location("host_vision_fixture", REPO / "scripts/qualify-native-goal-release.py")
+fixture = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(fixture)
+
+
+def vision(state="no_followup"):
+    return {
+        "schema_version": "goal_vision_replan_contract_v0", "state": state,
+        "vision_patch": {
+            "vision_summary": "Deliver the finite local ledger specification.",
+            "acceptance_summary": "Replay, reversal and CLI atomic output verified.",
+            "last_patch_summary": "Local acceptance tests passed; no external delivery is requested.",
+        },
+    }
+
+
+def control_at(root):
+    project, runtime, launcher = fixture.setup(root)
+    control = GoalModeMCPControlPlane(
+        GoalModeMCPConfig(server_name="vision-test", runtime_profile="claude_code", legacy_host_surface="claude_code"),
+        lambda: {"goal_id": fixture.GOAL, "agent_id": fixture.AGENT},
+    )
+    control.command_prefix = lambda: [str(launcher)]
+    return control, project, runtime
+
+
+def spends(runtime):
+    rows = [json.loads(row) for row in (runtime / "goals" / fixture.GOAL / "runs/index.jsonl").read_text().splitlines()]
+    return [row for row in rows if row.get("classification") == "quota_slot_spent"]
+
+
+@pytest.mark.parametrize("state,terminal", [("no_followup", True), ("vision_patch_proposed", False), ("vision_closed", False)])
+def test_completed_todos_require_vision_decision_not_automatic_goal_close(tmp_path, monkeypatch, state, terminal):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    first = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic lifecycle acceptance", successor_todo_ids=["todo_cli"]))
+    assert first["ok"] is True, first
+    last = json.loads(control.complete_task("todo_cli", fixture.AGENT, "Synthetic lifecycle acceptance", no_follow_up=True))
+    assert last["ok"] is True, last
+    before = json.loads(control.should_run())
+    assert before["should_run"] is True
+    assert before["interaction_contract"]["mode"] != "terminal_no_followup"
+    assert len(spends(runtime)) == 2
+    repaired = json.loads(control.review_task_vision("todo_cli", fixture.AGENT, vision(state)))
+    assert repaired["ok"] is True, repaired
+    assert repaired["vision_checkpoint"]["satisfied"] is True
+    assert repaired["refresh_recovery"]["decision"] == "supplement_checkpoint"
+    replay = json.loads(control.review_task_vision("todo_cli", fixture.AGENT, vision(state)))
+    assert replay["ok"] is True, replay
+    assert replay["appended"] is False
+    assert len(spends(runtime)) == 2
+    after = json.loads(control.should_run())
+    assert (after["interaction_contract"]["mode"] == "terminal_no_followup") is terminal, after
+
+
+@pytest.mark.parametrize("policy", ["as_needed", "repeat_until_closed", "repeat_until_closed_long"])
+def test_first_delivery_can_author_vision_and_later_delivery_can_preserve_it(tmp_path, monkeypatch, policy):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    authored = vision("vision_patch_proposed")
+    authored["vision_patch"]["advancement_policy"] = policy.removesuffix("_long")
+    if policy.endswith("_long"):
+        authored["vision_patch"]["acceptance_summary"] = (
+            "Reducer replay and reversal invariants have been verified against an independent oracle. "
+            "Remaining CLI acceptance includes atomic output, malformed input, deterministic ordering, "
+            "and documented local invocation. "
+        )
+        authored["vision_patch"]["vision_summary"] = (
+            "Deliver a finite standard-library ledger with deterministic replay and reversal behavior. "
+        ) * 4
+        authored["vision_patch"]["role_scope"] = "Local implementation, tests and README; no network delivery. " * 3
+    result = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"], agent_vision=authored))
+    assert result["ok"] is True, json.dumps(result)
+    assert result["settlement"]["durable_writeback"]["vision_checkpoint"]["satisfied"] is True
+    last = json.loads(control.complete_task("todo_cli", fixture.AGENT, "Synthetic acceptance",
+        no_follow_up=True, vision_unchanged_reason="The same acceptance remains in scope."))
+    assert last["ok"] is True, last
+    assert last["settlement"]["durable_writeback"]["vision_checkpoint"]["decision"] == "unchanged_with_reason"
+    assert len(spends(runtime)) == 2
+    assert json.loads(control.should_run())["should_run"] is True
+
+
+def test_recovery_cannot_manufacture_completion_or_override_bound_actor(tmp_path, monkeypatch):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    assert json.loads(control.review_task_vision("todo_reducer", "other-agent", vision()))["ok"] is False
+    result = json.loads(control.review_task_vision("todo_reducer", fixture.AGENT, vision()))
+    assert result["ok"] is False, result
+    assert not list(runtime.glob("goals/*/runs/index.jsonl"))
+
+
+def test_authored_closure_cannot_hide_independent_runnable_work(tmp_path, monkeypatch):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    result = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"], agent_vision=vision("no_followup")))
+    assert result["ok"] is True, result
+    assert len(spends(runtime)) == 1
+    quota = json.loads(control.should_run())
+    assert quota["should_run"] is True
+    assert quota["interaction_contract"]["mode"] != "terminal_no_followup"
+
+
+def test_bad_vision_is_rejected_before_todo_completion_and_can_be_corrected(tmp_path, monkeypatch):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    state = project / "ACTIVE_GOAL_STATE.md"
+    before = state.read_bytes()
+    invalid = vision("vision_patch_proposed")
+    invalid["vision_patch"]["acceptance_summary"] = "x" * 421
+    with pytest.raises(ValueError, match="vision_budget_exceeded"):
+        control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+            successor_todo_ids=["todo_cli"], agent_vision=invalid)
+    assert state.read_bytes() == before
+    assert not list(runtime.glob("goals/*/runs/index.jsonl"))
+    corrected = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"], agent_vision=vision("vision_patch_proposed")))
+    assert corrected["ok"] is True, corrected
+    assert len(spends(runtime)) == 1
+
+
+@pytest.mark.parametrize("length", [240, 241])
+def test_unchanged_reason_preflight_precedes_all_completion_effects(tmp_path, monkeypatch, length):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    first = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"], agent_vision=vision("vision_patch_proposed")))
+    assert first["ok"] is True, first
+    state = project / "ACTIVE_GOAL_STATE.md"
+    before = state.read_bytes()
+    index = runtime / "goals" / fixture.GOAL / "runs/index.jsonl"
+    before_index = index.read_bytes()
+    calls = []
+    original = control.run_cli
+
+    def recorded(args, **kwargs):
+        calls.append(args)
+        return original(args, **kwargs)
+
+    monkeypatch.setattr(control, "run_cli", recorded)
+    if length == 241:
+        with pytest.raises(ValueError, match="vision_unchanged_reason exceeds 240 chars"):
+            control.complete_task("todo_cli", fixture.AGENT, "Synthetic acceptance",
+                no_follow_up=True, vision_unchanged_reason="x" * length)
+        assert calls == []  # No lifecycle, writeback or spend command was executed.
+        assert state.read_bytes() == before
+        assert index.read_bytes() == before_index
+        assert len(spends(runtime)) == 1
+    # Both valid initial authoring and correcting a rejected request take the
+    # real CLI/MCP path. Whitespace is normalized by the same TS owner as refresh.
+    result = json.loads(control.complete_task("todo_cli", fixture.AGENT, "Synthetic acceptance",
+        no_follow_up=True, vision_unchanged_reason="  " + "x" * 240 + "  "))
+    assert result["ok"] is True, result
+    checkpoint = result["settlement"]["durable_writeback"]["vision_checkpoint"]
+    assert checkpoint["decision"] == "unchanged_with_reason"
+    assert checkpoint["unchanged_reason"] == "x" * 240
+    assert len(spends(runtime)) == 2
+    replay = json.loads(control.complete_task("todo_cli", fixture.AGENT, "Synthetic acceptance",
+        no_follow_up=True, vision_unchanged_reason="x" * 240))
+    assert replay["ok"] is True, replay
+    assert len(spends(runtime)) == 2
+
+
+def test_legacy_partial_completion_retries_corrected_vision_without_extra_spend(tmp_path, monkeypatch):
+    import loopx.control_plane.host_adapter_settlement as adapter
+
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    invalid = vision("vision_patch_proposed")
+    invalid["vision_patch"]["acceptance_summary"] = "x" * 421
+    # Emulate an older host without input preflight. Lifecycle and the rejecting
+    # writeback still execute through the real CLI and disposable runtime.
+    with monkeypatch.context() as old_host:
+        old_host.setattr(adapter, "prepare_vision_refresh", lambda *_args, **_kwargs: {})
+        partial = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+            successor_todo_ids=["todo_cli"], agent_vision=invalid))
+    assert partial["completed"] is True and partial["ok"] is False
+    assert partial["settlement"]["failed_stage"] == "durable_writeback"
+    assert partial["recovery"]["tool"] == "complete_task"
+    assert partial["recovery"]["settlement_identity"]["todo_id"] == "todo_reducer"
+    corrected = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"], agent_vision=vision("vision_patch_proposed")))
+    assert corrected["ok"] is True, corrected.get("settlement")
+    assert len(spends(runtime)) == 1
+    replay = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"], agent_vision=vision("vision_patch_proposed")))
+    assert replay["ok"] is True, replay.get("settlement")
+    assert len(spends(runtime)) == 1
+
+
+def test_checkpoint_recovery_missing_baseline_conflict_and_lost_response(tmp_path, monkeypatch):
+    control, project, runtime = control_at(tmp_path)
+    monkeypatch.chdir(project)
+    result = json.loads(control.complete_task("todo_reducer", fixture.AGENT, "Synthetic acceptance",
+        successor_todo_ids=["todo_cli"]))
+    assert result["ok"] is True
+    unchanged = json.loads(control.review_task_vision("todo_reducer", fixture.AGENT,
+        vision_unchanged_reason="Still correct"))
+    assert unchanged.get("vision_checkpoint", {}).get("satisfied") is not True
+    assert len(spends(runtime)) == 1
+    original = control.run_cli
+
+    def response_lost(args, **kwargs):
+        payload = original(args, **kwargs)
+        assert json.loads(payload)["ok"] is True, payload
+        raise TimeoutError("synthetic response loss after commit")
+
+    monkeypatch.setattr(control, "run_cli", response_lost)
+    with pytest.raises(TimeoutError):
+        control.review_task_vision("todo_reducer", fixture.AGENT, vision("vision_patch_proposed"))
+    monkeypatch.setattr(control, "run_cli", original)
+    replay = json.loads(control.review_task_vision("todo_reducer", fixture.AGENT, vision("vision_patch_proposed")))
+    assert replay["ok"] is True, replay
+    assert replay["appended"] is False
+    conflict = json.loads(control.review_task_vision("todo_reducer", fixture.AGENT, vision("no_followup")))
+    assert conflict["ok"] is False
+    assert len(spends(runtime)) == 1
+    # A valid vision decision never consumes the independent open successor.
+    assert json.loads(control.should_run())["should_run"] is True
+
+
+def test_native_outer_controller_owns_new_vision_tool(monkeypatch):
+    from loopx.kunluncode_goal_mode.guards import guard_native_controller_writeback
+    from types import SimpleNamespace
+
+    monkeypatch.setenv("LOOPX_KUNLUNCODE_OUTER_CONTROLLER", "1")
+    control = SimpleNamespace()
+    guard_native_controller_writeback(control)
+    assert json.loads(control.review_task_vision("todo_any", "agent", vision()))["ok"] is False