refactor(turn): qualify receipts and continuation in the typed owner

docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md

@@ -132,6 +132,47 @@ different questions; their proximity is not proof they should be one state.
   benefit experiments. Replace real consumers in the delivery slice; preserve
   historical decoding and original effect readback across code rollback.
 
+**Source proposal (2026-10-28).** The existing Effect runtime now composes
+journal receipt qualification and ordered continuation in the typed Turn owner.
+`managed_step.py` sends one complete journal/decision request instead of rebuilding
+an execution payload; the driver uses the same route classifier. The registered
+ordered contract, historical journal decoder and independent recovery decision
+remain. Public Python receipt/budget value objects keep their API, including exact
+integer transport. Native File/SQLite retry, saved-result replay and lost-spend
+response tests retain the original Turn and one debit. Rule mutations exercise
+the actual TS evaluator. This is a single-session source proposal, not C1 cutover,
+C4.2/C4.3 adoption, installed qualification or long-horizon benefit. Reconcile
+original receipts before rollback; no journal or settlement schema changes.
+The same inventory measures 43 maintained Python/TS pairs on the base and 45 on
+this proposal: the new controller and host-failure pairs retain the Python API
+and native failure producer while sharing TS decisions. The pair budget records
+that maintenance cost (43 → 45); generated exclusions and the over-budget
+negative check remain. Deleted Python decisions, rather than the pair count,
+establish convergence. Added bridge traffic is disclosed without a speedup claim.
+
+Review repair rebases this proposal onto the current shared
+`turnEnvelopeRoute` owner and retains `TurnControllerInput` at the Python API.
+It does not restore the former route table or Python policy partitions.
+Receipt qualification rejects empty legacy lineage values before creating a
+validated receipt. Historical Mapping callers retain Python scalar spelling
+through the existing route text codec, including `1.0` and `1e-07`; mismatched
+signatures still fail. The same public API regressions reproduce six empty
+container admissions and two numeric-route refusals at the original proposal
+head, and pass after repair. This corrects compatibility; it does not qualify
+numeric hashes as canonical producer output or close the remaining C4 work.
+
+A subsequent review reproduced a transport regression with a complete 2,420,000-
+character Goal: both Mapping and producer-qualified controller inputs failed
+before dispatch, although the existing envelope/plan accepted them. The pure
+`turn.controller` operation now reuses the envelope's bounded local snapshot
+fallback after a pre-send size rejection, with explicit client/server allowlist
+entries. Full facts and the original 2 MiB wire / 64 MiB snapshot bounds remain;
+signature rejection, private-file/hash/length checks and ambiguous-response
+non-retry behavior are retained. Real transport tests cover both input forms,
+small controls, escaped Unicode and the actual snapshot limit. This repairs the
+source proposal's large-input path; independent review, installed adoption and
+long-horizon cost/benefit remain separate qualification.
+
 ## C4.2 — Thin outer-worker semantic adoption
 
 **Candidate for bounded policy convergence.**

docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md

