refactor(turn): share route qualification across real callers

docs/architecture/rfcs/STATUS.md

@@ -68,7 +68,7 @@ appendix may keep dated history, but no dated log heading may precede it.
 | [RFC: Shared Goal Alignment and Governed Amendment Protocol (v0)](shared-goal-alignment-and-governed-amendment-v0.md) | Accepted | none | [2 entries](ledger/shared-goal-alignment-and-governed-amendment-v0/) |
 | [RFC: LoopX Shared Control-Plane Authority and Pluggable State Providers (v0)](shared-goal-authority-state-provider-v0.md) | Accepted | none | [24 entries](ledger/shared-goal-authority-state-provider-v0/) |
 | [RFC: Single-Owner Local Daemon (v0)](single-owner-local-daemon-v0.md) | Accepted | none | — |
-| [RFC: TypeScript Control-Plane Migration Direction v0](typescript-control-plane-migration-v0.md) | Accepted | none | [16 entries](ledger/typescript-control-plane-migration-v0/) |
+| [RFC: TypeScript Control-Plane Migration Direction v0](typescript-control-plane-migration-v0.md) | Accepted | none | [18 entries](ledger/typescript-control-plane-migration-v0/) |
 
 ## Superseded (1)
 

docs/architecture/rfcs/STATUS.zh-CN.md

@@ -65,7 +65,7 @@
 | [RFC：共享 Goal 对齐与受治理 Amendment 协议（v0）](shared-goal-alignment-and-governed-amendment-v0.zh-CN.md) | 已接受 | 无 | [2 条](ledger/shared-goal-alignment-and-governed-amendment-v0/) |
 | [RFC：LoopX 共享控制面权威与可插拔状态 Provider（v0）](shared-goal-authority-state-provider-v0.zh-CN.md) | 已接受 | 无 | [24 条](ledger/shared-goal-authority-state-provider-v0/) |
 | [RFC: Single-Owner Local Daemon (v0)](single-owner-local-daemon-v0.md) | 已接受 | none | — |
-| [RFC：LoopX 控制面 TypeScript 渐进迁移方向 v0](typescript-control-plane-migration-v0.zh-CN.md) | 已接受 | 无 | [16 条](ledger/typescript-control-plane-migration-v0/) |
+| [RFC：LoopX 控制面 TypeScript 渐进迁移方向 v0](typescript-control-plane-migration-v0.zh-CN.md) | 已接受 | 无 | [18 条](ledger/typescript-control-plane-migration-v0/) |
 
 ## 已被替代 (1)
 

docs/reference/protocols/turn-controller-contract-v0.md

@@ -35,8 +35,10 @@ routes remain failures in the semantic verifier.
 
 ## The decision is not a result-kind string map
 
-There is no unary result-kind-to-route projection. The driver qualifies the
-fresh quota envelope into a route. The controller then evaluates:
+There is no unary result-kind-to-route projection. The existing TypeScript
+`quota/turn_envelope.ts::turnEnvelopeRoute` qualifies the fresh quota envelope
+into a route; the Python driver and loop controller consume that same decision.
+The controller then evaluates:
 
 `decide_loop_disposition(receipt, quota_decision, predecessor_turn_key, budget)`.
 
@@ -46,12 +48,18 @@ adapter. A string-to-string result map would discard these inputs and precedence
 
 The generated contract contains 29 ordered rows: 10 admission checks and 19
 returns. Conditions use finite partitions; the first matching return wins.
-The existing Python controller remains the sole evaluator. TypeScript exports
-shared data/types and the projection data, not a second decision implementation.
-
-Receipt settlement/completion qualification, budget construction, envelope
-signature validation, lineage and retry normalization remain their existing
-Python-owned primitives. The JSON specifies when those checks run; it is not a
+The existing Python controller remains the sole evaluator of these ordered
+disposition rows. TypeScript owns envelope route qualification and the shared
+data/types; it does not implement another receipt/budget disposition engine.
+
+Receipt settlement/completion qualification, budget construction and retry
+normalization retain their existing owners. Python preserves the historical
+scalar/numeric codecs used by persisted Turn identities; TypeScript owns route
+and envelope-signature interpretation. Plan projection returns route and
+signature together, and managed-step consumes an immutable local envelope/route
+result from one producer call. An extra field in a wire dictionary cannot carry
+that qualification; ordinary or historical dictionaries still use the shared
+route reader. The JSON specifies when those checks run; it is not a
 self-contained validator for arbitrary raw objects. Rejection is part of the
 function contract, not another enum value. No host launch, quota spend or state
 write is authorized by a controller output.

