refactor(turn): unify plan, host input and recovery on one source

benchmark/runtime/checkpoint.py

@@ -274,7 +274,7 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
         view_path.write_text(guard_run.stdout, encoding="utf-8")
         view_path.chmod(0o600)
         reading = (f"Read the complete CLI view at {view_path} before work, then follow its interaction_contract "
-            "and the next step in the replan review when present. ")
+            "and replan review/writeback reads when present. ")
     preloaded_body = ("Host checkpoint preflight has already executed this wake's original quota guard once. "
         + reading +
         "Do not repeat the initial guard below; use its selection or recovery command when required.\n\n" + body)

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

@@ -549,13 +549,31 @@ are reconciled. Rollback likewise separates outstanding work from new admission.
 A version transition does not rewrite saved prompts, journals or source hashes
 as if they had always used the new contract.
 
-The first C1 PR must include the two production composition callsites above and
+The first cross-entrypoint C1 cutover must include the two production composition callsites above and
 at least one complete producer-to-final-input journey, with an explicit list of
 remaining readers. Qualifying that bounded slice does not enable the new writer
 for an unqualified host or close C1. The final cutover removes active format
 choice only after all supported readers qualify. Temporary decoders have named
 callers and removal conditions; they are not user-facing compatibility views.
 
+A Turn lifecycle extraction may be reviewed separately under the parent C1
+proposal when it includes planning, final Host input, capture publication,
+journal inspection, original-session recovery and settlement consumers together.
+The migrated Turn uses one complete source and its checked readers; the ordinary
+quota writer and thin-heartbeat presentation retain their existing contract in
+this staged boundary. Shared projection changes must preserve those unmigrated
+callers. The historical v0 decoder serves recorded plans, requests and journals;
+it cannot rescue an invalid supplied v1 source. Its removal still depends on
+reconciling outstanding work and the supported recovery window.
+
+This extraction is partial C1 delivery, not qualification of the cross-entrypoint
+cutover or a context/cost improvement. The parent retains ordinary quota and
+heartbeat migration, final-input/material-read coverage, all-host configuration,
+and the original single-session outcome and complete-cost comparison. Review
+must include matched baseline/head real-provider outcomes, rejection sensitivity
+and packaged recovery. Adoption reduces the parent's diff only after the slice
+lands on main and main is integrated into the parent; opening a PR does not.
+
 For each normal, selection, replan, wait, recovery and capability-off fixture,
 check independently authored obligations at **three boundaries**: composed
 contract, reconstructed delivery, and final model input plus required material
@@ -591,6 +609,35 @@ readback and downgrade behavior; retain the old reader until the supported
 retention/upgrade boundary is satisfied. No destructive migration is authorized
 by this document alone.
 
+#### Turn consumer preparation and instruction parity
+
+The separate Turn proposal keeps the ordinary quota producer on its recorded
+contract; the parent retains the broader quota/heartbeat cutover. Tests of an
+unmigrated producer must preserve that boundary. Current Turn planning consumes
+its complete source directly and publishes its own immutable capture; pairing a
+newly composed source with an older executor envelope is rejected.
+
+Native selected-task execution prepares its ordinary managed lease before the
+kernel delivery guard and reuses that proof during post-model validation. A
+claim label alone is insufficient, and another execution's lease remains
+unavailable. Source checks precede the independent task argv; rejection and
+original-result recovery retain the original session and once-only effects.
+
+Modern Codex and structured Host instructions share the existing audit read
+rule: read a declared audit before judging Goal acceptance or proposing
+closeout, including when the Goal stays open. Notification policy does not grant
+or cancel execution authority; supplied delivery, attempt, quiet-noop, execution
+and orchestration constraints still apply. Core retains state, quota and
+scheduler effects. A complete contract can declare a pending original material
+read: publication does not fulfill it, and tests must execute its reader and
+preserve the original body and stops.
+
+The disposable research-team launcher and initializer locate the actually loaded
+Core together, including an isolated wheel, while retaining exclusive runtime
+creation. Source and packaged consumer checks are distinct from installed App
+adoption, live optional-provider/model behavior and original-quota complete-cost
+benefit. These repairs do not close C1 or authorize rollout or merge.
+
 ### C1 baseline boundary and reuse inventory
 
 The baseline is the **original caller-visible quota delivery**, not the raw

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