@@ -104,6 +104,33 @@ execution 形状的回执，同时调用 `turn_journal.ts` 检查一致性。`re
   等待完整 C1 收益实验。在交付切片中替换真实消费者；代码回退时保留历史解码与原
   effect readback。
 
+**源码提案（2026-10-28）。** 既有 Effect runtime 在 typed Turn owner 内组合 journal
+回执资格与有序继续判定。`managed_step.py` 一次传入完整 journal/decision，不再重建
+execution 载荷；driver 复用同一路由分类器。原注册有序契约、历史 journal decoder 和
+独立恢复判定保持。公开 Python 回执/预算值对象保留 API，整数通过精确编码传递。
+真实 File/SQLite 重试、已存结果重放和扣额丢响应测试保留原 Turn 与单次扣额；规则
+mutation 直接验证实际 TS evaluator。本项仅为单会话源码提案，不代表 C1 切换、
+C4.2/C4.3 采用、已安装资格或长程收益。回滚前核对原回执；journal 和结算 schema 不变。
+同一 inventory 在 base 测得 43 对独立维护的 Python/TS 模块，本提案为 45 对：新增
+controller 与 host-failure 配对保留 Python API 和原生失败 producer，同时共享 TS
+决策。配对预算据此记录维护成本（43 → 45）；生成排除规则和超预算负例保持。
+收敛依据是删除了 Python 决策，不是配对计数；额外 bridge 传输如实披露，不宣称提速。
+
+评审修复将本提案重基到当前共享的 `turnEnvelopeRoute` owner，并保留 Python API 的
+`TurnControllerInput`，不恢复旧路由表或 Python 策略分区。回执资格在创建 validated
+receipt 前拒绝历史空身份值；历史 Mapping 调用方通过既有 route text codec 保留
+Python 标量文本，包括 `1.0` 和 `1e-07`，签名不一致仍拒绝。同一组公开 API 回归在
+原提案 head 复现六个空容器误准入和两个数字路由误拒绝，修复后通过。这是兼容修复，
+不表示 canonical producer 会生成数字 hash，也不关闭其余 C4 工作。
+
+后续评审用包含 2,420,000 字符完整 Goal 的输入复现传输回归：既有 envelope/plan
+可以接受，Mapping 与 producer-qualified 两种 controller 输入却都在派发前失败。
+纯 `turn.controller` 操作现于预发送大小拒绝后复用 envelope 的有界本地 snapshot
+回退，并显式加入客户端/服务端白名单。完整事实及原 2 MiB wire / 64 MiB snapshot
+限制保持；签名拒绝、私有文件/hash/长度校验和响应不确定时不重试的行为保持。
+真实传输测试覆盖两种输入、小输入控制、转义 Unicode 和实际 snapshot 上限。
+这修复源码提案的大输入路径；独立复审、安装采用和长程完整成本/收益仍分别验收。
+
 ## C4.2 — 外层 worker 的薄语义采用
 
 **有界策略收敛候选。** `scripts/external_scheduler_worker.py` 的 `_host_attempt`、

examples/semantic-vocabulary-drift-smoke.py

@@ -202,6 +202,7 @@
 LITERAL_SCAN_ROOTS = ["loopx"]
 # Input producers with an executable witness in loopx.semantics.production.
 INPUT_PRODUCER_ANCHOR = {
+    "turn_route": "loopx/control_plane/turn_driver/driver.py::_typed_route",
     "effective_action": "loopx/control_plane/quota/run_decision.ts::resolveQuotaRunDecision",
     "turn_result_kind": "loopx/control_plane/turn_driver/transaction.py::_result_kind",
     "loop_disposition": "loopx/control_plane/turn_driver/loop_controller.py::decide_loop_disposition",
@@ -215,15 +216,16 @@
     "settlement_binding_kind",
 }
 RETURN_PRODUCER_ANCHOR = {
-    "turn_route": {"loopx/control_plane/turn_driver/driver.py::_typed_route", "loopx/control_plane/turn_driver/loop_controller.py::_envelope_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan"},
+    "turn_route": {"loopx/control_plane/turn_driver/driver.py::_typed_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan"},
     "loop_disposition": {"loopx/control_plane/turn_driver/turn_contract_generated.py::project_turn_route"},
     "effective_action": {"loopx/control_plane/quota/run_decision.ts::quotaEffectiveAction", "loopx/control_plane/quota/run_decision.ts::taskOrchestrationEffectiveAction"},
     "turn_result_kind": {"loopx/control_plane/turn_driver/executor.py::_task_validation_receipt"},
 }
 # Reviewed output arguments/paths replace the old any-enum-use heuristic. These
 # anchors retain that coverage if registry metadata is accidentally removed.
 CALL_PRODUCER_ANCHOR = {
-    "loop_disposition": {"loopx/control_plane/turn_driver/loop_controller.py::_disposition": ["disposition"]},
+    # The retired Python output builder is covered by the existing executable
+    # complete-controller witness; routes now also have an input witness.
     "agent_scope_frontier_action": {"loopx/control_plane/agents/agent_scope_frontier.py::build_agent_scope_frontier_payload": ["action"]},
     "turn_result_kind": {
         "loopx/control_plane/turn_driver/executor.py::_host_failure": ["kind"],
@@ -236,7 +238,7 @@
     "turn_result_kind": {"loopx/control_plane/turn_driver/executor.py::_task_validation_receipt": ["recovery_kind"]},
 }
 TWIN_ROOT_ANCHOR = "loopx/control_plane"
-TWIN_BUDGET_ANCHOR = 43
+TWIN_BUDGET_ANCHOR = 45
 BUDGET_ANCHOR = {
     "same_runtime_forks": 1,
     "same_runtime_fork_definitions": 3,

loopx/cli_commands/turn_managed_step.py

@@ -1,9 +1,9 @@
 """CLI surface for the managed-step consumer of same-Turn continuation.
 
 Split from ``turn.py`` to respect the CLI command-owner size budget. The
-subcommand resolves the canonical Turn journal, rebuilds its validated receipt,
-projects the current control-plane decision as a fresh envelope, asks the pure
-controller for a disposition, and prints a typed answer. It never launches a
+subcommand reads the canonical Turn journal and a fresh control-plane envelope,
+then sends both to the typed controller for qualification and continuation.
+It prints the existing managed-step answer. It never launches a
 host, writes state, or spends quota.
 """
 
@@ -12,6 +12,7 @@
 import argparse
 from collections.abc import Callable
 from pathlib import Path
+from typing import cast
 
 from ..control_plane.runtime.status_projection_cache import (
     resolve_status_projection_cache_runtime_root,
@@ -46,7 +47,7 @@ def _load_journal(
     journal = load_turn_journal(path)
     if journal is None:
         raise ValueError("LoopX Turn journal does not exist")
-    return journal
+    return cast(dict[str, object], journal)
 
 
 def handle_turn_managed_step(

loopx/control_plane/effect_runtime.py

@@ -57,6 +57,7 @@
     "quota.settlement.read",
     "quota.spend.commit",
     "quota.turn_envelope.evaluate",
+    "turn.controller",
     "goal.checkpoint_read_context.resolve",
     "goal.checkpoint_read_context.commit",
     "goal.checkpoint_read_context.inspect_replay",

loopx/control_plane/effect_runtime_handlers.ts

@@ -264,6 +264,7 @@ export function createEffectRuntimeHandlers(
       "turn_journal.inspect",
       lazyHandler(() => import("./turn_driver/turn_journal.ts"), ({interpretTurnJournal}) => (params) => interpretTurnJournal(turnJournalInspectionRequest(params))),
     ],
+    ["turn.controller", lazyHandler(() => import("./turn_driver/loop_controller.ts"), ({controllerRequest}) => controllerRequest)],
     ["turn_journal.write", lazyHandler(() => import("./turn_driver/turn_journal_effects.ts"), ({commitTurnJournal}) => commitTurnJournal)],
     ["turn_journal.find_settlement", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({findTurnJournalBySettlement}) => findTurnJournalBySettlement)],
     ["turn_journal.observed_capabilities", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({readTurnJournalCapabilities}) => readTurnJournalCapabilities)],

loopx/control_plane/effect_runtime_server.ts

@@ -43,6 +43,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([
   "quota.settlement.read",
   "quota.spend.commit",
   "quota.turn_envelope.evaluate",
+  "turn.controller",
   "goal.checkpoint_read_context.resolve",
   "goal.checkpoint_read_context.commit",
   "goal.checkpoint_read_context.inspect_replay",

loopx/control_plane/turn_driver/host_failure.ts

@@ -0,0 +1,47 @@
+/** Qualification shared by journal recovery and continuation; no host IO. */
+import { isDeepStrictEqual } from "node:util";
+import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { jsonObject, type JsonObject } from "../runtime_decode.ts";
+
+const hostFailureKinds: ReadonlySet<string> = new Set([
+  "auth_failed", "contract_rejected", "executor_timeout", "output_budget_exhausted",
+  "provider_capacity", "provider_overloaded", "quota_exhausted", "rate_limited",
+  "session_missing", "transport_lost", "unknown",
+]);
+const hostRetryPolicies: Readonly<Record<string, readonly [number, number]>> = {
+  executor_timeout: [2, 5], provider_capacity: [3, 30], provider_overloaded: [3, 30],
+  rate_limited: [3, 60], transport_lost: [3, 10],
+};
+
+export function normalizeHostFailure(value: unknown, integerAttempt?: string | null): JsonObject {
+  const fail = (message: string): never => { throw new EffectRuntimeRequestError(message); };
+  const record = jsonObject(value) ?? fail("host failure record must be one object");
+  if (record.schema_version !== "loopx_turn_host_failure_v0")
+    fail("host failure record has an unsupported schema");
+  const kind = typeof record.kind === "string" ? record.kind : "";
+  if (!hostFailureKinds.has(kind)) fail("unsupported built-in host failure kind");
+  const attempt = record.attempt;
+  const exact = integerAttempt === undefined ? null
+    : integerAttempt !== null && /^\d+$/.test(integerAttempt) ? BigInt(integerAttempt) : 0n;
+  if (exact !== null ? exact < 1n || Number(exact) !== attempt
+    : typeof attempt !== "number" || !Number.isInteger(attempt) || attempt < 1)
+    fail("host failure record has an invalid attempt");
+  if (typeof record.retryable !== "boolean") fail("host failure record has an invalid retryable marker");
+  const policy = hostRetryPolicies[kind];
+  if (record.retryable !== (policy !== undefined))
+    fail("host failure record retryability does not match its kind");
+  const expected: JsonObject = {schema_version: "loopx_turn_host_failure_v0", kind,
+    attempt: exact !== null && exact > BigInt(Number.MAX_SAFE_INTEGER) ? String(exact) : attempt,
+    retryable: policy !== undefined};
+  if (policy) {
+    expected.retry = {strategy: "same_configuration", max_attempts: policy[0],
+      backoff_seconds: Math.min(policy[1] * 2 ** Math.max(0, Number(attempt) - 1), 300)};
+    if (!isDeepStrictEqual(record.retry, expected.retry)) fail("host failure record has an invalid retry policy");
+  } else if (Object.hasOwn(record, "retry")) {
+    fail("non-retryable host failure must not declare retry policy");
+  }
+  if (Object.keys(record).length !== Object.keys(expected).length
+      || Object.keys(record).some(key => !Object.hasOwn(expected, key)))
+    fail("host failure record contains unsupported fields");
+  return expected;
+}