loopx/cli_commands/turn_decision.py

@@ -32,7 +32,9 @@
 )
 from ..control_plane.quota.live_decision import build_live_quota_should_run_decision
 from ..control_plane.quota.effective_action import EffectiveAction
-from ..control_plane.quota.turn_envelope import build_turn_envelope
+from ..control_plane.quota.turn_envelope import (
+    TurnControllerInput, build_turn_controller_input, build_turn_envelope,
+)
 from ..control_plane.quota.settlement import SettlementIdentity, build_turn_scoped_cli_settlement_plan
 from ..control_plane.quota.heartbeat_receipt import (
     find_heartbeat_receipt,
@@ -447,32 +449,14 @@ def current_heartbeat_host_launch_decision(
         else current.get("effective_action"))}
 
 
-def fresh_turn_envelope(
-    decision: Mapping[str, Any],
-    *,
-    scheduler_execution_context: Any,
-) -> dict[str, Any]:
-    """Sign the decision into the envelope the loop controller consumes.
-
-    The causal link to a failed predecessor is *not* written here: the envelope
-    schema is a signed contract, so the managed step passes its predecessor
-    beside the envelope instead of mutating it.
-    """
-
-    return build_turn_envelope(
-        decision,
-        scheduler_execution_context=scheduler_execution_context,
-    )
-
-
 def build_fresh_envelope_for_managed_step(
     args: argparse.Namespace,
     *,
     registry_path: Path,
     runtime_root: Path,
     runtime_root_arg: str | None,
-) -> dict[str, Any]:
-    """Project the current decision as the managed step's fresh envelope.
+) -> TurnControllerInput:
+    """Project the current envelope and its route together for managed-step.
 
     Read-only: no Turn-start hooks are dispatched and no Turn instance is
     minted, because the managed step only decides whether the outer scheduler
@@ -485,7 +469,7 @@ def build_fresh_envelope_for_managed_step(
         runtime_root=runtime_root,
         runtime_root_arg=runtime_root_arg,
     )
