refactor(turn): derive native launch bindings from settlement identity

docs/architecture/rfcs/controller-executor-unified-contract-v0.md

@@ -991,6 +991,13 @@ and retirement, and adoption/full-cost qualification where each boundary is
 independently reviewable. Record dependencies in the existing tasks and PRs.
 Splitting delivery does not close C1 or waive §9's original quota-baseline test.
 
+Native launch rechecks use the original `SettlementIdentity` as their sole
+Todo/obligation binding. Both `codex-cli` and `codex-app-server` call the existing
+`current_turn_host_launch_decision` owner; an observed replan packet is context,
+not an additional binding argument. This transport simplification can ship before
+wire cutover. Qualify real launch, refusal and replay against the same original
+identity; retain the current typed eligibility, Stop and settlement owners.
+
 The first **cutover** slice pairs the complete field/clause migration with a
 real single-session producer and consumer replacement. Start at the
 shared quota emission boundary and trace both attached heartbeat delivery and

docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md

@@ -750,6 +750,12 @@ owner。列明后续切换消费者；不能添加无人使用的基础设施或
 可按独立可审查的边界拆分，在既有任务和 PR 中记录依赖。拆分交付不代表 C1 完成，
 也不豁免第 9 节相对原 quota 基线的验证。
 
+原生启动重查只从原 `SettlementIdentity` 取得 Todo/obligation 绑定。
+`codex-cli` 和 `codex-app-server` 都调用既有
+`current_turn_host_launch_decision` owner；观察到的 replan packet 是上下文，
+不再作为另一份绑定参数。这项运输简化可以先于 wire 切换交付。沿相同原身份验证真实
+启动、拒绝与重放，保留既有 typed 资格、Stop 和结算 owner。
+
 首个**切换**片必须把完整字段/条款迁移与真实单会话 producer、consumer 替换一起交付。
 从共享 quota emission 边界开始，同时追踪 attached heartbeat 交付与 Turn adapter，
 保留各自真实宿主归属。不能新增无人使用的第三个 serializer，而两个活跃 writer 继续

loopx/cli_commands/turn_decision.py

