diff --git a/docs/reference/protocols/goal-vision-replan-contract-v0.md b/docs/reference/protocols/goal-vision-replan-contract-v0.md
index 2d98bd24ed..dc06dacd8d 100644
--- a/docs/reference/protocols/goal-vision-replan-contract-v0.md
+++ b/docs/reference/protocols/goal-vision-replan-contract-v0.md
@@ -518,14 +518,36 @@ For the same `agent_id`, a newer satisfied checkpoint with `patched` or
 `unchanged_with_reason` supersedes older
 `missing_required` checkpoints; `not_required` does not.
 
-The ordinary CLI settlement plan now puts a compact material-closeout reminder
-in its existing writeback step, before the first write. The shared authoring
-example includes the acceptance claim, continuation path and evidence reference;
-replace its illustrative claims and reference with observed facts. Explicit
-`in_flight_continuation` plans retain their existing short writeback precondition.
-This changes guidance in CLI/TurnEnvelope and the shared MCP/replan authoring
-projection, not validation, admission or receipt authority. It does not establish
-lower model token cost or eliminate all replanning; those require live measurement.
+The ordinary Todo-bound CLI settlement plan carries `vision_authoring` in its
+existing durable-writeback step, including in-flight plans. It reuses the same
+TypeScript example and field limits as MCP and replan; it is authoring guidance,
+not a requirement to patch vision on every in-flight step. Replace example claims
+and references with observed facts and pass the packet via `--agent-vision-json`.
+`checkpoint-context` is recovery **after** the original committed Turn writeback,
+not a first-write preparation command. MCP removes the manual CLI settlement plan
+and retains its own authoring projection. The default quota CLI also retains
+autonomous replan's authoring schema. When an identical schema is already present
+in its durable-writeback step, the replan contract uses a same-response
+`vision_authoring_ref`; it never requires a diagnostic detail read for this input.
+Different schemas remain separate, and progress-only replan does not acquire a
+Vision requirement. TurnEnvelope preserves the ordinary plan's guidance.
+
+Shared `todo plan` and guided-start authoring steps also provide a routed exact
+readback command for each authored or reused Todo id. Inventory text is a bounded
+excerpt. Verify `todo_detail_projection.source_complete=true`, compare `.todo.text`
+with the intended requirements and check current status/claim before handoff;
+missing, ambiguous or changed work requires reinspection. Readback does not replace
+quota admission or lease authority. Lists remain compact and exact bodies lossless.
+
+These default guidance changes preserve validation, admission, original-Turn
+recovery and receipt authority. A larger self-contained packet can avoid discovery
+calls, but lower model cost or better outcomes still require live measurement.
+
+普通 Todo 结算包在写回步骤直接携带既有的 vision 示例和字段限制；示例须替换为实际
+事实与证据，不要求每个 in-flight 步骤都修改 vision。`checkpoint-context` 仅用于原
+Turn 写回后的恢复。规划与接续指引提供逐条精确读回命令，用完整 `.todo.text` 核对
+需求和当前状态，不能拿列表摘要判断正文丢失。读取不授予执行权限；准入、租约、
+恢复和单次记账规则保持不变。包体增量与模型效率、效果收益须分别测量。
 
 A satisfied checkpoint is protocol-complete, but a material closeout also has
 to qualify its relationship to the final outcome. A patched checkpoint must
diff --git a/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md b/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md
index 11541e3b41..96d7780b82 100644
--- a/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md
+++ b/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md
@@ -68,12 +68,16 @@ projection. Candidate lists and peer action lists retain counts and point to
 detail section.
 
 When a replan action carries a complete `vision_authoring` schema, the default
-`quota should-run` packet keeps its executable writeback summary (`required_fields`,
-accepted path outcomes, and rule) and replaces only that nested schema with a
-`vision_authoring_detail_ref`. `--include-detail vision` restores the schema.
-`turn plan` is different: its TurnEnvelope preserves the complete schema because
-the plan must be capable of authoring the exact input its validator accepts.
-The crowded Turn budget therefore accounts for that fixed contract without
+`quota should-run` packet now retains it, alongside its executable writeback
+summary (`required_fields`, accepted path outcomes, and rule). Only when the
+durable-writeback step already contains an identical schema does the replan action
+replace its duplicate with `vision_authoring_ref`, an exact path within the same
+response. Missing or different authoring stays inline. This replaces the former
+`vision_authoring_detail_ref` cold read; diagnostic vision audits remain optional
+detail. Progress-only replan does not gain a Vision input requirement.
+`--include-detail vision` retains full explicit detail. `turn plan` preserves the
+complete schema because it must author the exact input its validator accepts.
+The crowded quota and Turn budgets account for that fixed contract without
 relaxing Todo-count growth or the small and multi-Agent ceilings.
 
 ## Qualification Contract
diff --git a/examples/control_plane/cli-output-probe-runner.py b/examples/control_plane/cli-output-probe-runner.py
index 859d3cc4a3..02902e0266 100644
--- a/examples/control_plane/cli-output-probe-runner.py
+++ b/examples/control_plane/cli-output-probe-runner.py
@@ -130,6 +130,7 @@ def _receipt_row(
             else []
         ),
         "next_action_basis_count": semantics.next_action_basis_count(payload),
+        "authoring_inputs": semantics.authoring_input_observations(text),
     }
 
 
diff --git a/loopx/control_plane/effect_program.py b/loopx/control_plane/effect_program.py
index c7877c76dd..e091d2efa3 100644
--- a/loopx/control_plane/effect_program.py
+++ b/loopx/control_plane/effect_program.py
@@ -614,6 +614,7 @@ class SettlementStep:
     command_template: str | None = None
     conditional: bool = False
     command_condition: str | None = None