-    return fresh_turn_envelope(
+    return build_turn_controller_input(
         owner.resolve(),
         scheduler_execution_context=owner.scheduler_execution_context,
     )
@@ -498,5 +482,4 @@ def build_fresh_envelope_for_managed_step(
     "build_fresh_envelope_for_managed_step",
     "build_fresh_turn_decision_owner",
     "collect_turn_status_payload",
-    "fresh_turn_envelope",
 ]

loopx/control_plane/quota/turn_envelope.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import json
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
@@ -26,6 +28,26 @@
 PLANNING_HORIZON_DETAIL_REFS_REF = "$.detail_ref"
 
 
+@dataclass(frozen=True, init=False, repr=False, slots=True)
+class TurnControllerInput:
+    """Local producer result; never an agent wire or a retained authority cache.
+
+    The envelope and its typed route originate in one projection. Store the
+    envelope as immutable bytes so a caller cannot mutate it after routing;
+    each consumer receives its own decoded object. Ordinary wire dictionaries
+    cannot supply this qualification through an extra JSON field.
+    """
+
+    envelope_json: str
+    route: str
+
+    def __init__(self) -> None:
+        raise TypeError("Use build_turn_controller_input with the original quota source")
+
+    def envelope(self) -> dict[str, Any]:
+        return json.loads(self.envelope_json)
+
+
 def _turn_envelope_result(
     operation: str,
     **params: object,
@@ -91,14 +113,39 @@ def build_turn_envelope(
 ) -> dict[str, Any]:
     """Project one full quota decision through the canonical TS transaction."""
 
+    return _build_turn_projection(payload, scheduler_execution_context=scheduler_execution_context,
+        captured_decision_path=captured_decision_path)
+
+
+def build_turn_controller_input(
+    payload: Mapping[str, Any], *, scheduler_execution_context: object = None,
+) -> TurnControllerInput:
+    """Build the managed controller's envelope and route in one existing RPC."""
+    result = _build_turn_projection(payload,
+        scheduler_execution_context=scheduler_execution_context,
+        operation="build_controller_input")
+    envelope, route = result.get("envelope"), result.get("route")
+    if not isinstance(envelope, Mapping) or not isinstance(route, str):
+        raise RuntimeError("TypeScript controller input must include an envelope and route")
+    projected = object.__new__(TurnControllerInput)
+    object.__setattr__(projected, "envelope_json", json.dumps(dict(envelope), ensure_ascii=False,
+        separators=(",", ":")))
+    object.__setattr__(projected, "route", route)
+    return projected
+
+
+def _build_turn_projection(
+    payload: Mapping[str, Any], *, scheduler_execution_context: object = None,
+    captured_decision_path: str | None = None, operation: str = "build",
+) -> dict[str, Any]:
     resolution = resolve_scheduler_execution_context(scheduler_execution_context)
     scheduler_execution_args = ""
     if resolution.ok and resolution.context is not None:
         scheduler_execution_args = render_scheduler_execution_args(
             scheduler_execution_context=resolution,
         )
     return _turn_envelope_result(
-        "build",
+        operation,
         payload=dict(payload),
         scheduler_execution_args=scheduler_execution_args,
         **({"captured_decision_path": captured_decision_path}

loopx/control_plane/quota/turn_envelope.ts

@@ -1,4 +1,5 @@
-import { EffectiveAction } from "./effective_action.generated.ts";
+import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts";
+import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts";
 import { createHash } from "node:crypto";
 import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
 import {omitUnboundExploreResultWriteback, referenceExploreResultWriteback} from "../capabilities/explore_turn_context.ts";
@@ -1232,6 +1233,88 @@ function bindTurnDecisionCapture(request: JsonObject): JsonObject {
   return bound;
 }
 
+// Explicit coverage of the registered decision/frontier union. New actions
+// require a deliberate classification; unknown historical values retain Host routing.
+export const TURN_ACTION_ROUTES = {
+  [EffectiveAction.AGENT_WORKSPACE_REPAIR]: "repair_required",
+  [EffectiveAction.BOUNDARY_PROJECTION_REPAIR]: "repair_required",
+  [EffectiveAction.CAPABILITY_BRIDGE_REPAIR]: "repair_required",
+  [EffectiveAction.CONTROL_PLANE_HEALTH_REPAIR]: "repair_required",
+  [EffectiveAction.CONTROL_PLANE_PROJECTION_REPAIR]: "repair_required",
+  [EffectiveAction.CONTROL_PLANE_REPAIR]: "repair_required",
+  [EffectiveAction.RUNTIME_USER_GATE_PROJECTION_REPAIR]: "repair_required",
+  [EffectiveAction.STATE_PROJECTION_GAP_REPAIR]: "repair_required",
+  [EffectiveAction.TODO_DECISION_SCOPE_PROJECTION_REPAIR]: "repair_required",
+  [EffectiveAction.AGENT_MONITOR_ONLY]: "ready_for_host",
+  [EffectiveAction.AUTOMATION_PROMPT_UPGRADE_REQUIRED]: "ready_for_host",
+  [EffectiveAction.BLOCKED_HEALTH]: "ready_for_host",
+  [EffectiveAction.BLOCKED_WAIT]: "ready_for_host",
+  [EffectiveAction.COORDINATE_TASK_BUNDLE]: "ready_for_host",
+  [EffectiveAction.EXTERNAL_EVIDENCE_OBSERVE]: "ready_for_host",
+  [EffectiveAction.HEARTBEAT_RECEIPT_WRITE_FAILED]: "ready_for_host",
+  [EffectiveAction.HEARTBEAT_SETTLED_SKIP]: "ready_for_host",
+  [EffectiveAction.LARK_INBOX_REPLY_DUE]: "ready_for_host",
+  [EffectiveAction.MONITOR_DUE]: "ready_for_host",
+  [EffectiveAction.MONITOR_QUIET_SKIP]: "ready_for_host",
+  [EffectiveAction.NORMAL_RUN]: "ready_for_host",
+  [EffectiveAction.OPERATOR_GATE_NOTIFY]: "ready_for_host",
+  [EffectiveAction.OPERATOR_INBOX_MATERIAL_REVIEW_DUE]: "ready_for_host",
+  [EffectiveAction.OUTCOME_FLOOR_RECOVERY]: "ready_for_host",
+  [EffectiveAction.PEER_COORDINATION_BLOCKED]: "ready_for_host",
+  [EffectiveAction.QUOTA_SKIP]: "ready_for_host",
+  [EffectiveAction.SCOPED_USER_GATE_FALLBACK]: "ready_for_host",
+  [EffectiveAction.TERMINAL_NO_FOLLOWUP]: "ready_for_host",
+  [EffectiveAction.THROTTLED_SKIP]: "ready_for_host",
+  [EffectiveAction.UNSETTLED_HOST_TURN_RECOVERY]: "ready_for_host",
+  [EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED]: "replan_required",
+  [EffectiveAction.GOVERNED_CAPABILITY_INTENT]: "capability_action_required",
+  agent_scope_exhausted: "ready_for_host",
+  agent_scope_wait: "ready_for_host",
+  reassignment_required: "ready_for_host",
+  autonomous_replan: "replan_required",
+  successor_replan_required: "replan_required",
+} satisfies Record<QuotaEffectiveActionValue | "autonomous_replan" | "successor_replan_required", LoopXTurnRouteValue>;
+
+/** Route one admitted input, independent of controller continuation/budget policy. */
+export function turnEnvelopeRoute(envelope: JsonObject, historicalText?: JsonObject): LoopXTurnRouteValue {
+  const signature = object(envelope.action_signature);
+  // The historical Python reader treats false/zero/null as an empty hash and
+  // renders scalar booleans with Python spelling. Valid hashes remain strings.
+  const sourceHash = scalarString(historicalText ? historicalText.source_hash : signature.source_hash || "", "route source hash");
+  const envelopeHash = scalarString(historicalText ? historicalText.envelope_hash : signature.envelope_hash || "", "route envelope hash");
+  if (signature.matches !== true || !sourceHash || sourceHash !== envelopeHash) return "contract_error";
+  // Size diagnostics grant no authority and do not veto a valid contract.
+  const action = object(envelope.action);
+  if (envelope.should_run === true) {
+    if (action.must_attempt !== true) return "blocked";
+    const name = scalarString(historicalText ? historicalText.effective_action : envelope.effective_action || "", "route effective action");
+    const route = Object.hasOwn(TURN_ACTION_ROUTES, name)
+      ? TURN_ACTION_ROUTES[name as keyof typeof TURN_ACTION_ROUTES] : "ready_for_host";
+    if (route === "replan_required") {
+      try { turnEnvelopeReplanSettlementIdentity(envelope); }
+      catch (error) {
+        if (!(error instanceof EffectRuntimeRequestError)) throw error;
+        return "contract_error";
+      }
+      return "replan_required";
+    }
+    if (action.delivery_allowed !== true) return "blocked";
+    if (route === "capability_action_required") {
+      const intent = object(action.capability_intent), command = intent.command;
+      // Preserve empty-container falsiness from historical Python inputs.
+      const hasCommand = Array.isArray(command) ? command.length > 0
+        : command !== null && typeof command === "object" ? Object.keys(command).length > 0 : Boolean(command);
+      if (intent.schema_version !== "pending_capability_intent_projection_v0"
+          || (intent.goal_id ?? null) !== (envelope.goal_id ?? null)
+          || (intent.agent_id ?? null) !== (envelope.agent_id ?? null) || !hasCommand) return "contract_error";
+      return "capability_action_required";
+    }
+    return route === "repair_required" ? "repair_required" : "ready_for_host";
+  }
+  if (object(envelope.user).action_required === true) return "user_action_required";
+  return action.quiet_noop_allowed === true ? "wait" : "blocked";
+}
+
 export function evaluateTurnEnvelope(value: unknown): JsonObject {
   const request = requireJsonObject(value, "turn envelope request");
   const operation = request.operation;
@@ -1268,6 +1351,18 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
       autonomous_replan_obligation: {...obligation, context_read: reads.review},
       replan_action_packet: replanActionPacket(payload)};
   }
+  if (operation === "driver_route") return {route: turnEnvelopeRoute(requireJsonObject(request.envelope, "Turn route envelope"),
+    request.historical_route_text === undefined ? undefined : requireJsonObject(request.historical_route_text, "historical route text"))};
+  if (operation === "driver_lineage") {
+    const envelope = requireJsonObject(request.envelope, "Turn driver envelope");
+    return {route: turnEnvelopeRoute(envelope, request.historical_route_text === undefined
+      ? undefined : requireJsonObject(request.historical_route_text, "historical route text")),
+      action_signature: turnEnvelopeActionSignatureDocument(envelope)};
+  }
+  if (operation === "build_controller_input") {
+    const envelope = buildTurnEnvelope(request);
+    return {envelope, route: turnEnvelopeRoute(envelope)};
+  }
   if (operation === "consumer_version") {
     assertTurnEnvelopeConsumerVersion(requireJsonObject(request.envelope, "turn_envelope"));
     return {ok: true};