fix(turn): preserve replan requirements at the shared Host

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md

@@ -482,6 +482,17 @@ File/SQLite CLI and a protocol-only DSH subprocess. Keep the 8 KiB warning and
 count packet plus necessary detail/source IO before choosing a default. Live
 model behavior, installed adoption and the three acceptances below remain open.
 
+A bound native replan exposed a further shared Host adapter omission: it kept
+the generic action but dropped the signed replan packet and its evidence route.
+The DSH/cloud adapter now forwards both, retaining the existing full Goal read
+and same-capture replan context. File/SQLite selection-to-replan qualification
+checks actual DSH provider input with a protocol-only runner, signed mutation
+refusal and unchanged source/state; it does not invoke a model. The outer Turn
+executor retains writeback/quota/scheduler ownership. Ordinary requests retain
+their previous input. This qualifies that transport boundary only; installed
+adoption, subsequent model decisions, complete costs and the three acceptances
+below remain open.
+
 Current source-path audit for the first convergence slice:
 
 | Execution fact | Existing owner / projection | Remaining boundary |

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md

@@ -337,6 +337,14 @@ File/SQLite CLI 和无模型 DSH 子进程覆盖开启／关闭、历史引用
 保持；选择默认形态前须计算 packet 与必要补读／源 IO。真实模型行为、安装采用及以下
 三项验收仍未完成。
 
+真实绑定的重规划暴露了共享 Host adapter 的另一处遗漏：保留了泛化动作，却丢失
+签名 replan packet 及其证据路由。DSH/cloud adapter 现在同时传递两者，保留既有
+完整 Goal 必读和同次捕获的 replan context。File/SQLite 从选择到重规划的资格验证
+通过仅使用协议 fixture 的 runner 核对实际 DSH provider 输入、签名篡改拒绝及来源／
+状态未变，不调用模型。写回、quota 和 scheduler 仍归外层 Turn executor；普通请求
+保持原输入。本切片只验证该传输边界，安装采用、后续模型决策、完整成本及以下三项
+验收仍未完成。
+
 第一批收敛的当前源码路径核对：
 
 | 执行事实 | 既有 owner／投影 | 保留的边界 |

docs/reference/protocols/turn-envelope-v0.md

@@ -148,6 +148,22 @@ guard. Missing or invalid evidence holds dependent planning while independent
 work retains its authority. Capture validation and later mutation checks remain
 with their existing owners; prompt delivery does not prove model adoption.
 
+For admitted replans, the shared DSH/cloud adapter also carries the signed
+`replan_action_packet` and the original capture route/signature. Read that saved
+decision's `autonomous_replan_obligation.replan_context` before choosing the
+semantic outcome, preserving the existing full Goal read. Missing or invalid
+original evidence holds dependent replanning; never rerun admission to replace
+it. Return the typed candidate rather than executing the packet's writeback
+commands: the outer Turn executor owns state, quota and scheduler effects.
+This repairs an adapter omission without adding an evidence-read receipt or
+changing non-replan requests.
+
+中文：已准入的重规划通过共享 DSH/cloud adapter 传递签名 action packet，以及原捕获
+路由和签名。选择语义结果前从原决策读取既有 replan context，保留完整 Goal 必读；
+来源缺失或无效时暂停依赖它的重规划，不能重跑准入替代。返回 typed candidate，
+不执行 packet 内写回命令；状态、quota 和 scheduler 效果仍归外层 Turn executor。
+本修复不增加补读回执，也不改变非重规划请求。
+
 The unchanged public multi-subagent CLI fixture grows by 1,147 JSON characters
 and UTF-8 bytes when this context replaces its reference. Its differential
 allows 1,280 only for that reference-to-inline transition (133 headroom);

loopx/control_plane/turn_driver/host_candidate.py

@@ -113,12 +113,15 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
     context = _mapping(envelope.get("agent_context"))
     if context:
         authority["agent_context"] = context