@@ -405,11 +405,22 @@ Draft，不是仅等待评审。尚无模型或成本收益证据。历史 effec
 受支持恢复窗口核清前保留历史 decoder。回退同样区分未结工作与新准入。版本迁移不能
 把保存的 prompt、journal 或源 hash 改写成仿佛一直使用新契约。
 
-C1 首个 PR 必须覆盖上述两个生产组合调用点，并交付至少一条从生产端到最终输入的
+C1 首次跨入口切换必须覆盖上述两个生产组合调用点，并交付至少一条从生产端到最终输入的
 完整旅程，同时明确剩余 reader。这个有界切片通过资格，不代表可向未验收 host 开启
 新 writer，也不关闭 C1。最终切换须在全部受支持 reader 验收后删除活跃格式选择。
 临时 decoder 有具名 caller 与退出条件，不是面向用户的兼容视图。
 
+Turn 生命周期可以在 C1 父提案下单独评审，但须一起覆盖规划、最终 Host 输入、capture
+发布、journal 检查、原会话恢复及结算消费者。已迁移 Turn 使用一个完整来源及其已校验
+reader；这个分阶段边界保留普通 quota writer 和 thin-heartbeat 的现有契约。共享投影
+改动须保留这些未迁移 caller。历史 v0 decoder 服务已记录的 plan、request 和 journal，
+不能为非法的已提供 v1 来源兜底；其删除仍依赖未结工作及受支持恢复窗口核清。
+
+这属于 C1 的部分交付，不代表跨入口切换取得资格，也不证明上下文或成本收益。父提案
+仍负责普通 quota 与 heartbeat 迁移、最终输入／材料读取覆盖、全宿主配置，以及原单会话
+成果与完整成本对照。评审须包含匹配基线／候选版本的真实 provider 成果、拒绝敏感性和
+打包恢复。只有切片合入 main、父提案集成 main 后，父提案的差异才会减少；开 PR 不算。
+
 对 normal、selection、replan、wait、recovery 和 capability-off 的每个 fixture，
 在**三个边界**核对独立编写的义务：组合契约、重建交付、最终模型输入及必需材料读取。
 包含旧 action signature 未覆盖的源变化、必需来源缺失、不兼容消费者和结算响应丢失。
@@ -435,6 +446,26 @@ C1 首个 PR 必须覆盖上述两个生产组合调用点，并交付至少一
 decoder 前证明 export、旧版本 readback 与降级行为；满足受支持保留/升级边界前保留旧
 reader。本文不单独授权破坏性迁移。
 
+#### Turn 消费方准备与指令保真
+
+独立 Turn 提案让普通 quota producer 保持已记录契约；更广的 quota／heartbeat 切换
+仍归父提案。未迁移 producer 的测试须守住这个边界。当前 Turn 规划直接消费完整来源，
+发布自己的不可变 capture；新组合来源与较早 executor envelope 错配时继续拒绝。
+
+原生所选 Task 在 kernel delivery guard 之前准备普通 managed lease，并在模型后验证时
+复用该 proof。claim 标签本身不足，不能借用另一次执行的 lease。来源检查先于独立
+Task argv；拒绝和原结果恢复保留原会话与一次性效果。
+
+现代 Codex 与结构化 Host 指令共用既有 audit 读取规则：在判断 Goal acceptance 或提出
+closeout 前读取已声明 audit，即使 Goal 保持开放。通知策略不授予或取消执行权；所供
+delivery、attempt、quiet-noop、execution 与 orchestration 约束仍须遵守。状态、quota
+和 scheduler 效果仍归 Core。完整契约可以声明待执行的原材料读取：发布不等于已读，
+测试须真正执行 reader，并保留原正文与停止条件。
+
+一次性 research-team launcher 与 initializer 共同定位实际加载的 Core，包括隔离 wheel，
+仍保留 runtime 独占创建。源码和打包消费者验证，分别区别于已安装 App 采用、真实可选
+provider／模型行为及相对原 quota 的完整成本收益。这些修复不关闭 C1，也不授权部署或合并。
+
 ### C1 基线边界与复用盘点
 
 基线是**原路径中调用者实际收到的 quota 交付**，不是内部原始 decision。