+    vision_authoring: Mapping[str, Any] | None = None
 
     def as_dict(self) -> dict[str, Any]:
         step: dict[str, Any] = {
@@ -629,6 +630,8 @@ def as_dict(self) -> dict[str, Any]:
             step["conditional"] = True
         if self.command_condition:
             step["command_condition"] = self.command_condition
+        if self.vision_authoring is not None:
+            step["vision_authoring"] = dict(self.vision_authoring)
         return step
 
 
diff --git a/loopx/control_plane/effect_program.ts b/loopx/control_plane/effect_program.ts
index 70d09896de..1c5cfe4402 100644
--- a/loopx/control_plane/effect_program.ts
+++ b/loopx/control_plane/effect_program.ts
@@ -184,6 +184,7 @@ export interface SettlementStep {
   expected_receipt: string;
   command_template?: string;
   command_condition?: "todo_deliverable_complete";
+  vision_authoring?: JsonObject;
   conditional?: true;
 }
 
@@ -606,6 +607,7 @@ export function settlementStepPayload(step: SettlementStep): JsonObject {
   if (step.command_template) payload.command_template = step.command_template;
   if (step.command_condition) payload.command_condition = step.command_condition;
   if (step.conditional) payload.conditional = true;
+  if (step.vision_authoring) payload.vision_authoring = step.vision_authoring;
   return payload;
 }
 
diff --git a/loopx/control_plane/goals/start_goal_todo_delta.py b/loopx/control_plane/goals/start_goal_todo_delta.py
index 5067cd6108..0deb72fbdb 100644
--- a/loopx/control_plane/goals/start_goal_todo_delta.py
+++ b/loopx/control_plane/goals/start_goal_todo_delta.py
@@ -148,6 +148,23 @@ def todo_authoring_steps(
         agent_id=agent_id,
         registry_path=registry_path,
     )
+    # This adapter already owns routed CLI templates for both planning entrypoints.
+    readback = {
+        "id": "read_back_authored_todos",
+        "kind": "operator_or_agent_actions",
+        "command_template": (
+            f"{render_cli_command_prefix(cli_bin=cli_bin, runtime_root=runtime_root)} "
+            + (f"--registry {shell_arg(str(registry_path))} " if registry_path is not None else "")
+            + f"--format json todo list --goal-id {shell_arg(str(goal_id))} "
+            + (f"--agent-id {shell_arg(str(agent_id))} " if agent_id else "")
+            + "--todo-id '<todo-id>'"
+        ),
+        "purpose": (
+            "Read each authored/reused id: one match, todo_detail_projection.source_complete=true; "
+            "compare .todo.text, status/claim. Excerpts cannot verify writes. "
+            "Missing/ambiguous/changed: reinspect before handoff. Readback grants no guard/lease authority."
+        ),
+    }
     if not existing_runnable_frontier:
         return [
             {
@@ -173,6 +190,7 @@ def todo_authoring_steps(
                     "the admitted action_kind and target_key for later quota re-entry"
                 ),
             },
+            readback,
         ]
     return [
         {
@@ -205,6 +223,7 @@ def todo_authoring_steps(
             "add_new_command_template": add_template,
             "purpose": "takeover continues the frontier; authoring is a delta",
         },
+        readback,
     ]
 
 
diff --git a/loopx/control_plane/goals/vision_checkpoint.ts b/loopx/control_plane/goals/vision_checkpoint.ts
index 9b7b05df60..9eedc372c8 100644
--- a/loopx/control_plane/goals/vision_checkpoint.ts
+++ b/loopx/control_plane/goals/vision_checkpoint.ts
@@ -95,7 +95,10 @@ export function visionAuthoringContract(): JsonObject {
         evidence_refs: ["evidence:verified-result"],
       },
     },
-    authoring_hint: "Replace example claims/refs with evidence; obey the live contract and total limit.",
+    authoring_hint: "Replace example claims/refs with evidence; obey the live contract and total limit. " +
+      "For ordinary CLI writeback, pass the packet with --agent-vision-json <file>. " +
+      "checkpoint-context is only for recovery after the original committed Turn writeback; " +
+      "follow its returned same-Turn recovery action, not a fresh-turn preflight.",
     total_text_limit: GOAL_VISION_TOTAL_LIMIT,
     unchanged_reason_limit: VISION_UNCHANGED_REASON_LIMIT,
     path_delta: {
diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py
index a491f1fda8..2062623631 100644
--- a/loopx/control_plane/quota/cli_projection.py
+++ b/loopx/control_plane/quota/cli_projection.py
@@ -462,24 +462,44 @@ def _compact_vision_continuation_audit(
     return compact
 
 
-def _compact_replan_action_packet(packet: dict[str, Any]) -> dict[str, Any]:
-    """Keep the executable writeback summary hot and move its schema cold."""
+def _reference_inline_replan_authoring(
+    packet: dict[str, Any], *, payload: dict[str, Any]
+) -> dict[str, Any]:
+    """Reference an identical executable input already present in this response."""
 
     writeback = packet.get("writeback_contract")
     if not isinstance(writeback, dict) or not isinstance(
         writeback.get("vision_authoring"), dict
     ):
         return packet
+    interaction = payload.get("interaction_contract")
+    cli = interaction.get("cli_channel") if isinstance(interaction, dict) else None
+    plan = cli.get("settlement_plan") if isinstance(cli, dict) else None
+    steps = plan.get("ordered_steps") if isinstance(plan, dict) else None
+    if not isinstance(steps, list):
+        return packet
+    index = next(
+        (index for index, step in enumerate(steps)
+         if isinstance(step, dict) and step.get("kind") == "durable_writeback"
+         and step.get("vision_authoring") == writeback["vision_authoring"]),
+        None,
+    )
+    if index is None:
+        return packet
+    reference = (
+        "$.interaction_contract.cli_channel.settlement_plan.ordered_steps"
+        f"[{index}].vision_authoring"
+    )
     compact_writeback = dict(writeback)
     compact_writeback.pop("vision_authoring")
-    compact_writeback["vision_authoring_detail_ref"] = QUOTA_CLI_VISION_DETAIL_COMMAND
+    compact_writeback["vision_authoring_ref"] = reference
     compact = dict(packet)
     compact["writeback_contract"] = compact_writeback
     compact["payload_compaction"] = {
         "schema_version": QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION,
         "mode": "compact_hot_path",
         "compacted_fields": ["writeback_contract.vision_authoring"],
-        "full_detail_cold_path": QUOTA_CLI_VISION_DETAIL_COMMAND,
+        "projection_ref": reference,
     }
     return compact
 
@@ -759,7 +779,9 @@ def compact_quota_should_run_cli_payload(
         )
     replan_action = payload.get("replan_action_packet")
     if not include_vision_detail and isinstance(replan_action, dict):
-        compact_replan_action = _compact_replan_action_packet(replan_action)
+        compact_replan_action = _reference_inline_replan_authoring(
+            replan_action, payload=compact
+        )
         if compact_replan_action is not replan_action:
             compact = dict(compact)
             compact["replan_action_packet"] = compact_replan_action
diff --git a/loopx/control_plane/quota/effect_program.py b/loopx/control_plane/quota/effect_program.py
index 6602c33953..a24cae6977 100644
--- a/loopx/control_plane/quota/effect_program.py
+++ b/loopx/control_plane/quota/effect_program.py
@@ -212,6 +212,7 @@ def build_turn_scoped_cli_settlement_plan(
             precondition=row["precondition"], idempotency_key_ref=row["idempotency_key_ref"],
             expected_receipt=row["expected_receipt"], command_template=row.get("command_template"),
             conditional=row.get("conditional", False), command_condition=row.get("command_condition"),
+            vision_authoring=row.get("vision_authoring"),
         ) for row in payload["ordered_steps"]
     ), _runtime_payload=payload)
 
diff --git a/loopx/control_plane/quota/settlement_plan.ts b/loopx/control_plane/quota/settlement_plan.ts
index da5e2d62d7..fbb0a9e257 100644
--- a/loopx/control_plane/quota/settlement_plan.ts
+++ b/loopx/control_plane/quota/settlement_plan.ts
@@ -4,7 +4,7 @@ import {
   settlementIdentity, type JsonObject, type SettlementIdentityInput,
   type SettlementPlan, type SettlementStep,
 } from "../effect_program.ts";
-import {VISION_MATERIAL_CLOSEOUT_HINT} from "../goals/vision_checkpoint.ts";
+import {VISION_MATERIAL_CLOSEOUT_HINT, visionAuthoringContract} from "../goals/vision_checkpoint.ts";
 import {parseExactGoalRef} from "../goals/goal_instance_identity.ts";
 import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
 
@@ -56,6 +56,8 @@ export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan
         "outcome_gap: blocked + blocker/evidence IDs; continuation checks.",
       idempotency_key_ref: "$.identity.effect_id", expected_receipt: "durable_writeback_receipt",
       command_template: writeback,
+      // Autonomous replan already carries this owner in its writeback contract.
+      ...(identity.binding_kind === "todo" ? {vision_authoring: visionAuthoringContract()} : {}),
     },
     {
       kind: "quota_spend", owner: "agent",
diff --git a/loopx/control_plane/testing/cli_output_budget.py b/loopx/control_plane/testing/cli_output_budget.py
index 73f7ceefb5..2a52311834 100644
--- a/loopx/control_plane/testing/cli_output_budget.py
+++ b/loopx/control_plane/testing/cli_output_budget.py
@@ -143,22 +143,24 @@ class CliOutputCommandClassification:
             # Fixed-path base/head emit 21,871 JSON / 8,690 Markdown chars.
             # Keep the selected-Todo source, freshness, ordering and stop rules.
             "small": {"json": 22_000, "markdown": 9_000},
-            # 36 Todos / 12 runs emit 35,710 chars; 36,000 leaves 290.
+            # Same 36-Todo / 12-run source: 35,383 -> 38,432 chars when
+            # required replan authoring is inline instead of a diagnostic read.
             # Per-Todo growth and fixed semantic growth remain independent.
-            "crowded": {"json": 36_000, "markdown": 7_800},
+            "crowded": {"json": 39_000, "markdown": 7_800},
             "multi_agent": {"json": 23_000, "markdown": 7_000},
         },
         max_lines={
             "small": {"json": 600, "markdown": 140},
-            "crowded": {"json": 850, "markdown": 90},
+            "crowded": {"json": 950, "markdown": 90},
             "multi_agent": {"json": 650, "markdown": 80},
         },
         scale_axis="todo_count",
         max_json_growth_chars_per_unit=300,
         # Required replan carries dense decision evidence from the full index.