@@ -349,7 +349,7 @@ def build_fresh_turn_decision_owner(
 
 def current_turn_host_launch_decision(
     args: argparse.Namespace, *, registry_path: Path, runtime_root: Path,
-    identity: SettlementIdentity, replan_obligation_id: str | None,
+    identity: SettlementIdentity,
     planned_effective_action: object,
     scheduler_execution_context: Mapping[str, Any],
     operator_inbox_urgency_projector: Callable[..., dict[str, Any]],
@@ -361,6 +361,8 @@ def current_turn_host_launch_decision(
     The caller supplies the real receipt identity; a Host session or turn_key
     cannot reconstruct it. This read neither creates another receipt nor
     substitutes a newly selected task into the already planned Host request.
+    An observed replan packet is context, not a second settlement binding.
+    Both native callers transport the obligation recorded in this identity.
     """
     current = build_live_quota_should_run_decision(
         collect_turn_status_payload(args, registry_path=registry_path,
@@ -373,7 +375,7 @@ def current_turn_host_launch_decision(
         operator_inbox_urgency_projector=operator_inbox_urgency_projector,
         bounded_research_frontier_projector=project_live_explore_composition_frontier,
         receipt_bound_todo_id=identity.todo_id or None,
-        receipt_bound_replan_obligation_id=replan_obligation_id,
+        receipt_bound_replan_obligation_id=identity.replan_obligation_id,
         turn_instance_id=identity.turn_instance_id,
         turn_start_hook_dispatch=turn_start_hook_dispatch,
         workspace_path=Path(args.project).expanduser().resolve(), goal_ref=goal_ref,

loopx/cli_commands/turn_run_once.py

@@ -803,7 +803,6 @@ def scheduler(_spend_payload: dict[str, object]) -> dict[str, object]:
             def native_launch_decision(current_registry: Path) -> Mapping[str, Any]:
                 return current_turn_host_launch_decision(args, registry_path=current_registry,
                     runtime_root=runtime_root, identity=settlement_identity,
-                    replan_obligation_id=replan_obligation_id,
                     planned_effective_action=payload["turn_envelope"].get("effective_action"),
                     scheduler_execution_context=payload.get("scheduler_execution_context") or {},
                     operator_inbox_urgency_projector=operator_inbox_urgency_projector,
@@ -827,7 +826,7 @@ def current_launch_decision(current_registry: Path) -> Mapping[str, Any]:
                 # when its mirror still advertises old quota or health.
                 return current_turn_host_launch_decision(
                     args, registry_path=current_registry, runtime_root=runtime_root,
-                    identity=settlement_identity, replan_obligation_id=replan_obligation_id,
+                    identity=settlement_identity,
                     planned_effective_action=payload["turn_envelope"].get("effective_action"),
                     scheduler_execution_context=payload.get("scheduler_execution_context") or {},
                     operator_inbox_urgency_projector=operator_inbox_urgency_projector,

tests/control_plane/test_turn_launch_identity.py

@@ -0,0 +1,124 @@
+"""Native launch preserves the admitted binding through real CLI and stores."""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from host_process_fixture import write_python_host_fixture
+from test_replan_successor_guard_reentry import AGENT, GOAL, _fixture
+
+
+def _host(root: Path, *, material: bool) -> Path:
+    # This is a deterministic process transport, not a model or task evaluator.
+    # The material arm writes a distinct artifact checked by the outer validator;
+    # the replan arm ends the iteration and earns no writeback or debit.
+    return write_python_host_fixture(root / "native-host", f'''#!/usr/bin/env python3
+import json, pathlib, sys
+request = json.loads(sys.stdin.read().split("Turn request:\\n", 1)[1])
+with pathlib.Path("host-requests.jsonl").open("a") as stream:
+    stream.write(json.dumps(request) + "\\n")
+material = {material!r}
+if material:
+    pathlib.Path("checked-source.txt").write_text("independently checked source\\n")
+print(json.dumps({{"type": "thread.started", "thread_id": "fixture-launch-session"}}), flush=True)
+result = {{
+    "schema_version": "loopx_turn_result_v0", "turn_key": request["turn_key"],
+    "result_kind": "validated_progress" if material else "iteration_failed",
+    "completed_phases": ["host_execute", "typed_result"],
+    "classification": "source_checked" if material else "",
+    "recommended_action": "Retain the original source criteria." if material else "",
+    "next_action": "Continue independent source validation." if material else "",
+    "delivery_batch_scale": "single_surface" if material else "",
+    "delivery_outcome": "outcome_progress" if material else "",
+    "path_delta_mode": "unchanged" if material else "", "agent_vision_json": "",
+    "vision_unchanged_reason": "Original source criteria still apply." if material else "",
+    "progress_observation_json": "",
+    "summary": "Source checked." if material else "Source remains unverified.",
+}}
+if material:
+    result.update(work_evidence=["artifact:checked-source"],
+                  validation_evidence=["check:independent-source-validator"])
+args = sys.argv[1:]
+pathlib.Path(args[args.index("--output-last-message") + 1]).write_text(json.dumps(result))
+''')
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("binding", ["todo", "autonomous_replan"])
+def test_native_cli_uses_original_binding_and_replays_without_new_effects(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, binding: str,
+) -> None:
+    call, runtime, _ = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    todo_id = None
+    declared_text = None
+    if binding == "todo":
+        todo_id = call("todo", "add", "--goal-id", GOAL, "--role", "agent",
+            "--claimed-by", AGENT, "--text", "Independently check the original source.",
+            "--task-class", "advancement_task", "--action-kind", "validate")["todo_id"]
+        declared_text = call("todo", "list", "--goal-id", GOAL,
+            "--todo-id", todo_id)["todo"]["text"]
+    project = tmp_path / "project"
+    executable = _host(tmp_path, material=binding == "todo")
+    decision = ["--goal-id", GOAL, "--agent-id", AGENT, "--host", "codex-cli",
+        "--execution-mode", "isolated-headless", "--scheduler-owner", "outer_controller",
+        "--iteration-context", "fresh", "--session-scope", "agent"]
+    if todo_id:
+        decision += ["--todo-id", todo_id]
+    common = [*decision, "--project", str(project), "--codex-bin", str(executable),
+        "--no-global-sync"]
+    if todo_id:
+        common += ["--validation-command-json", json.dumps([
+            sys.executable, "-c", "from pathlib import Path; "
+            "assert Path('checked-source.txt').read_text() == 'independently checked source\\n'",
+        ])]
+    turn = "native-launch-binding"
+    preview = call("turn", "plan", *decision, "--turn-instance-id", turn,
+        "--include-transaction-detail")
+    assert preview["route"]["kind"] == ("ready_for_host" if todo_id else "replan_required")
+    obligation_id = None
+    subject = todo_id
+    if not todo_id:
+        envelope = preview["turn_envelope"]
+        obligation_id = envelope["replan_action_packet"]["obligation_id"]
+        assert obligation_id.startswith("replan-")
+        subject = f"autonomous_replan:{obligation_id}"
+    expected_effect = f"{GOAL}:{AGENT}:{subject}:{turn}"
+    assert not (project / "host-requests.jsonl").exists()
+
+    result = call("turn", "run-once", *common, "--turn-instance-id", turn, "--execute")
+    assert result["effects"]["host_invoked"] is True, result
+    assert result["receipt"]["settlement_effect_id"] == expected_effect
+    lineage = result["receipt"]["lineage"]
+    assert lineage["todo_id"] == (todo_id or "")
+    assert lineage.get("replan_obligation_id") == obligation_id
+    assert result["effects"]["state_written"] is bool(todo_id)
+    assert result["quota_slot_spend_count"] == int(bool(todo_id))
+    if todo_id:
+        assert result["status"] == "committed" and result["validation"]["ok"] is True
+        current = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo_id)
+        assert current["todo"]["text"] == declared_text
+    else:
+        assert result["status"] == "stopped" and result["result_kind"] == "iteration_failed"
+        assert not (project / "checked-source.txt").exists()
+
+    requests = (project / "host-requests.jsonl").read_bytes()
+    sent = json.loads(requests)
+    assert sent["turn_key"] == result["receipt"]["turn_key"]
+    captures = {path: path.read_bytes() for path in runtime.rglob("decisions/*.json")}
+    assert captures
+    replay_args = list(common)
+    if todo_id:
+        rejected = call("turn", "run-once", *common, "--resume-turn-key", sent["turn_key"],
+            "--execute", expected_code=1)
+        assert "cannot retarget" in rejected["error"] and not any(rejected["effects"].values())
+        index = replay_args.index("--todo-id")
+        del replay_args[index:index + 2]
+    replay = call("turn", "run-once", *replay_args, "--resume-turn-key", sent["turn_key"],
+        "--execute")
+    assert replay["replayed"] is True
+    assert not any(replay["effects"].values())
+    assert (project / "host-requests.jsonl").read_bytes() == requests
+    assert all(path.read_bytes() == original for path, original in captures.items())