examples/managed-research-team/bootstrap.ts

@@ -4,21 +4,31 @@
  */
 import {readFile, mkdir} from "node:fs/promises";
 import {resolve, join, isAbsolute} from "node:path";
+import {pathToFileURL} from "node:url";
 import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
-import {canonicalAuthoritySha256, authorityUnicodeCompare} from "../../loopx/control_plane/coordination/authority_store_codec.ts";
-import {canonicalTodoDomainRecord, TODO_DOMAIN_ITEM_SCHEMA, TODO_DOMAIN_READ_RECORD_SCHEMA}
-  from "../../loopx/control_plane/coordination/coordination_state_contract.ts";
-import {coordinationTodoReadModel} from "../../loopx/control_plane/coordination/coordination_projection.ts";
-import {openLocalAuthorityStore, selectLocalSqliteAuthority}
-  from "../../loopx/control_plane/coordination/local_authority_provider.ts";
-import {engageLegacyCoordinationWriterFence}
-  from "../../loopx/control_plane/coordination/legacy_writer_fence.ts";
-import {configureGoalAcceptance} from "../../loopx/control_plane/goals/acceptance_authority.ts";
 
-const [directory, provider = "file"] = process.argv.slice(2);
-if (!directory || !isAbsolute(directory) || !["file", "sqlite"].includes(provider)) {
-  throw new Error("absolute disposable directory and file|sqlite provider required");
+const [directory, provider = "file", packageRoot] = process.argv.slice(2);
+if (!directory || !isAbsolute(directory) || !packageRoot || !isAbsolute(packageRoot)
+    || !["file", "sqlite"].includes(provider)) {
+  throw new Error("absolute disposable directory, package root and file|sqlite provider required");
 }
+// The Python launcher pins its actually loaded package, including a wheel.
+// Keep this disposable initializer on that same production authority owner.
+const moduleUrl = (path: string) => pathToFileURL(resolve(packageRoot, "control_plane", path)).href;
+const [codec, contract, projectionOwner, providerOwner, fenceOwner, acceptanceOwner] = await Promise.all([
+  import(moduleUrl("coordination/authority_store_codec.ts")),
+  import(moduleUrl("coordination/coordination_state_contract.ts")),
+  import(moduleUrl("coordination/coordination_projection.ts")),
+  import(moduleUrl("coordination/local_authority_provider.ts")),
+  import(moduleUrl("coordination/legacy_writer_fence.ts")),
+  import(moduleUrl("goals/acceptance_authority.ts")),
+]);
+const {canonicalAuthoritySha256, authorityUnicodeCompare} = codec;
+const {canonicalTodoDomainRecord, TODO_DOMAIN_ITEM_SCHEMA, TODO_DOMAIN_READ_RECORD_SCHEMA} = contract;
+const {coordinationTodoReadModel} = projectionOwner;
+const {openLocalAuthorityStore, selectLocalSqliteAuthority} = providerOwner;
+const {engageLegacyCoordinationWriterFence} = fenceOwner;
+const {configureGoalAcceptance} = acceptanceOwner;
 const root = resolve(directory);
 const setup = JSON.parse(await readFile(join(root, "bootstrap.json"), "utf8"));
 const goalId = "synthetic-managed-research";

examples/managed-research-team/research_team.py

@@ -13,13 +13,15 @@
 import sys
 import uuid
 
+import loopx
 from loopx.control_plane.operator_provider import operator_provider_environ
 
 from scenario import REVISIONS, roster, encoded, evidence, task
 from acceptance import GOAL, canonical_tasks, require_completed, todo_id, validate_delivery, validate_member
 from execution import host_arguments, configure_delegations
 
 HERE = Path(__file__).resolve().parent
+PACKAGE_ROOT = Path(loopx.__file__).resolve().parent
 
 
 def write(path: Path, value: dict) -> None:
@@ -122,7 +124,7 @@ def git(*args: str) -> None:
     }})
     initialized = subprocess.run(
         ["node", "--no-warnings", "--experimental-sqlite", "--experimental-strip-types",
-         str(HERE / "bootstrap.ts"), str(root), provider],
+         str(HERE / "bootstrap.ts"), str(root), provider, str(PACKAGE_ROOT)],
         capture_output=True, text=True, timeout=45,
     )
     if initialized.returncode:

loopx/ark_managed_agent_host.py

@@ -49,7 +49,7 @@ def build_ark_managed_agent_host_contract() -> dict[str, Any]:
                 "quota_should_run.interaction_contract.cli_channel."
                 "runtime_capability_reentry"
             ),
-            "cli_projection_ref": "quota_should_run.runtime_capability_reentry",
+            "cli_projection_ref": "quota_should_run.interaction_contract.cli_channel.runtime_capability_reentry",
             "packet_schema_version": RUNTIME_CAPABILITY_REENTRY_SCHEMA_VERSION,
             "delivery_channel": "quota_tool_result",
             "goal_prompt_mutated": False,

loopx/capabilities/native_chat/steward_managed_mode.py

@@ -442,11 +442,19 @@ def _validation_source(self, request: dict[str, Any], todo_id: str) -> dict[str,
             raise ManagedTaskValidationUnavailable("the selected task needs its existing independent argv validation declaration")
         return rows[0]
 
-    def prepare_validation(self, request: dict[str, Any], native_request: dict[str, Any]) -> None:
-        from ...control_plane.turn_driver.driver import selected_turn_todo
+    def prepare_validation(self, request: dict[str, Any], native_request: dict[str, Any], *,
+                           _checked_work_identity: dict[str, str] | None = None) -> None:
+        from ...control_plane.turn_driver.driver import selected_turn_todo, turn_host_lineage
         from ...control_plane.turn_driver.command_validation import build_loopx_turn_command_validator
         from ...control_plane.todos.completion_validation_store import read_completion_validation_declaration
-        selected = selected_turn_todo(native_request.get("turn_envelope") or {})
+        # Dispatch can reuse its just-checked identity within this invocation;
+        # direct callers still check the complete source through the same owner.
+        if "work_contract" in native_request:
+            selected = _checked_work_identity
+            if selected is None:
+                selected = turn_host_lineage(native_request)
+        else:
+            selected = selected_turn_todo(native_request.get("turn_envelope") or {})
         if not selected.get("todo_id"):
             raise ManagedTaskValidationUnavailable("select an existing native task with independent validation")
         self.validation_basis = self._validation_source(request, selected["todo_id"])
@@ -467,11 +475,16 @@ def validator(self, request: dict[str, Any], *, turn_id: str):
         The captured canonical digest and fresh source must agree before and
         after the existing command validator reads the task's real artifact.
         """
-        from ...control_plane.turn_driver.driver import selected_turn_todo
+        from ...control_plane.turn_driver.driver import (
+            COMPLETE_TURN_PLAN_SCHEMA_VERSION, selected_turn_todo, turn_host_lineage,
+        )
 
         def validate(plan, result):
             actor = request["actor"]
-            selected = selected_turn_todo(plan.get("turn_envelope") or {})
+            # Model work is a reentry boundary. Recheck the complete source;
+            # only an absent contract retains historical envelope decoding.
+            selected = (turn_host_lineage(plan) if "work_contract" in plan or plan.get("schema_version") == COMPLETE_TURN_PLAN_SCHEMA_VERSION
+                        else selected_turn_todo(plan.get("turn_envelope") or {}))
             basis = self.validation_basis
             if not basis or not self.task_validator or basis["todo_id"] != selected["todo_id"]:
                 raise ValueError("the original task validation source was not captured before model work")