-        if context.get("detail_ref"):
-            # Historical signed references still need their original evidence
-            # route and hash. Native Turn capture validation owns its binding;
-            # this adapter neither reads the file nor re-enters admission.
-            authority["detail_ref"] = _mapping(envelope.get("detail_ref"))
-            authority["action_signature"] = signature
+    replan = _mapping(envelope.get("replan_action_packet"))
+    if replan:
+        authority["replan_action_packet"] = replan
+    if context.get("detail_ref") or replan:
+        # Replan evidence and historical capability references need the saved
+        # observation's route and hash. Native Turn capture validation owns its
+        # binding; this adapter neither reads the file nor re-enters admission.
+        authority["detail_ref"] = _mapping(envelope.get("detail_ref"))
+        authority["action_signature"] = signature
     return authority
 
 
@@ -157,6 +160,18 @@ def render_prompt(authority: Mapping[str, Any]) -> str:
         "independent work retains its existing authority.\n"
         if authority.get("agent_context") else ""
     )
+    replan_instruction = (
+        "Use replan_action_packet to choose one permitted semantic outcome. "
+        "Before choosing it, read autonomous_replan_obligation.replan_context "
+        "through detail_ref.full_decision from the saved original observation "
+        "identified by detail_ref.captured_decision; check Goal/Agent/Turn and "
+        "action_signature.source_decision_hash. Never re-enter quota admission "
+        "for a detail read. Missing or invalid original evidence holds dependent "
+        "replanning; independent work retains its existing authority. "
+        "Do not execute its writeback commands; the outer Turn executor owns "
+        "state, quota and scheduler effects. Guidance grants no permissions.\n"
+        if authority.get("replan_action_packet") else ""
+    )
     return (
         "You are executing one bounded LoopX-governed work segment.\n"
         "The JSON below is the complete host authority for this Turn. Execute "
@@ -167,6 +182,7 @@ def render_prompt(authority: Mapping[str, Any]) -> str:
         "Do not infer authority from other prose.\n\n"
         f"{context_instruction}"
         f"{planning_context_instruction}"
+        f"{replan_instruction}"
         f"Turn authority JSON:\n{authority_json}\n\n"
         "When finished, return only one JSON object (no Markdown fence) with "
         "these public-safe fields:\n"

tests/control_plane/test_turn_host_context_delivery.py

@@ -11,10 +11,15 @@
 
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 from loopx.control_plane.quota.turn_envelope import build_turn_envelope, turn_envelope_action_signature_document
+from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan
 from loopx.control_plane.turn_driver.host_candidate import _canonical_hash
 from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
 from tests.control_plane.test_agent_context import POLICY
-from tests.control_plane.test_quota_settlement_cli import AGENT_ID, GOAL_ID, TODO_ID, _run_cli, _write_fixture
+from tests.control_plane.test_quota_settlement_cli import (
+    AGENT_ID, GOAL_ID, TODO_ID, SELECTED_REPLAN_TODO_ID,
+    _bind_selected_replan_guard, _configure_selected_todo_replan_fixture,
+    _initialize_git_checkout, _run_cli, _write_fixture,
+)
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
@@ -107,3 +112,86 @@ def test_captured_context_reaches_actual_host_without_readmission(tmp_path, monk
         assert "action_signature" not in authority
     assert (capture / "decision.json").read_bytes() == original
     assert {p: p.read_bytes() for p in runtime.rglob("*") if p.is_file()} == before
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_bound_replan_reaches_actual_host_with_original_evidence(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    _configure_selected_todo_replan_fixture(project, registry)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    stop = "Retain every accepted outcome; stop before modifying production data."
+    state.write_text(state.read_text() + f"\n## Accepted scope and stop\n\n{stop}\n")
+    goal_bytes = state.read_bytes()
+    _initialize_git_checkout(project)
+    code, listed = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID)
+    assert code == 0, listed
+    initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection(
+        goal_id=GOAL_ID, handoff_mode="soft_claim", todos=listed["todos"],
+    ), state_path=state, provider=provider)
+    turn = "host-replan-turn"
+    code, initial = _run_cli(registry, runtime, "quota", "should-run",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", turn,
+        "--codex-app", "--scan-path", str(project), cwd=project)
+    assert code == 0 and initial["decision"] == "autonomous_replan_required", initial
+    _bind_selected_replan_guard(registry, runtime, project, turn)
+    capture = tmp_path / "capture"
+    code, envelope = _run_cli(registry, runtime, "quota", "should-run",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--todo-id", SELECTED_REPLAN_TODO_ID, "--turn-instance-id", turn,
+        "--codex-app", "--turn-envelope", "--decision-output-dir", str(capture), cwd=project)
+    assert code == 0, envelope
+    plan = build_loopx_turn_plan(envelope, host="dsh", execution_mode="isolated-headless",
+                                 turn_instance_id=turn)
+    assert plan["route"]["kind"] == "replan_required"
+    assert plan["route"]["would_invoke_host"] is True
+    original = (capture / "decision.json").read_bytes()
+    source = json.loads(original)
+    assert source["heartbeat_receipt"]["settlement_identity"]["todo_id"] == SELECTED_REPLAN_TODO_ID
+
+    runner = tmp_path / "runner.py"
+    runner.write_text(
+        "import json\nfrom pathlib import Path\n"
+        "def run_dsh_turn(*, prompt, workspace, **kwargs):\n"
+        "    Path(workspace, 'received.txt').write_text(prompt)\n"
+        "    return json.dumps({'result_kind':'wait','summary':'Replan input recorded.'})\n"
+    )
+    command = [sys.executable, "-m", "loopx.dsh_goal_mode", "--workspace", str(project),
+               "--dsh-home", str(tmp_path / "host-home"), "--dsh-runner", str(runner)]
+    request = {"schema_version": "loopx_turn_host_request_v0", "turn_key": plan["transaction"]["turn_key"],
+               "route": "replan_required", "turn_envelope": envelope}
+    before = {p: p.read_bytes() for p in runtime.rglob("*") if p.is_file()}
+    result = subprocess.run(command, input=json.dumps(request), text=True, capture_output=True, timeout=30)
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert json.loads(result.stdout)["result_kind"] == "wait"
+    prompt_path = project / "received.txt"
+    prompt = prompt_path.read_text()
+    authority = json.loads(prompt.split("Turn authority JSON:\n", 1)[1].split("\n\nWhen finished", 1)[0])
+    # Compare the actual provider input with the full native source, not a
+    # second reduced projection. Replan guidance and outcome choices are signed.
+    assert authority["replan_action_packet"] == source["replan_action_packet"]
+    assert authority["action_signature"] == envelope["action_signature"]
+    detail = authority["detail_ref"]
+    assert detail == envelope["detail_ref"]
+    captured = detail["captured_decision"]
+    assert (captured["goal_id"], captured["agent_id"], captured["turn_instance_id"]) == (GOAL_ID, AGENT_ID, turn)
+    read = subprocess.run(shlex.split(detail["full_decision"]), capture_output=True, check=True).stdout
+    assert read == original
+    assert build_turn_envelope(json.loads(read))["action_signature"]["source_decision_hash"] == authority["action_signature"]["source_decision_hash"]
+    context = json.loads(read)["autonomous_replan_obligation"]["replan_context"]
+    assert context["obligation_id"] == authority["replan_action_packet"]["obligation_id"]
+    goal_read = next(r for r in authority["required_reads"] if r["source"] == "goal_state")
+    delivered_goal = subprocess.run(shlex.split(goal_read["command"]), capture_output=True, check=True).stdout
+    assert delivered_goal == goal_bytes and stop.encode() in delivered_goal
+    assert "replan_action_packet" in prompt and "replan_context" in prompt
+    assert "Never re-enter quota admission" in prompt
+    assert "Do not execute its writeback commands" in prompt
+    for field in ("obligation_id", "planning_guidance", "uncovered_frontier"):
+        invalid = deepcopy(request)
+        invalid["turn_envelope"]["replan_action_packet"][field] = "changed after signing"
+        rejected = subprocess.run(command, input=json.dumps(invalid), text=True, capture_output=True, timeout=30)
+        assert rejected.returncode != 0 and "action signature" in rejected.stderr
+        assert prompt_path.read_text() == prompt
+    assert (capture / "decision.json").read_bytes() == original
+    assert state.read_bytes() == goal_bytes
+    assert {p: p.read_bytes() for p in runtime.rglob("*") if p.is_file()} == before