diff --git a/docs/guides/context-delivery.md b/docs/guides/context-delivery.md
index b7fc90c7ec..f7f27966d9 100644
--- a/docs/guides/context-delivery.md
+++ b/docs/guides/context-delivery.md
@@ -75,6 +75,14 @@ One shared preparation call does not prove that every final caller has migrated.
 | Saved-file delta | Verified immutable file versions with compatible identity | Model retention, comprehension or fresh permission |
 | Provider prompt cache | Provider-specific processing/billing reuse of input | Source correctness, task success or LoopX net savings |
 
+Replan authoring uses the same-response form: CLI display can reference the
+complete `vision_authoring` in a durable-writeback step. The existing TypeScript
+Turn owner chooses an exact owner and resolves it for ordinary and captured
+readers. Dangling references, reference chains, conflicting inline values and
+references to another kind of step are rejected. A small or different value
+stays inline. Sharing neither changes the original settlement identity nor
+establishes cross-Turn retention or net cost savings.
+
 Semantic summarization is another operation: it can change wording or meaning.
 The exact codec is not a semantic summary. A controller-authored brief remains
 a separate proposed workflow with its own retained-requirement and outcome
diff --git a/docs/guides/context-delivery.zh-CN.md b/docs/guides/context-delivery.zh-CN.md
index 7982e5c7bf..f4c46acb3e 100644
--- a/docs/guides/context-delivery.zh-CN.md
+++ b/docs/guides/context-delivery.zh-CN.md
@@ -62,6 +62,12 @@ flowchart TD
 | 保存文件差分 | 身份兼容、经过校验的不可变文件版本 | 模型保留、理解或新鲜权限 |
 | Provider prompt cache | Provider 自己的输入处理／计费复用 | 来源正确、任务成功或 LoopX 净收益 |
 
+重规划 authoring 使用同响应引用：CLI 展示可指向 durable-writeback 步骤中完整的
+`vision_authoring`。现有 TypeScript Turn owner 选择精确一致的正文，并为普通及捕获
+读取器展开它。悬空引用、引用链、与引用冲突的内联值，以及指向其他种类步骤的引用
+都必须拒绝；很小或不同的值保持内联。共享不改变原结算身份，也不证明跨 Turn 保留
+或整体成本节省。
+
 语义摘要是另一种操作，会改变措辞，可能改变含义。精确 codec 不做语义摘要。
 Controller 编写 brief 仍是单独提议的工作流，须验收保留要求与实际成果；更短的文字或
 provider cache 命中都不能替它通过。
diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py
index 9503d144b2..714d756976 100644
--- a/loopx/control_plane/quota/cli_projection.py
+++ b/loopx/control_plane/quota/cli_projection.py
@@ -23,9 +23,6 @@
     "quota_cli_vision_continuation_compaction_v0"
 )
 QUOTA_CLI_VISION_DETAIL_COMMAND = "quota should-run --include-detail vision"
-QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION = (
-    "quota_cli_replan_action_compaction_v0"
-)
 QUOTA_CLI_MONITOR_POLL_DECISION_COMPACTION_SCHEMA_VERSION = (
     "quota_cli_monitor_poll_decision_compaction_v0"
 )
@@ -94,16 +91,6 @@
 )
 
 
-def _json_content_key(value: Any) -> str:
-    """Compare original JSON values for sharing, including types and signed zero.
-
-    This is local serialization IO, not another quota or context-selection rule.
-    Python equality merges booleans with integers and integral floats; those
-    distinct source facts must not borrow each other's in-response owner.
-    """
-    return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False)
-
-
 def _compact_monitor_poll_interaction_contract(
     contract: Any,
 ) -> dict[str, Any] | None:
@@ -248,49 +235,6 @@ def _compact_vision_continuation_audit(
     return compact
 
 
-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
-    authoring_key = _json_content_key(writeback["vision_authoring"])
-    index = next(
-        (index for index, step in enumerate(steps)
-         if isinstance(step, dict) and step.get("kind") == "durable_writeback"
-         and _json_content_key(step.get("vision_authoring")) == authoring_key),
-        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_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"],
-        "projection_ref": reference,
-    }
-    return compact
-
-
 def _deduplicate_next_action_warning(
     warning: dict[str, Any],
     *,
@@ -441,6 +385,7 @@ def _promote_interaction_contract(payload: dict[str, Any]) -> dict[str, Any]:
 def _reference_context(
     payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True,
     goal_acceptance: bool = False, vision_audit: bool = False,
+    replan_authoring: bool = False,
     diagnostics: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """One coarse transport call for the existing typed context owner."""
@@ -452,7 +397,7 @@ def _reference_context(
     if not needs_context and (diagnostics is None or (
         diagnostics.get("agent_detail") is True and diagnostics.get("user_detail") is True
     )):
-        if not goal_acceptance and not vision_audit:
+        if not goal_acceptance and not vision_audit and not replan_authoring:
             return payload
         if goal_acceptance:
             # A reference can only be created or relocated when two literal
@@ -469,7 +414,7 @@ def contains_contract(value: Any, *path: str) -> bool:
                 contains_contract(payload, "autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"),
                 contains_contract(payload, "agent_todo_summary", "goal_acceptance_contract"),
             )
-            if sum(carriers) < 2 and not vision_audit:
+            if sum(carriers) < 2 and not vision_audit and not replan_authoring:
                 return payload
     from .turn_envelope import _turn_envelope_result
 
@@ -480,6 +425,8 @@ def contains_contract(value: Any, *path: str) -> bool:
         "todo_summary_projection", "planning_horizon", "autonomous_replan_obligation",
         "interaction_contract", "agent_lane_next_action", "selected_todo",
     ) if key in payload}
+    if replan_authoring and "replan_action_packet" in payload:
+        source["replan_action_packet"] = payload["replan_action_packet"]
     if (
         isinstance(contract, dict)
         and isinstance(agent, dict)
@@ -512,6 +459,7 @@ def contains_contract(value: Any, *path: str) -> bool:
         task_title=task_title, explore_guidance=explore_guidance,
         goal_acceptance=goal_acceptance,
         **({"vision_audit": True} if vision_audit else {}),
+        **({"replan_authoring": True} if replan_authoring else {}),
         **({"diagnostics": diagnostics} if diagnostics is not None else {}))
     return _apply_context_projection_changes(payload, result)
 
@@ -591,13 +539,9 @@ def compact_quota_should_run_cli_payload(
     vision_audit = payload.get("vision_continuation_audit")
     share_audit = not include_vision_detail and isinstance(vision_audit, dict)
     replan_action = payload.get("replan_action_packet")
-    if not include_vision_detail and isinstance(replan_action, dict):
-        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
+    writeback = replan_action.get("writeback_contract") if isinstance(replan_action, dict) else None
+    share_authoring = (not include_vision_detail and isinstance(writeback, dict)
+        and bool({"vision_authoring", "vision_authoring_ref"}.intersection(writeback)))
     if not include_todo_summary_detail:
         warning = payload.get("next_action_projection_warning")
         if isinstance(warning, dict):
@@ -629,7 +573,7 @@ def compact_quota_should_run_cli_payload(
         diagnostics = {"agent_detail": include_todo_summary_detail,
             "user_detail": include_user_todo_summary_detail}
     compact = _reference_context(compact, task_title=not include_todo_summary_detail,
-        goal_acceptance=True, vision_audit=share_audit,
+        goal_acceptance=True, vision_audit=share_audit, replan_authoring=share_authoring,
         explore_guidance=not include_vision_detail, diagnostics=diagnostics)
     if share_audit and isinstance(vision_audit, dict):
         compact = {**compact, "vision_continuation_audit": _compact_vision_continuation_audit(vision_audit)}
diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index 6b74d81944..77f22ab38f 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -57,7 +57,7 @@ def _turn_envelope_result(
 
     request = {"operation": operation, **params}
     if operation == "reference_context" and (
-        params.get("goal_acceptance") or params.get("vision_audit")
+        params.get("goal_acceptance") or params.get("vision_audit") or params.get("replan_authoring")
     ):
         tokens = source_number_tokens(params["payload"])
         if tokens:
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index f7a97458c2..52ec2676d1 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -228,6 +228,50 @@ function canonicalHash(value: unknown): string {
   return `sha256:${createHash("sha256").update(JSON.stringify(canonicalValue(value)), "utf8").digest("hex")}`;
 }
 
+function replanAuthoringContract(payload: JsonObject): JsonObject {
+  const contract = object(object(payload.replan_action_packet).writeback_contract);
+  if (!Object.hasOwn(contract, "vision_authoring_ref")) return contract;
+  const reference = contract.vision_authoring_ref;
+  const match = typeof reference === "string"
+    ? /^\$\.interaction_contract\.cli_channel\.settlement_plan\.ordered_steps\[(0|[1-9]\d*)\]\.vision_authoring$/.exec(reference) : null;
+  const steps = object(object(object(payload.interaction_contract).cli_channel).settlement_plan).ordered_steps;
+  const step = object(match && Array.isArray(steps) ? steps[Number(match[1])] : null);
+  if (Object.hasOwn(contract, "vision_authoring") || step.kind !== "durable_writeback" ||
+      Object.hasOwn(step, "vision_authoring_ref") || !Object.keys(object(step.vision_authoring)).length) {
+    throw new EffectRuntimeRequestError("Replan authoring reference must resolve to one complete inline durable-writeback owner");
+  }
+  delete contract.vision_authoring_ref;
+  return {...contract, vision_authoring: step.vision_authoring};
+}
+
+function referenceReplanAuthoring(payload: JsonObject,
+    identity = contextSourceIdentity(payload)): JsonObject {
+  const packet = object(payload.replan_action_packet);
+  const contract = object(packet.writeback_contract);
+  if (Object.hasOwn(contract, "vision_authoring_ref")) {
+    replanAuthoringContract(payload);
+    return payload;
+  }
+  const authoring = object(contract.vision_authoring);
+  const steps = object(object(object(payload.interaction_contract).cli_channel).settlement_plan).ordered_steps;
+  if (!Object.keys(authoring).length || !Array.isArray(steps)) return payload;
+  const encoded = JSON.stringify(canonicalValue(authoring));
+  const source = identity(authoring, ["replan_action_packet", "writeback_contract", "vision_authoring"]);
+  const index = steps.findIndex((value, index) => {
+    const step = object(value);
+    return step.kind === "durable_writeback" && !Object.hasOwn(step, "vision_authoring_ref") &&
+      jsonObject(step.vision_authoring) !== null &&
+      identity(step.vision_authoring, ["interaction_contract", "cli_channel", "settlement_plan",
+        "ordered_steps", index, "vision_authoring"]) === source;
+  });
+  if (index < 0) return payload;
+  const reference = `$.interaction_contract.cli_channel.settlement_plan.ordered_steps[${index}].vision_authoring`;
+  if (Buffer.byteLength(encoded, "utf8") <= Buffer.byteLength(reference, "utf8") + 8) return payload;
+  delete contract.vision_authoring;
+  return {...payload, replan_action_packet: {...packet, writeback_contract: {
+    ...contract, vision_authoring_ref: reference}}};
+}
+
 function resolveExploreGuidance(payload: JsonObject, context: JsonObject): JsonObject {
   if (!Array.isArray(context.sources)) return context;
   const steps = object(object(object(payload.interaction_contract).cli_channel).settlement_plan).ordered_steps;
@@ -647,6 +691,7 @@ function replanView(request: JsonObject): JsonObject {
     throw new EffectRuntimeRequestError("Replan read requires the original successful decision and matching source hash");
   }
   requireVisionAuditReferences(payload);
+  replanAuthoringContract(payload);
   const packet = object(payload.replan_action_packet);
   const obligation = object(payload.autonomous_replan_obligation);
   const context = object(obligation.replan_context);
@@ -675,7 +720,7 @@ function replanView(request: JsonObject): JsonObject {
     // The current nested review syntax belongs to authoring, not duplicate
     // decision evidence. Top-level required_fields cannot describe that shape.
     for (const key of ["decision_followthrough", "explore_basis", "decision_value_review"]) delete stride[key];
-    const contract = {...object(packet.writeback_contract)};
+    const contract = replanAuthoringContract(payload);
     if (contract.vision_authoring) {
       const authoring = {...object(contract.vision_authoring)};
       delete authoring.minimal_example;
@@ -760,6 +805,7 @@ function replanView(request: JsonObject): JsonObject {
 function replanActionPacket(payload: JsonObject, reads = replanReadCommands(payload)): JsonObject | null {
   const source = object(payload.replan_action_packet);
   if (Object.keys(source).length === 0) return null;
+  const writeback = replanAuthoringContract(payload);
   if (reads) return {...compactFields(source, ["schema_version", "decision", "obligation_id", "required_outcome", "allowed_terminal"]),
     review: reads.review, writeback: reads.writeback,
     guidance: "Read review state and applicable acceptance/evidence; choose the next work, then follow review.next for Todo selection or writeback. Reads grant no authority and require no acknowledgement."};
@@ -770,7 +816,6 @@ function replanActionPacket(payload: JsonObject, reads = replanReadCommands(payl
   // The typed replan owner supplies bounded instructions; do not truncate their
   // authority/stop qualifiers through generic diagnostic compaction.
   if (Array.isArray(source.planning_guidance)) compact.planning_guidance = source.planning_guidance;
-  const writeback = object(source.writeback_contract);
   if (writeback.vision_authoring) compact.writeback_contract = writeback;
   return Object.keys(compact).length > 0 ? compact : null;
 }
@@ -1660,6 +1705,17 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject);
     if (request.goal_acceptance === true) payload = referenceGoalAcceptance(payload,
       contextSourceIdentity(source, request.source_number_tokens));
+    if (request.replan_authoring === true) {
+      const shared = referenceReplanAuthoring(payload, contextSourceIdentity(source, request.source_number_tokens));
+      if (shared !== payload) {
+        const packet = object(shared.replan_action_packet);
+        payload = {...shared, replan_action_packet: {...packet, payload_compaction: {
+          schema_version: "quota_cli_replan_action_compaction_v0", mode: "compact_hot_path",
+          compacted_fields: ["writeback_contract.vision_authoring"],
+          projection_ref: object(packet.writeback_contract).vision_authoring_ref,
+        }}};
+      }
+    }
     if (request.task_title === true) payload = referenceCurrentTaskTitle(payload);
     if (request.explore_guidance === true) payload = referenceExploreGuidance(payload);
     if (request.refresh_result === true) payload = referenceExploreResultDelivery(payload, request.source_number_tokens);
diff --git a/tests/control_plane/canonical_authority_fixture.py b/tests/control_plane/canonical_authority_fixture.py
index 057d3f14ff..a64e360a7e 100644
--- a/tests/control_plane/canonical_authority_fixture.py
+++ b/tests/control_plane/canonical_authority_fixture.py
@@ -2,6 +2,8 @@
 
 This fixture runs the real FileAuthorityStore in a Node process. It deliberately
 does not claim that a shadow qualification can promote canonical authority.
+Resolve TypeScript modules from the imported package so source and installed
+wheel tests exercise the same Python and TypeScript build.
 """
 from __future__ import annotations
 
@@ -12,6 +14,8 @@
 import subprocess
 import tempfile
 
+import loopx
+
 
 def assert_todo_read_unchanged(current, previous):
     """Compare complete readback except validated evaluator/observation clocks."""
@@ -56,10 +60,10 @@ def without_resume_observation_time(value):
 
 
 def initialize_canonical_authority(runtime_root: Path, goal_id: str, projection: dict, *, state_path: Path, provider: str = "file") -> dict:
-    repository = Path(__file__).resolve().parents[2]
-    module = repository / "loopx/control_plane/coordination/file_authority_store.ts"
-    fence_module = repository / "loopx/control_plane/coordination/legacy_writer_fence.ts"
-    codec_module = repository / "loopx/control_plane/coordination/authority_store_codec.ts"
+    package = Path(loopx.__file__).resolve().parent
+    module = package / "control_plane/coordination/file_authority_store.ts"
+    fence_module = package / "control_plane/coordination/legacy_writer_fence.ts"
+    codec_module = package / "control_plane/coordination/authority_store_codec.ts"
     script = (
         f"import {{FileAuthorityStore}} from {json.dumps(module.as_uri())};"
         f"import {{selectLocalSqliteAuthority,openLocalAuthorityStore}} from {json.dumps((module.parent / 'local_authority_provider.ts').as_uri())};"
@@ -107,7 +111,7 @@ def advance_canonical_fixture_authority(
     Consumer fixtures may author a later synthetic state, but must not replace
     storage bytes, reset lineage, re-import Markdown or re-engage its fence.
     """
-    module = Path(__file__).resolve().parents[2] / "loopx/control_plane/coordination/local_authority_provider.ts"
+    module = Path(loopx.__file__).resolve().parent / "control_plane/coordination/local_authority_provider.ts"
     script = (
         f"import {{openLocalAuthorityStore}} from {json.dumps(module.as_uri())};"
         "let input='';for await(const chunk of process.stdin)input+=chunk;"
diff --git a/tests/control_plane/test_replan_authoring_owner.py b/tests/control_plane/test_replan_authoring_owner.py
new file mode 100644
index 0000000000..748453b4b2
--- /dev/null
+++ b/tests/control_plane/test_replan_authoring_owner.py
@@ -0,0 +1,212 @@
+"""One captured authoring owner survives display sharing and executable reads."""
+from copy import deepcopy
+import hashlib
+import json
+import subprocess
+import sys
+
+import pytest
+
+from loopx.control_plane.quota.cli_projection import _reference_context
+from loopx.control_plane.quota.turn_envelope import (
+    _turn_envelope_result, build_turn_envelope, prepare_context_delivery, restore_context_delivery,
+)
+from test_replan_host_context_projection import _quota_payload
+
+
+def packet():
+    source = _quota_payload()
+    authoring = {
+        "schema_version": "goal_vision_replan_contract_v0",
+        "rules": ["Retain the original requirements and negative evidence. " * 12,
+                  "Publication and spending still require their original authority."],
+        "provenance": {"ref": "fixture:original", "revision": 3},
+        "required": False,
+        "minimal_example": {"state": "vision_patch_proposed"},
+    }
+    source["replan_action_packet"]["writeback_contract"]["vision_authoring"] = deepcopy(authoring)
+    source["interaction_contract"]["cli_channel"]["settlement_plan"] = {
+        "identity": {"turn_instance_id": "original-turn"},
+        "ordered_steps": [{"kind": "validation"},
+                          {"kind": "durable_writeback", "vision_authoring": authoring}],
+    }
+    return source, authoring
+
+
+def shared_packet(source):
+    return _reference_context(source, task_title=False, explore_guidance=False, replan_authoring=True)
+
+
+def test_cli_compaction_shares_authoring_in_one_runtime_call(monkeypatch):
+    from loopx.control_plane.quota import turn_envelope
+    from loopx.control_plane.quota.cli_projection import compact_quota_should_run_cli_payload
+
+    source, authoring = packet()
+    original = deepcopy(source)
+    calls = []
+    invoke = turn_envelope.effect_runtime_projection_result
+
+    def observed(name, request):
+        calls.append((name, request))
+        return invoke(name, request)
+
+    monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", observed)
+    compact = compact_quota_should_run_cli_payload(source)
+    projection_calls = len(calls)
+    assert source == original
+    assert build_turn_envelope(compact)["replan_action_packet"]["writeback_contract"]["vision_authoring"] == authoring
+    assert projection_calls == 1
+
+
+def test_explicit_vision_detail_keeps_authoring_inline():
+    from loopx.control_plane.quota.cli_projection import compact_quota_should_run_cli_payload
+
+    source, authoring = packet()
+    compact = compact_quota_should_run_cli_payload(source, include_vision_detail=True)
+    writeback = compact["replan_action_packet"]["writeback_contract"]
+    assert writeback["vision_authoring"] == authoring
+    assert "vision_authoring_ref" not in writeback
+
+
+def source_digest(source):
+    # The fixture's keys use Python's declared Unicode ordering; the existing
+    # source binding uses JavaScript JSON number rendering (including 1.0).
+    raw = subprocess.check_output(["node", "-e",
+        "process.stdout.write(JSON.stringify(JSON.parse(require('node:fs').readFileSync(0,'utf8'))))"],
+        input=json.dumps(source, sort_keys=True, ensure_ascii=False).encode())
+    return "sha256:" + hashlib.sha256(raw).hexdigest()
+
+
+def cli_writer(directory, source, digest):
+    path = directory / "captured-authoring.json"
+    path.write_text(json.dumps(source))
+    return subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+        "quota", "replan-view", "--decision-file", str(path), "--source-hash", digest,
+        "--replan-stage", "writeback"], cwd=directory, capture_output=True, text=True)
+
+
+@pytest.mark.parametrize("reader", ["envelope", "staged", "cli"])
+def test_ordinary_and_captured_readers_resolve_the_same_complete_authoring(tmp_path, reader):
+    source, authoring = packet()
+    original = deepcopy(source)
+    shared = shared_packet(source)
+    assert source == original
+    assert shared_packet(shared) == shared
+    assert "vision_authoring" not in shared["replan_action_packet"]["writeback_contract"]
+    envelope = build_turn_envelope(shared)
+    if reader == "envelope":
+        actual = envelope["replan_action_packet"]["writeback_contract"]["vision_authoring"]
+    else:
+        digest = envelope["action_signature"]["source_decision_hash"]
+        if reader == "staged":
+            result = _turn_envelope_result("replan_view", payload=shared, source_hash=digest, stage="writeback")
+        else:
+            run = cli_writer(tmp_path, shared, digest)
+            assert run.returncode == 0, run.stdout + run.stderr
+            result = json.loads(run.stdout)
+        actual = result["writeback_contract"]["vision_authoring"]
+        authoring.pop("minimal_example")  # Existing staged authoring omits only this example.
+    assert actual == authoring
+
+
+@pytest.mark.parametrize("mutation", ["dangling", "wrong-step", "chain", "both", "malformed"])
+@pytest.mark.parametrize("captured", [False, True])
+def test_invalid_authoring_reference_is_refused_by_every_reader(tmp_path, mutation, captured):
+    source, _ = packet()
+    source = shared_packet(source)
+    if captured:
+        source["replan_read_path"] = "/synthetic/captured.json"
+    writeback = source["replan_action_packet"]["writeback_contract"]
+    owner = source["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"][1]
+    if mutation == "dangling":
+        writeback["vision_authoring_ref"] = writeback["vision_authoring_ref"].replace("[1]", "[99]")
+    elif mutation == "wrong-step":
+        owner["kind"] = "validation"
+    elif mutation == "chain":
+        owner.pop("vision_authoring")
+        owner["vision_authoring_ref"] = "$.replan_action_packet.writeback_contract.vision_authoring"
+    elif mutation == "both":
+        writeback["vision_authoring"] = {"rules": ["Conflicting obligation"]}
+    else:
+        writeback["vision_authoring_ref"] = {"path": "$.unknown"}
+    # The composition operation validates the reference before creating an
+    # envelope; the raw source hash is not a substitute for that validation.
+    with pytest.raises(RuntimeError, match="authoring reference"):
+        build_turn_envelope(source)
+    with pytest.raises(RuntimeError, match="authoring reference"):
+        shared_packet(source)
+    # A matching hash for the altered source must not conceal a bad reference.
+    digest = source_digest(source)
+    for stage in ("review", "writeback"):
+        with pytest.raises(RuntimeError, match="authoring reference"):
+            _turn_envelope_result("replan_view", payload=source, source_hash=digest, stage=stage)
+    run = cli_writer(tmp_path, source, digest)
+    assert run.returncode != 0
+    assert "authoring reference" in run.stdout + run.stderr
+
+
+@pytest.mark.parametrize("left,right", [(1, 1.0), (False, 0), (-0.0, 0),
+                                       (2**60, 2**60 + 1), (2**60, float(2**60))])
+def test_exact_numeric_facts_choose_the_matching_later_owner(left, right):
+    source, _ = packet()
+    writeback = source["replan_action_packet"]["writeback_contract"]
+    writeback["vision_authoring"]["future_numeric_fact"] = right
+    steps = source["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"]
+    steps[1]["vision_authoring"]["future_numeric_fact"] = left
+    steps.append({"kind": "durable_writeback", "vision_authoring": deepcopy(writeback["vision_authoring"])})
+    before = json.dumps(source)
+    shared = shared_packet(source)
+    assert json.dumps(source) == before
+    assert shared["replan_action_packet"]["writeback_contract"]["vision_authoring_ref"].endswith("[2].vision_authoring")
+    assert json.dumps(shared["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"]) == json.dumps(steps)
+
+
+def test_small_or_different_authoring_stays_inline():
+    source, _ = packet()
+    writeback = source["replan_action_packet"]["writeback_contract"]
+    owner = source["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"][1]
+    owner["vision_authoring"]["rules"].append("A distinct cancellation condition.")
+    assert shared_packet(source)["replan_action_packet"] == source["replan_action_packet"]
+    writeback["vision_authoring"] = owner["vision_authoring"] = {"required": True}
+    assert shared_packet(source)["replan_action_packet"] == source["replan_action_packet"]
+
+
+def test_full_delta_and_missing_baseline_keep_the_exact_authoring():
+    source, authoring = packet()
+    before = shared_packet(source)
+    source["replan_action_packet"]["uncovered_frontier"] = ["New evidence to assess"]
+    after = shared_packet(source)
+    scope = dict(goal_id="fixture", agent_id="worker", session_id="session", context_epoch="one")
+    baseline = prepare_context_delivery(before, scope=scope, capture_id="one", surface="quota_packet")
+    full = prepare_context_delivery(after, scope=scope, capture_id="two", surface="quota_packet")
+    delta = prepare_context_delivery(after, scope=scope, capture_id="two", surface="quota_packet",
+        baseline=baseline, retained={"complete": True, "target": baseline["target"]})
+    assert delta["mode"] == "delta"
+    for retained in (baseline, None):
+        restored = restore_context_delivery(delta, baseline=retained, current=full)["context"]
+        assert restored == after
+        assert build_turn_envelope(restored)["replan_action_packet"]["writeback_contract"]["vision_authoring"] == authoring
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_capture_authoring_read_does_not_repeat_work_or_spend(tmp_path, monkeypatch, provider):
+    from test_turn_replan_writeback import _fixture
+    from test_quota_settlement_cli import _run_cli, _spend_run_count, GOAL_ID, AGENT_ID, SELECTED_REPLAN_TODO_ID
+
+    project, runtime, registry, state, turn = _fixture(tmp_path, monkeypatch, provider)
+    capture = tmp_path / "decision"
+    code, result = _run_cli(registry, runtime, "quota", "should-run", "--codex-app",
+        "--context-delivery", "off", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--todo-id", SELECTED_REPLAN_TODO_ID, "--turn-instance-id", turn,
+        "--decision-output-dir", str(capture), cwd=project)
+    assert code == 0, result
+    source = json.loads((capture / "decision.json").read_text())
+    authoring = deepcopy(source["replan_action_packet"]["writeback_contract"]["vision_authoring"])
+    authoring.pop("minimal_example", None)
+    shared = shared_packet(source)
+    before = state.read_bytes()
+    run = cli_writer(project, shared, source_digest(shared))
+    assert run.returncode == 0, run.stdout + run.stderr
+    assert json.loads(run.stdout)["writeback_contract"]["vision_authoring"] == authoring
+    assert state.read_bytes() == before
+    assert _spend_run_count(runtime) == 0
diff --git a/tests/control_plane/test_turn_replan_writeback.py b/tests/control_plane/test_turn_replan_writeback.py
index aecb510de2..23b4b30b86 100644
--- a/tests/control_plane/test_turn_replan_writeback.py
+++ b/tests/control_plane/test_turn_replan_writeback.py
@@ -50,16 +50,16 @@ def _fixture(tmp_path, monkeypatch, provider):
     return project, runtime, registry, state, turn
 
 
-def _runner(root, candidate, *, follow=False):
+def _runner(root, candidate, *, follow=False, reviewed=False):
     # Exercise semantic writeback after the existing execution-profile evidence
     # gate; the independent command below checks these actual Host artifacts.
     candidate = {**candidate, "work_evidence": ["artifact:answer.txt" if follow else "artifact:plan.json"],
                  "validation_evidence": ["check:independent-command"]}
     runner = root / ("follow.py" if follow else "host.py")
     runner.write_text(
-        "import json\nfrom pathlib import Path\n"
+        "import json, shlex, subprocess\nfrom pathlib import Path\n"
         f"candidate = {candidate!r}\n"
-        "def run_dsh_turn(*, workspace, **kwargs):\n"
+        "def run_dsh_turn(*, workspace, prompt, **kwargs):\n"
         "    root = Path(workspace)\n"
         "    counter = root / 'host-count.txt'\n"
         "    counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n"
@@ -69,6 +69,27 @@ def _runner(root, candidate, *, follow=False):
             if follow else
             "    (root / 'plan.json').write_text(json.dumps({'inputs': [17, 25]}))\n"
         )
+        + (
+            # Consume the structured authority actually delivered to this Host,
+            # including a required captured read; do not manufacture its refs.
+            "    authority, _ = json.JSONDecoder().raw_decode(prompt.split('Turn authority JSON:\\n', 1)[1])\n"
+            "    packet = authority['replan_action_packet']\n"
+            "    if 'stride_control' not in packet:\n"
+            "        packet = json.loads(subprocess.check_output(shlex.split(packet['writeback']), text=True))\n"
+            "    control = packet['stride_control']\n"
+            "    vision = json.loads(candidate['agent_vision_json'])\n"
+            "    decision = {**control['minimal_example'], 'basis': 'uncertain_evidence', 'action': 'verify',\n"
+            "        'axes': {\n"
+            "            'work_slice': {'plan': 'Validate the complete plan.', 'reason': 'The plan is new evidence.'},\n"
+            "            'route_distance': {'plan': 'Retain the current task and route.', 'reason': 'The original task remains applicable.'},\n"
+            "            'validation_scope': {'plan': 'Check the plan, then consume it.', 'reason': 'An artifact alone does not prove use.'}},\n"
+            "        'next_probe': {'scope': 'One plan and its consumer.', 'prediction': 'The retained task can use the plan.',\n"
+            "            'minimum_evidence': 'Independent validation and the consumer result.',\n"
+            "            'adoption_criterion': 'The original expected result is correct.', 'cost_bound': 'Two bounded fixture operations.'}}\n"
+            "    vision['stride_decision'] = decision\n"
+            "    candidate['agent_vision_json'] = json.dumps(vision)\n"
+            if reviewed else ""
+        )
         + "    return json.dumps(candidate)\n",
         encoding="utf-8",
     )
@@ -120,7 +141,8 @@ def test_semantic_refusal_preserves_task_context(tmp_path, monkeypatch, provider
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_accepted_replan_retains_context_and_drives_next_work(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("reviewed", [False, True])
+def test_replan_retains_context_and_requires_shared_authoring(tmp_path, monkeypatch, provider, reviewed):
     project, runtime, registry, _state, turn = _fixture(tmp_path, monkeypatch, provider)
     before = _todo(registry, runtime)
     vision = {"schema_version": "goal_vision_replan_contract_v0", "state": "active",
@@ -133,8 +155,13 @@ def test_accepted_replan_retains_context_and_drives_next_work(tmp_path, monkeypa
     candidate = {"result_kind": "replan_required", "classification": "fixture_verified_plan",
         "summary": "A verified plan is ready.", "next_action": "Execute the verified plan.",
         "path_delta_mode": "material_replan", "agent_vision_json": json.dumps(vision)}
-    runner = _runner(tmp_path, candidate)
+    runner = _runner(tmp_path, candidate, reviewed=reviewed)
     rc, accepted = _run_turn(project, runtime, registry, runner, turn=turn)
+    if not reviewed:
+        assert rc == 1 and "stride_decision" in accepted["error"], accepted
+        assert _todo(registry, runtime) == before
+        assert _classification_count(runtime, "quota_slot_spent") == 0
+        return
     assert rc == 0 and accepted["status"] == "committed", accepted
     assert _todo(registry, runtime) == before
     rc, replay = _run_turn(project, runtime, registry, runner, resume=accepted["resume_turn_key"])