-        # The unchanged fixture grows 21,871 -> 35,710 chars / 579 -> 836 lines.
+        # The required authoring adds 3,049 fixed chars / 94 lines; small and
+        # multi-agent cases stay identical. It does not grow with Todo count.
         # This fixed decision packet must not relax per-Todo growth or other routes.
-        max_json_fixed_semantic_growth_chars=6_000,
+        max_json_fixed_semantic_growth_chars=9_000,
     ),
     CliOutputBudgetSpec(
         surface_id="loopx_turn_plan",
@@ -381,7 +383,10 @@ class CliOutputCommandClassification:
             "packet_summary",
         ),
         markdown_anchor="# Guided Start Goal",
-        max_chars={"json": 64_000, "markdown": 3_200},
+        # The real differential fixture grows from 2,781 to 3,217 Markdown
+        # characters with the complete Todo readback step and command. Keep
+        # that decision guidance; only the explicit detail-mode ceiling grows.
+        max_chars={"json": 64_000, "markdown": 3_600},
         max_lines={"json": 650, "markdown": 55},
     ),
     CliOutputModeVariantSpec(
diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py
index 6066d79744..e0230216ea 100644
--- a/loopx/control_plane/testing/cli_output_differential.py
+++ b/loopx/control_plane/testing/cli_output_differential.py
@@ -721,40 +721,53 @@ def _task_step_read_fence_migration(
     ]
 
 
-def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]:
+def _authoring_input_migration(
+    base: dict[str, Any], candidate: dict[str, Any],
+) -> tuple[dict[Metric, int], list[str]]:
+    """Bound only the observed addition; an already upgraded base gets none."""
+    surface = str(base["row_id"]).partition("/")[2].partition("/")[0]
+    before = base.get("authoring_inputs") or {}
+    after = candidate.get("authoring_inputs") or {}
+    limits: dict[Metric, int] = {}
+    reason = ""
+    if (surface in {"start_goal_guided", "start_goal_guided_command_pack_detail"}
+            and before.get("complete_todo_readback") is False
+            and after.get("complete_todo_readback") is True):
+        # Same fixture: JSON +948 chars/+894 compact/+6 lines; Markdown
+        # +436 chars/+2 lines. Preserve full-body and authority guidance.
+        limits = (
+            {"chars": 1_000, "utf8_bytes": 1_000, "lines": 8, "compact_payload_chars": 960}
+            if base.get("format") == "json" else
+            {"chars": 480, "utf8_bytes": 480, "lines": 3, "compact_payload_chars": 0}
+        )
+        reason = "complete Todo readback step added"
+    elif (surface in {"quota_should_run", "loopx_turn_plan"}
+            and base.get("format") == "json"
+            and before.get("vision_cli_authoring") is False
+            and after.get("vision_cli_authoring") is True):
+        # Default required-replan quota gains the full authoring input:
+        # +3,049 chars/+1,807 compact/+94 lines. Turn already had the schema
+        # and gains only +237 chars of first-write/recovery guidance.
+        limits = (
+            {"chars": 3_200, "utf8_bytes": 3_200, "lines": 96, "compact_payload_chars": 1_920}
+            if surface == "quota_should_run" else
+            {"chars": 256, "utf8_bytes": 256, "lines": 0, "compact_payload_chars": 256}
+        )
+        reason = "Vision CLI authoring input added"
+    return limits, [reason + "; bounded one-time growth"] if limits else []
+
+
+def _compare_row_growth(
+    base: dict[str, Any], candidate: dict[str, Any], *,
+    migration: _SchemaMigrationState,
+    command_route_allowances: dict[Metric, int],
+) -> tuple[dict[str, int | None], dict[str, int | None], list[str], list[str]]:
+    """Compare output cost independently of contract and semantic coverage."""
     row_id = str(base["row_id"])
-    failures: list[str] = []
-    review_signals: list[str] = []
     policy = str(base.get("qualification_policy") or "")
     output_format = str(base.get("format") or "")
-    if candidate.get("qualification_policy") != policy:
-        failures.append("qualification_policy changed")
-    if candidate.get("format") != output_format:
-        failures.append("format changed")
-
-    base_output_contract = base.get("output_contract_version")
-    candidate_output_contract = candidate.get("output_contract_version")
-    heartbeat_agent_input_migration = bool(
-        row_id.startswith("surface/heartbeat_prompt_thin/")
-        and output_format == "json"
-        and base_output_contract is None
-        and candidate_output_contract == HEARTBEAT_AGENT_INPUT_SCHEMA_VERSION
-    )
-    if (
-        candidate_output_contract != base_output_contract
-        and not heartbeat_agent_input_migration
-    ):
-        failures.append("output_contract_version changed")
-
-    migration = _schema_migration_state(
-        base,
-        candidate,
-        output_format=output_format,
-    )
-    added_command_routes, command_route_allowances = (
-        _command_route_growth_allowances(base, candidate)
-    )
-
+    failures: list[str] = []
+    review_signals: list[str] = []
     projection_allowance, projection_failures, projection_signals = _projection_envelope_migration(
         base, candidate, output_format=output_format,
     )
@@ -762,6 +775,8 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
     review_signals.extend(projection_signals)
     fence_allowance, fence_signals = _task_step_read_fence_migration(base, candidate)
     review_signals.extend(fence_signals)
+    authoring_allowance, authoring_signals = _authoring_input_migration(base, candidate)
+    review_signals.extend(authoring_signals)
     deltas: dict[str, int | None] = {}
     allowances: dict[str, int | None] = {}
     for metric in ("chars", "utf8_bytes", "lines", "compact_payload_chars"):
@@ -802,6 +817,7 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
             _decision_evidence_migration_allowance(base, candidate, metric),
             projection_allowance.get(metric, 0),
             fence_allowance.get(metric, 0),
+            authoring_allowance.get(metric, 0),
         )
         # Thin installed prompts contain bilingual lifecycle instructions. A
         # small character-level clarification can cost three bytes per CJK
@@ -846,6 +862,50 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
         if delta > allowance:
             failures.append(f"{metric} grew by {delta}; allowance is {allowance}")
 
+    return deltas, allowances, failures, review_signals
+
+
+def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]:
+    row_id = str(base["row_id"])
+    failures: list[str] = []
+    review_signals: list[str] = []
+    policy = str(base.get("qualification_policy") or "")
+    output_format = str(base.get("format") or "")
+    if candidate.get("qualification_policy") != policy:
+        failures.append("qualification_policy changed")
+    if candidate.get("format") != output_format:
+        failures.append("format changed")
+
+    base_output_contract = base.get("output_contract_version")
+    candidate_output_contract = candidate.get("output_contract_version")
+    heartbeat_agent_input_migration = bool(
+        row_id.startswith("surface/heartbeat_prompt_thin/")
+        and output_format == "json"
+        and base_output_contract is None
+        and candidate_output_contract == HEARTBEAT_AGENT_INPUT_SCHEMA_VERSION
+    )
+    if (
+        candidate_output_contract != base_output_contract
+        and not heartbeat_agent_input_migration
+    ):
+        failures.append("output_contract_version changed")
+
+    migration = _schema_migration_state(
+        base,
+        candidate,
+        output_format=output_format,
+    )
+    added_command_routes, command_route_allowances = (
+        _command_route_growth_allowances(base, candidate)
+    )
+
+    deltas, allowances, growth_failures, growth_signals = _compare_row_growth(
+        base, candidate, migration=migration,
+        command_route_allowances=command_route_allowances,
+    )
+    failures.extend(growth_failures)
+    review_signals.extend(growth_signals)
+
     if output_format == "json":
         missing = _removed(base, candidate, "semantic_json_keys")
         if missing:
diff --git a/loopx/control_plane/testing/cli_output_semantics.py b/loopx/control_plane/testing/cli_output_semantics.py
index a020021b47..03ddda5b0e 100644
--- a/loopx/control_plane/testing/cli_output_semantics.py
+++ b/loopx/control_plane/testing/cli_output_semantics.py
@@ -9,6 +9,53 @@
 from loopx.control_plane.content_digest import ENVELOPED_SHA256_PATTERN
 
 
+def authoring_input_observations(text: str) -> dict[str, bool]:
+    """Recognize complete reviewed guidance for one-time output qualification.
+
+    These observations never classify runtime authority. Match the structured
+    owner and full instruction, not a keyword in arbitrary operator prose.
+    """
+    readback_purpose = (
+        "Read each authored/reused id: one match, todo_detail_projection.source_complete=true; "
+        "compare .todo.text, status/claim. Excerpts cannot verify writes. "
+        "Missing/ambiguous/changed: reinspect before handoff. Readback grants no guard/lease authority."
+    )
+    vision_hint = (
+        "Replace example claims/refs with evidence; obey the live contract and total limit. "
+        "For ordinary CLI writeback, pass the packet with --agent-vision-json <file>. "
+        "checkpoint-context is only for recovery after the original committed Turn writeback; "
+        "follow its returned same-Turn recovery action, not a fresh-turn preflight."
+    )
+    observed = {"complete_todo_readback": False, "vision_cli_authoring": False}
+
+    def collect(value: Any) -> None:
+        if isinstance(value, dict):
+            if (value.get("id") == "read_back_authored_todos"
+                    and value.get("kind") == "operator_or_agent_actions"
+                    and value.get("purpose") == readback_purpose
+                    and "--todo-id '<todo-id>'" in str(value.get("command_template", ""))):
+                observed["complete_todo_readback"] = True
+            if (value.get("schema_version") == "goal_vision_replan_contract_v0"
+                    and value.get("authoring_hint") == vision_hint):
+                observed["vision_cli_authoring"] = True
+            for child in value.values():
+                collect(child)
+        elif isinstance(value, list):
+            for child in value:
+                collect(child)
+
+    try:
+        collect(json.loads(text))
+    except json.JSONDecodeError:
+        observed["complete_todo_readback"] = bool(re.search(
+            r"^\d+\. `read_back_authored_todos` \(operator_or_agent_actions\): "
+            + re.escape(readback_purpose)
+            + r"\n   - command/source: `loopx [^\n]*--todo-id '<todo-id>'`$",
+            text, re.MULTILINE,
+        ))
+    return observed
+
+
 def heartbeat_user_language_prompt_revision(text: str) -> str | None:
     """Attribute the one-time user-language prompt transition in CLI probes.
 
diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json
index 19d8939169..1aa5b3f25d 100644
--- a/loopx/semantics/project_registry_io_manifest_v1.json
+++ b/loopx/semantics/project_registry_io_manifest_v1.json
@@ -1487,7 +1487,7 @@
     },
     {
       "site": "loopx/control_plane/goals/start_goal_todo_delta.py::<module>._read_registry::codec_read:load_registry#1",
-      "line": 229,
+      "line": 248,
       "column": 19,
       "kind": "codec_read",
       "api": "load_registry",
diff --git a/tests/control_plane/test_cli_output_budget.py b/tests/control_plane/test_cli_output_budget.py
index 5891024188..5b6deccd18 100644
--- a/tests/control_plane/test_cli_output_budget.py
+++ b/tests/control_plane/test_cli_output_budget.py
@@ -1167,7 +1167,7 @@ def test_malformed_todo_state_fails_closed_without_dropping_bounded_detail(
         assert default_payload.get(key) == detail_payload.get(key)
 
 
-def test_quota_cli_bounds_real_scale_vision_audit_and_keeps_cold_detail(
+def test_quota_cli_keeps_required_authoring_inline_and_diagnostic_audit_cold(
     tmp_path: Path,
 ) -> None:
     with _stable_budget_fixture_root(tmp_path / "quota-vision-detail") as stable_root:
@@ -1196,9 +1196,10 @@ def test_quota_cli_bounds_real_scale_vision_audit_and_keeps_cold_detail(
 
     assert default_exit_code == 0, default_text
     assert detail_exit_code == 0, detail_text
-    # Pinned base/head emit 41,503 chars on the same 36-Todo / 12-run vision
-    # fixture. Keep complete decision semantics; 42k leaves 497 chars.
-    assert len(default_text) <= 42_000
+    # Same 36-Todo / 12-run command: 41,176 -> 44,225 chars (+7.4%).
+    # Required authoring replaces a second diagnostic read; preserve its rules
+    # and field limits. This fixed allowance does not relax other lane budgets.
+    assert len(default_text) <= 45_000
     default_payload = json.loads(default_text)
     detail_payload = json.loads(detail_text)
     compact_audit = default_payload["vision_continuation_audit"]
@@ -1212,16 +1213,8 @@ def test_quota_cli_bounds_real_scale_vision_audit_and_keeps_cold_detail(
     assert "registry_read_instruction" in detailed_audit["vision_gap_judge"]
     compact_replan = default_payload["replan_action_packet"]
     detailed_replan = detail_payload["replan_action_packet"]
-    assert compact_replan["payload_compaction"] == {
-        "schema_version": "quota_cli_replan_action_compaction_v0",
-        "mode": "compact_hot_path",
-        "compacted_fields": ["writeback_contract.vision_authoring"],
-        "full_detail_cold_path": "quota should-run --include-detail vision",
-    }
-    assert "vision_authoring" not in compact_replan["writeback_contract"]
-    assert compact_replan["writeback_contract"]["vision_authoring_detail_ref"] == (
-        "quota should-run --include-detail vision"
-    )
+    assert compact_replan["writeback_contract"] == detailed_replan["writeback_contract"]
+    assert "payload_compaction" not in compact_replan
     assert detailed_replan["writeback_contract"]["vision_authoring"][
         "schema_version"
     ] == "goal_vision_replan_contract_v0"
diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py
index 72c15e44c3..30c374f921 100644
--- a/tests/control_plane/test_cli_output_differential.py
+++ b/tests/control_plane/test_cli_output_differential.py
@@ -61,6 +61,84 @@ def _receipt(*rows: dict[str, object]) -> dict[str, object]:
     }
 
 
+@pytest.mark.parametrize("surface,output_format,feature,growth", [
+    ("start_goal_guided", "json", "complete_todo_readback", (948, 6, 894)),
+    ("start_goal_guided_command_pack_detail", "markdown", "complete_todo_readback", (436, 2, 0)),
+    ("quota_should_run", "json", "vision_cli_authoring", (3049, 94, 1807)),
+    ("loopx_turn_plan", "json", "vision_cli_authoring", (237, 0, 237)),
+])
+def test_authoring_growth_is_bounded_one_time_and_surface_scoped(
+    surface, output_format, feature, growth,
+):
+    from loopx.control_plane.testing.cli_output_differential import _compare_row
+
+    base = _row(
+        row_id=f"surface/{surface}/small/{output_format}", format=output_format,
+        chars=1000, utf8_bytes=1000, lines=20, compact_payload_chars=800,
+        authoring_inputs={feature: False},
+    )
+    candidate = {
+        **base, "authoring_inputs": {feature: True},
+        "chars": 1000 + growth[0], "utf8_bytes": 1000 + growth[0],
+        "lines": 20 + growth[1], "compact_payload_chars": 800 + growth[2],
+    }
+    result = _compare_row(base, candidate)
+    assert not result["failures"]
+    assert result["review_signals"]
+    # No waiver for an already-upgraded or unobserved base, or another surface.
+    assert _compare_row({**base, "authoring_inputs": {feature: True}}, candidate)["failures"]
+    assert _compare_row({**base, "authoring_inputs": {}}, candidate)["failures"]
+    assert _compare_row(base, {**candidate, "authoring_inputs": {feature: False}})["failures"]
+    assert _compare_row(
+        {**base, "row_id": f"surface/status/small/{output_format}"},
+        {**candidate, "row_id": f"surface/status/small/{output_format}"},
+    )["failures"]
+    # Retain independent metric limits; a declared transition is not unlimited.
+    for metric in ("chars", "utf8_bytes", "lines", "compact_payload_chars"):
+        inflated = {**candidate, metric: base[metric] + result["allowances"][metric] + 1}
+        assert _compare_row(base, inflated)["failures"]
+
+
+def test_authoring_observations_require_complete_structured_or_rendered_guidance():
+    from loopx.control_plane.testing.cli_output_semantics import authoring_input_observations
+
+    purpose = (
+        "Read each authored/reused id: one match, todo_detail_projection.source_complete=true; "
+        "compare .todo.text, status/claim. Excerpts cannot verify writes. "
+        "Missing/ambiguous/changed: reinspect before handoff. Readback grants no guard/lease authority."
+    )
+    step = {
+        "id": "read_back_authored_todos", "kind": "operator_or_agent_actions",
+        "purpose": purpose,
+        "command_template": "loopx --format json todo list --goal-id test --todo-id '<todo-id>'",
+    }
+    rendered = (
+        f"5. `read_back_authored_todos` (operator_or_agent_actions): {purpose}\n"
+        f"   - command/source: `{step['command_template']}`"
+    )
+    for text in (json.dumps({"steps": [step]}), rendered):
+        assert authoring_input_observations(text)["complete_todo_readback"]
+        assert not authoring_input_observations(text.replace("Excerpts cannot verify writes. ", ""))[
+            "complete_todo_readback"
+        ]
+        assert not authoring_input_observations(text.replace("--todo-id", "--limit"))[
+            "complete_todo_readback"
+        ]
+    assert not authoring_input_observations(json.dumps({"note": rendered}))["complete_todo_readback"]
+    hint = (
+        "Replace example claims/refs with evidence; obey the live contract and total limit. "
+        "For ordinary CLI writeback, pass the packet with --agent-vision-json <file>. "
+        "checkpoint-context is only for recovery after the original committed Turn writeback; "
+        "follow its returned same-Turn recovery action, not a fresh-turn preflight."
+    )
+    authoring = {"schema_version": "goal_vision_replan_contract_v0", "authoring_hint": hint}
+    assert authoring_input_observations(json.dumps(authoring))["vision_cli_authoring"]
+    assert not authoring_input_observations(json.dumps({"authoring_hint": hint}))["vision_cli_authoring"]
+    assert not authoring_input_observations(json.dumps({**authoring, "authoring_hint": hint[:100]}))[
+        "vision_cli_authoring"
+    ]
+
+
 def test_thin_bilingual_byte_allowance_does_not_relax_character_or_quota_limits():
     from loopx.control_plane.testing.cli_output_differential import _compare_row
 
diff --git a/tests/control_plane/test_effective_turn_replan.py b/tests/control_plane/test_effective_turn_replan.py
index 9234de8eed..e28c8c1318 100644
--- a/tests/control_plane/test_effective_turn_replan.py
+++ b/tests/control_plane/test_effective_turn_replan.py
@@ -231,16 +231,28 @@ def periodic():
     assert selected_rc == 0, selected
     contract = selected["replan_action_packet"]["writeback_contract"]
     assert contract["preferred_input"] == "evidence_linked_vision_path"
+    plan = selected["interaction_contract"]["cli_channel"]["settlement_plan"]
+    assert contract["vision_authoring_ref"] == (
+        "$.interaction_contract.cli_channel.settlement_plan.ordered_steps[1].vision_authoring"
+    )
+    authoring = plan["ordered_steps"][1]["vision_authoring"]
+    assert authoring["schema_version"] == "goal_vision_replan_contract_v0"
+    if threshold_override is None and device_count is None:
+        rc, envelope = _run_cli(
+            registry, runtime, "quota", "should-run", "--codex-app", "--turn-envelope",
+            "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
+            "--turn-instance-id", "effective-turn-next", "--scan-path", str(project),
+        )
+        assert rc == 0, envelope
+        assert envelope["replan_action_packet"]["writeback_contract"]["vision_authoring"] == authoring
+        assert envelope["writeback"]["settlement_plan"] == plan
     vision_path = tmp_path / "periodic-vision.json"
-    vision = {
-        "schema_version": "goal_vision_replan_contract_v0",
-        "state": "vision_patch_proposed",
-        "vision_patch": {"acceptance_summary": "Original acceptance remains open."},
-        "path_delta": {"schema_version": "goal_path_delta_v0", "outcome": "replan",
+    vision = json.loads(json.dumps(authoring["minimal_example"]))
+    vision["vision_patch"] = {"acceptance_summary": "Original acceptance remains open."}
+    vision["path_delta"] = {"schema_version": "goal_path_delta_v0", "outcome": "replan",
                        "prior_assumption": "The existing method remains viable.",
                        "observed_reality": "Current validation supports continuing it.",
-                       "retained": ["Existing method and open Todo"], "evidence_refs": []},
-    }
+                       "retained": ["Existing method and open Todo"], "evidence_refs": []}
     command = shlex.split(selected["interaction_contract"]["cli_channel"]["next_cli_actions"][0])
     command = command[command.index("refresh-state"):]
     command[command.index("--agent-vision-json") + 1] = str(vision_path)
diff --git a/tests/control_plane/test_quota_cli_projection.py b/tests/control_plane/test_quota_cli_projection.py
index 35101edf42..979dc7d887 100644
--- a/tests/control_plane/test_quota_cli_projection.py
+++ b/tests/control_plane/test_quota_cli_projection.py
@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import json
+from copy import deepcopy
+
+import pytest
 
 from loopx.control_plane.quota.cli_projection import (
     QUOTA_CLI_GOAL_BOUNDARY_DETAIL_COMMAND,
@@ -14,6 +17,47 @@
 )
 
 
+@pytest.mark.parametrize("inline", ["absent", "identical", "different", "wrong-step"])
+def test_replan_authoring_is_available_without_another_read(inline: str) -> None:
+    authoring = {"schema_version": "goal_vision_replan_contract_v0",
+                 "minimal_example": {"state": "vision_patch_proposed"},
+                 "fields": {"vision_patch": {"acceptance_summary": 420}}}
+    packet = {"obligation_id": "replan-fixture", "writeback_contract": {
+        "vision_authoring": authoring, "preferred_input": "evidence_linked_vision_path",
+        "rule": "Retain current requirements and cite observed evidence."}}
+    steps = [{"kind": "validation"}]
+    if inline != "absent":
+        steps.append({"kind": "validation" if inline == "wrong-step" else "durable_writeback",
+                      "vision_authoring": {**authoring, **({"additional_rule": "Keep this."}
+                                                          if inline == "different" else {})}})
+    payload = {"replan_action_packet": packet, "heartbeat_receipt": {"turn_instance_id": "turn-one"},
+               "interaction_contract": {"cli_channel": {"settlement_plan": {"ordered_steps": steps}}}}
+    original = deepcopy(payload)
+    compact = compact_quota_should_run_cli_payload(payload)
+    writeback = compact["replan_action_packet"]["writeback_contract"]
+    assert "vision_authoring_detail_ref" not in writeback
+    assert writeback["rule"] == packet["writeback_contract"]["rule"]
+    if inline == "identical":
+        assert "vision_authoring" not in writeback
+        assert writeback["vision_authoring_ref"] == (
+            "$.interaction_contract.cli_channel.settlement_plan.ordered_steps[1].vision_authoring")
+        assert compact["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"][1][
+            "vision_authoring"] == authoring
+    else:
+        assert writeback["vision_authoring"] == authoring
+        assert "vision_authoring_ref" not in writeback
+    assert compact["heartbeat_receipt"] == payload["heartbeat_receipt"]
+    assert compact["interaction_contract"] == payload["interaction_contract"]
+    assert payload == original
+    assert compact_quota_should_run_cli_payload(payload, include_vision_detail=True) == original
+
+
+def test_progress_only_replan_does_not_invent_a_vision_input() -> None:
+    payload = {"replan_action_packet": {"obligation_id": "progress-fixture",
+                                       "writeback_contract": {}}}
+    assert compact_quota_should_run_cli_payload(payload) == payload
+
+
 def _items(count: int, *, prefix: str) -> list[dict[str, object]]:
     return [
         {
diff --git a/tests/control_plane/test_settlement_input_discovery_cli.py b/tests/control_plane/test_settlement_input_discovery_cli.py
new file mode 100644
index 0000000000..4132559a5f
--- /dev/null
+++ b/tests/control_plane/test_settlement_input_discovery_cli.py
@@ -0,0 +1,180 @@
+"""Executable planning/readback and first-write guidance on real local stores."""
+from __future__ import annotations
+
+import json
+import shlex
+from copy import deepcopy
+
+import pytest
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.quota.turn_envelope import build_turn_envelope
+from loopx.control_plane.todos.active_state_todo_parser import parse_todo_source
+from tests.control_plane.test_quota_settlement_cli import (
+    AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _run_cli, _run_generated_cli,
+    _spend_run_count, _write_fixture,
+)
+
+BODY = "[P1] " + "Retain the original requirements and verify the result. " * 24 + "TAIL: preserve cancellation."
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+@pytest.mark.parametrize("in_flight", [False, True])
+def test_projected_inputs_support_real_readback_writeback_and_one_debit(tmp_path, monkeypatch, provider, in_flight):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    state.write_text(state.read_text().replace("[P1] Validate and settle the selected delivery.", BODY).replace(
+        "action_kind=validate -->", f"action_kind=validate claimed_by={AGENT_ID} -->"))
+    if provider != "legacy":
+        goal = json.loads(registry.read_text())["goals"][0]
+        active, archived, _ = parse_todo_source(state.read_text(), goal=goal, state_path=state)
+        rows = [{"schema_version": "todo_item_v0", **row}
+                for row in [*active["agent"], *active["user"], *archived]]
+        initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection(
+            goal_id=GOAL_ID, todos=rows, handoff_mode="legacy", leases=[]),
+            state_path=state, provider=provider)
+
+    def run(*args):
+        return _run_cli(registry, runtime, *args, cwd=project)
+
+    rc, planning = run("todo", "plan", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+                       "--text", "Verify the implementation while retaining all requirements.")
+    assert rc == 0, planning
+    row = next(t for t in planning["existing_todos"] if t["todo_id"] == TODO_ID)
+    assert len(row["text"]) < len(BODY)
+    step = planning["ordered_steps"][-1]
+    readback = step["command_template"].replace("<todo-id>", TODO_ID)
+    rc, exact = _run_generated_cli(readback, registry_path=registry)
+    assert rc == 0 and exact["matched"], exact
+    assert exact["todo_detail_projection"]["source_complete"] is True
+    assert exact["todo"]["text"] == BODY
+    assert exact["todo"]["status"] == "open"
+    assert exact["todo"]["claimed_by"] == AGENT_ID
+    rc, missing = _run_generated_cli(readback.replace(TODO_ID, "todo_missing"), registry_path=registry)
+    assert rc == 0 and missing["not_found"] and missing["todo"] is None
+    assert _spend_run_count(runtime) == 0
+
+    binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
+               "--turn-instance-id", TURN_ID)
+    rc, guard = run("quota", "should-run", "--codex-app", *binding, "--scan-path", str(project))
+    assert rc == 0 and guard["interaction_contract"]["agent_channel"]["delivery_allowed"], guard
+    plan = guard["interaction_contract"]["cli_channel"]["settlement_plan"]
+    authoring = plan["ordered_steps"][1]["vision_authoring"]
+    assert authoring["schema_version"] == "goal_vision_replan_contract_v0"
+    assert build_turn_envelope(guard)["writeback"]["settlement_plan"] == plan
+    rc, premature = run("checkpoint-context", *binding)
+    assert rc != 0 and "original committed Turn writeback" in premature["error"]
+    rc, no_writeback = _run_generated_cli(plan["ordered_steps"][2]["command_template"], registry_path=registry)
+    assert rc != 0 and _spend_run_count(runtime) == 0, no_writeback
+
+    if not in_flight:
+        completion = plan["ordered_steps"][0]["command_template"].replace(
+            "<validated evidence>", "Exact original body, status and claim verified.")
+        rc, completed = _run_generated_cli(completion, registry_path=registry)
+        assert rc == 0, completed
+
+    # Author the documented shape with synthetic, locally verified facts; the
+    # sample's evidence never substitutes for this fixture's actual readback.
+    vision = deepcopy(authoring["minimal_example"])
+    vision["vision_patch"]["acceptance_summary"] = "Exact source body and claim verified."
+    vision["path_delta"].update(observed_reality="The complete current Todo matches its original body.",
+                               retained=["Original requirements"], evidence_refs=["validation:exact-body"])
+    path = tmp_path / "vision.json"
+    refresh = plan["ordered_steps"][1]["command_template"].replace("<validated_progress>", "validated_progress").replace(
+        "<scale>", "implementation").replace("<outcome>", "outcome_progress")
+    if in_flight:
+        refresh += " --delivery-boundary in_flight_continuation"
+    refresh += f" --agent-vision-json {shlex.quote(str(path))} --no-global-sync --suppress-external-sinks"
+    invalid = deepcopy(vision)
+    invalid["vision_patch"]["acceptance_summary"] = "x" * (authoring["fields"]["vision_patch"]["acceptance_summary"] + 1)
+    path.write_text(json.dumps(invalid))
+    rc, rejected = _run_generated_cli(refresh, registry_path=registry)
+    assert rc != 0 and _spend_run_count(runtime) == 0, rejected
+    path.write_text(json.dumps(vision))
+    rc, written = _run_generated_cli(refresh, registry_path=registry)
+    assert rc == 0 and written["vision_checkpoint"]["satisfied"], written
+    assert written["settlement_identity"] == plan["identity"]
+    for replay in (False, True):
+        rc, spent = _run_generated_cli(written["settlement_owed"]["command"], registry_path=registry)
+        assert rc == 0 and spent["appended"] is not replay, spent
+        assert _spend_run_count(runtime) == 1
+    rc, final_todo = _run_generated_cli(readback, registry_path=registry)
+    assert rc == 0 and final_todo["todo"]["status"] == ("open" if in_flight else "done")
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_todoless_periodic_review_uses_default_authoring_and_settles_once(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+
+    def run(*args):
+        return _run_cli(registry, runtime, *args, cwd=project)
+
+    rc, configured = run("configure-goal", "--goal-id", GOAL_ID,
+                         "--execution-replan-after-turns", "1", "--execute")
+    assert rc == 0, configured
+    binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+               "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID)
+    rc, guard = run("quota", "should-run", "--codex-app", *binding, "--scan-path", str(project))
+    assert rc == 0, guard
+    rc, written = run("refresh-state", *binding, "--classification", "validated_progress",
+                      "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+                      "--delivery-boundary", "in_flight_continuation", "--no-global-sync",
+                      "--suppress-external-sinks")
+    assert rc == 0, written
+    rc, spent = run("quota", "spend-slot", *binding, "--slots", "1", "--source", "heartbeat", "--execute")
+    assert rc == 0 and _spend_run_count(runtime) == 1, spent
+
+    # Synthetic absence of runnable work; retain the real committed receipt.
+    # Initialize each authority from this fixture, never rewrite a live store.
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    state.write_text(state.read_text().replace(
+        "- [ ] [P1] Validate and settle the selected delivery.",
+        "- [x] [P1] Validate and settle the selected delivery.").replace(
+        f"todo_id={TODO_ID} status=open", f"todo_id={TODO_ID} status=done"))
+    if provider != "legacy":
+        goal = json.loads(registry.read_text())["goals"][0]
+        active, archived, _ = parse_todo_source(state.read_text(), goal=goal, state_path=state)
+        rows = [{"schema_version": "todo_item_v0", **row}
+                for row in [*active["agent"], *active["user"], *archived]]
+        initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection(
+            goal_id=GOAL_ID, todos=rows, handoff_mode="legacy", leases=[]),
+            state_path=state, provider=provider)
+
+    current = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+               "--turn-instance-id", "periodic-without-todo", "--scan-path", str(project))
+    rc, guard = run("quota", "should-run", "--codex-app", *current)
+    assert rc == 0 and guard["selected_todo"] is None, guard
+    identity = guard["heartbeat_receipt"]["settlement_identity"]
+    assert identity["binding_kind"] == "autonomous_replan"
+    writeback = guard["replan_action_packet"]["writeback_contract"]
+    authoring = writeback["vision_authoring"]
+    rc, detail = run("quota", "should-run", "--codex-app", *current, "--include-detail", "vision")
+    assert rc == 0 and detail["replan_action_packet"]["writeback_contract"] == writeback, detail
+    assert build_turn_envelope(guard)["replan_action_packet"]["writeback_contract"] == writeback
+    rc, envelope = run("quota", "should-run", "--codex-app", *current, "--turn-envelope")
+    assert rc == 0 and envelope["replan_action_packet"]["writeback_contract"] == writeback, envelope
+    vision = deepcopy(authoring["minimal_example"])
+    vision["path_delta"].update(
+        outcome="replan", observed_reality="The settled fixture has no runnable Todo.",
+        retained=["Original acceptance"], evidence_refs=[])
+    path = tmp_path / "periodic-vision.json"
+    command = shlex.split(guard["interaction_contract"]["cli_channel"]["next_cli_actions"][0])
+    command = command[command.index("refresh-state"):]
+    command[command.index("--agent-vision-json") + 1] = str(path)
+    command += ["--no-global-sync", "--suppress-external-sinks"]
+    path.write_text(json.dumps(vision))
+    rc, rejected = run(*command)
+    assert rc != 0 and _spend_run_count(runtime) == 1, rejected
+    assert "typed semantic delta" in rejected["error"], rejected
+    vision["path_delta"]["evidence_refs"] = ["validation:no-runnable-fixture"]
+    path.write_text(json.dumps(vision))
+    rc, accepted = run(*command)
+    assert rc == 0 and accepted["settlement_identity"] == identity, json.dumps(accepted)
+    assert accepted["autonomous_replan_ack"]["semantic_delta"]["satisfying_outcomes"] == ["fresh_vision_path_outcome"]
+    for replay in (False, True):
+        rc, spent = _run_generated_cli(accepted["settlement_owed"]["command"], registry_path=registry)
+        assert rc == 0 and spent["appended"] is not replay, spent
+        assert _spend_run_count(runtime) == 2
diff --git a/tests/control_plane/test_start_goal_compact_projection.py b/tests/control_plane/test_start_goal_compact_projection.py
index 33ec404732..cbb082c8ba 100644
--- a/tests/control_plane/test_start_goal_compact_projection.py
+++ b/tests/control_plane/test_start_goal_compact_projection.py
@@ -591,7 +591,8 @@ def test_issue_fix_goal_projects_capability_guard_without_todo_fields(
     assert "loopx issue-fix workflow-plan " in rendered
     assert "loopx issue-fix feasibility " in rendered
     assert "proceed: `loopx issue-fix feasibility " in rendered
-    assert len(rendered.replace(str(project), "<project>")) <= 3_600
+    # Full-body readback adds a routed command and authority/completeness guidance.
+    assert len(rendered.replace(str(project), "<project>")) <= 4_100
     todo_command = next(
         step["command_template"]
         for step in transaction["ordered_steps"]
@@ -872,7 +873,9 @@ def test_projection_materially_reduces_repetition_without_hiding_measurement(
     compact_json = json.dumps(compact, ensure_ascii=False, sort_keys=True)
     detailed_json = json.dumps(detailed, ensure_ascii=False, sort_keys=True)
 
-    assert len(compact_json) <= int(len(detailed_json) * 0.65)
+    # Exact-readback guidance is useful common content in both projections;
+    # retain the duplicate-content checks below rather than remove that guidance.
+    assert len(compact_json) <= int(len(detailed_json) * 0.66)
     compact_duplication = compact["packet_summary"]["duplication_measurement"]
     detailed_duplication = detailed["packet_summary"]["duplication_measurement"]
     assert compact_duplication["objective_content"]["duplicate_occurrences"] <= 11
@@ -2079,12 +2082,15 @@ def test_guided_takeover_with_runnable_frontier_projects_todo_delta(
     validate_shared_todo_options(add_args)
     validate_todo_add_options(add_args)
     # The takeover continues through the normal refresh/host-loop/quota tail.
-    assert step_ids.index("apply_todo_delta") < step_ids.index("refresh_state")
+    assert step_ids.index("apply_todo_delta") < step_ids.index("read_back_authored_todos") < step_ids.index("refresh_state")
     assert "quota_guard" in step_ids
     rendered = render_start_goal_guided_markdown(payload)
     assert "apply_todo_delta" in rendered
     assert "reuse" in rendered
     assert "todo_delta:" in rendered
+    readback = next(step for step in steps if step["id"] == "read_back_authored_todos")
+    assert readback["command_template"] in rendered
+    assert "todo_detail_projection.source_complete=true" in rendered
 
 
 def test_guided_takeover_without_runnable_frontier_keeps_unconditional_authoring(
@@ -2095,6 +2101,7 @@ def test_guided_takeover_without_runnable_frontier_keeps_unconditional_authoring
     step_ids = [step["id"] for step in payload["guided_transaction"]["ordered_steps"]]
     assert "write_ordered_todos" in step_ids
     assert "apply_todo_delta" not in step_ids
+    assert step_ids.index("write_ordered_todos") < step_ids.index("read_back_authored_todos") < step_ids.index("refresh_state")
 
 
 def _write_connected_project_with_blocked_agent_todo(root: Path) -> Path:
diff --git a/tests/control_plane/test_task_planning.py b/tests/control_plane/test_task_planning.py
index 826a78746a..6f5fba6631 100644
--- a/tests/control_plane/test_task_planning.py
+++ b/tests/control_plane/test_task_planning.py
@@ -52,6 +52,7 @@ def test_empty_frontier_plans_before_writes_without_starting_a_loop(bound_goal):
     assert [step["id"] for step in packet["ordered_steps"]] == [
         "plan_ranked_todos",
         "write_ordered_todos",
+        "read_back_authored_todos",
     ]
     assert packet["planner"]["required_before_todo_write"] is True
     assert packet["execution_handoff"] == {
@@ -82,6 +83,7 @@ def test_existing_plan_is_an_incremental_frontier_not_a_new_goal(bound_goal):
     assert [step["id"] for step in packet["ordered_steps"]] == [
         "compare_planned_todos_with_frontier",
         "apply_todo_delta",
+        "read_back_authored_todos",
     ]
     assert all(item["todo_id"] != "todo_peer" for item in packet["existing_todos"])
 
diff --git a/tests/control_plane_ts/quota_settlement_plan.test.ts b/tests/control_plane_ts/quota_settlement_plan.test.ts
index af6292ade4..e0755e8bda 100644
--- a/tests/control_plane_ts/quota_settlement_plan.test.ts
+++ b/tests/control_plane_ts/quota_settlement_plan.test.ts
@@ -1,6 +1,8 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import {settlementPlanPayload} from "../../loopx/control_plane/effect_program.ts";
+import {visionAuthoringContract} from "../../loopx/control_plane/goals/vision_checkpoint.ts";
+import {projectMcpInteraction} from "../../loopx/control_plane/turn_driver/host_interaction.ts";
 import {turnScopedCliSettlementPlan} from "../../loopx/control_plane/quota/settlement_plan.ts";
 
 const input = {
@@ -61,6 +63,7 @@ test("autonomous replan never manufactures a Todo completion step", () => {
   assert.deepEqual(plan.steps.map(step => step.kind), ["validation", "durable_writeback", "quota_spend"]);
   assert.equal(plan.steps[0].command_template, undefined);
   assert.equal(plan.identity.binding_kind, "autonomous_replan");
+  assert.ok(plan.steps.every(step => step.vision_authoring === undefined));
   assert.match(plan.steps[1].precondition, /^validation succeeded/);
   assert.match(plan.steps[1].precondition, /Route elimination needs evidence/);
   assert.match(plan.steps[1].precondition, /failure alone is not progress/);
@@ -80,3 +83,19 @@ test("incomplete command facts or ambiguous identity fail closed", () => {
     identity: {...input.identity, replan_obligation_id: "replan"},
   }), /cannot bind both/);
 });
+
+
+test("ordinary and in-flight CLI expose one authoring owner without duplicate MCP guidance", () => {
+  for (const delivery_boundary of [undefined, "in_flight_continuation"]) {
+    const wire = settlementPlanPayload(turnScopedCliSettlementPlan({...input, delivery_boundary}));
+    const steps = wire.ordered_steps as Record<string, unknown>[];
+    assert.equal(steps.filter(step => step.vision_authoring).length, 1);
+    assert.deepEqual(steps[1].vision_authoring, visionAuthoringContract());
+    const mcp = projectMcpInteraction({ok: true, normal_delivery_allowed: true,
+      selected_todo: {todo_id: "todo"}, interaction_contract: {cli_channel: {settlement_plan: wire}}});
+    const contract = mcp.interaction_contract as Record<string, Record<string, unknown>>;
+    assert.equal(contract.cli_channel.settlement_plan, undefined);
+    assert.deepEqual(contract.mcp_channel.vision_authoring, steps[1].vision_authoring);
+    assert.match(visionAuthoringContract().authoring_hint as string, /original committed Turn writeback/);
+  }
+});