refactor(turn): share selected-work resolution with settlement

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

@@ -169,6 +169,16 @@ workload and limit, including number-dense overhead. Neither this prerequisite
 nor a component byte saving qualifies the unified writer/consumer cutover or
 long-horizon net benefit.
 
+The proposed Turn selection-reader slice removes the Python adaptive-selection
+decision from the driver. Its existing plan, Host, Session, completion and
+settlement callers use the interaction owner's `selectedWorkTodo` rule through
+the existing source-edit transport. The selection adapter retains Python's historical
+scalar/whitespace decoding; original records and numeric values stay intact.
+This retires one duplicate rule, not the historical envelope format. Complete
+writer/consumer cutover and original-quota outcome/full-cost qualification remain
+open. The added bridge calls must be counted; code convergence alone is no
+evidence of lower cost.
+
 | Current source / obligation | Present envelope treatment | Required C1 disposition |
 | --- | --- | --- |
 | `interaction_contract.agent_channel.work_context` and pending required reads | `work_context` plus `required_reads`; identical inline material can be referenced | One complete requirement/evidence source with resolvable references; preserve pending reads and their completion conditions |

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

@@ -132,6 +132,13 @@ Journal owner 在原子发布前恢复这些数值，并在重放与不可变 pl
 负载与上限下比较，包括数值密集时的元数据开销。该前置或组件字节减少，都不
 构成统一 writer/consumer 切换或长程净收益的资格。
 
+本次提议的 Turn 选择 reader 切片删除 driver 中 Python 的 adaptive 选择判断。
+现有 plan、Host、Session、completion 和 settlement 调用方经既有 source-edit
+传输复用 interaction owner 的 `selectedWorkTodo` 规则。此选择 adapter 仅保留
+Python 标量与空白解码，原记录和数字值不变。这退役的是一条重复规则，不是历史
+envelope 格式；完整 writer/consumer 切换和相对原 quota 的成果／完整成本资格仍
+开放。新增 bridge 调用必须计入，代码收敛本身不证明成本降低。
+
 | 当前来源 / 义务 | 当前 envelope 处理 | C1 必须保留的归属与语义 |
 | --- | --- | --- |
 | `interaction_contract.agent_channel.work_context` 与未完成必读 | `work_context` 加 `required_reads`；相同内联材料可改用引用 | 一份完整要求/证据来源与可解析引用；保留未完成必读及其完成条件 |

examples/personal-workspace-browser/replan-cadence.mjs

@@ -83,7 +83,7 @@ export const replanCadenceScenario = {
       const count = page.getByLabel("两次复核间的数量", { exact: true });
       const strategy = page.getByLabel(/^步幅策略/);
       await waitForInputValue(count, "6");
-      if (await strategy.inputValue() !== "off") throw new Error("Stride must default off");
+      if (await strategy.inputValue() !== "evidence_conditioned_v1") throw new Error("Stride must default to evidence_conditioned_v1");
       if (await unit.inputValue() !== "effective_turns") throw new Error("Product default is not settled work Turns");
       const horizonGoal = page.getByLabel(/^在规划视野中显示 Goal/);
       const advanced = page.getByText("高级设置", {exact: true});
@@ -302,7 +302,7 @@ export const replanCadenceScenario = {
       if (inherited.source !== "capability_default" || inherited.configuration.count_unit !== "effective_turns" || inherited.configuration.count !== 6) throw new Error("Removing device policy did not restore product default");
       await reopenGoalCadence();
       await waitForInputValue(englishCount, "6");
-      if (await englishUnit.inputValue() !== "effective_turns" || await englishStrategy.inputValue() !== "off") throw new Error("Goal editor did not restore product defaults");
+      if (await englishUnit.inputValue() !== "effective_turns" || await englishStrategy.inputValue() !== "evidence_conditioned_v1") throw new Error("Goal editor did not restore product defaults");
       await englishCount.fill("5");
       await page.getByRole("button", { name: "Preview changes", exact: true }).click();
       const failedReadback = holdGoalReadback();

loopx/control_plane/quota/turn_envelope.ts

@@ -20,7 +20,7 @@ import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { projectTurnStartUnavailableContext } from "../capability_hooks.ts";
 import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral } from "../runtime_decode.ts";
 import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts";
-import { projectInteractionRequiredReads, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
+import { projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
 import { measureTurnEnvelope, TURN_ENVELOPE_SECTION_TARGETS } from "./turn_envelope_budget.ts";
 export { TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts";
 
@@ -1456,6 +1456,11 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
       autonomous_replan_obligation: {...obligation, context_read: reads.review},
       replan_action_packet: replanActionPacket(payload)};
   }
+  if (operation === "selected_work_todo") {
+    const source = requireJsonObject(request.payload, "Turn work selection source");
+    const historical = scalarString(request.historical_primary_text, "historical primary Todo");
+    return contextProjectionChanges(source, {...source, selected_todo: selectedWorkTodo(source, historical)});
+  }
   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") {

loopx/control_plane/turn_driver/driver.py

@@ -84,21 +84,20 @@ def _typed_route(envelope: Mapping[str, Any]) -> LoopXTurnRoute:
 
 
 def selected_turn_todo(envelope: Mapping[str, Any]) -> dict[str, Any]:
-    """Resolve the todo that owns one Turn across adaptive bundle execution."""
+    """Transport historical selection to the shared work/settlement owner.
+
+    Only Python's original scalar codec remains here. Pure source edits retain
+    unknown fields and exact numbers without another JSON value decoder.
+    """
+    from ..quota.cli_projection import _apply_context_projection_changes
 
     orchestration = _mapping(envelope.get("task_orchestration_contract"))
-    primary_todo_id = str(orchestration.get("primary_todo_id") or "").strip()
-    if (
-        orchestration.get("schema_version") == "task_orchestration_contract_v2"
-        and orchestration.get("mode") == "adaptive"
-        and primary_todo_id
-    ):
-        return {
-            "todo_id": primary_todo_id,
-            "source": "task_orchestration_contract.primary_todo_id",
-        }
-    action = _mapping(envelope.get("action"))
-    return _mapping(action.get("selected_todo"))
+    source = {"selected_todo": _mapping(_mapping(envelope.get("action")).get("selected_todo")),
+              "task_orchestration_contract": orchestration}
+    edits = _turn_envelope_result("selected_work_todo", payload=source,
+        historical_primary_text=str(orchestration.get("primary_todo_id") or "").strip())
+    selected: dict[str, Any] = _apply_context_projection_changes(source, edits)["selected_todo"]
+    return selected
 
 
 def _receipt_bound_scoped_identity(

loopx/control_plane/work_items/interaction_contract.ts

@@ -735,14 +735,23 @@ function shellQuote(value: string): string {
   return /^[A-Za-z0-9_./:-]+$/.test(value) ? value : `'${value.replaceAll("'", `'"'"'`)}'`;
 }
 
-/** The adaptive primary is the work identity for both reads and settlement. */
-export function selectedWorkTodoId(packet: JsonObject): string | null {
+/** The adaptive primary is the work identity for both reads and settlement.
+ * Historical callers may supply their already-decoded text: reapplying JS
+ * trim would change Python whitespace/BOM identities in persisted journals.
+ */
+export function selectedWorkTodo(packet: JsonObject, historicalPrimaryText?: string): JsonObject {
   const orchestration = jsonObject(packet.task_orchestration_contract) ?? {};
-  const primaryTodoId = typeof orchestration.primary_todo_id === "string"
-    ? orchestration.primary_todo_id.trim() : "";
+  const primaryTodoId = historicalPrimaryText ?? (typeof orchestration.primary_todo_id === "string"
+    ? orchestration.primary_todo_id.trim() : "");
   if (orchestration.schema_version === "task_orchestration_contract_v2"
-      && orchestration.mode === "adaptive" && primaryTodoId) return primaryTodoId;
-  const selected = jsonObject(packet.selected_todo) ?? {};
+      && orchestration.mode === "adaptive" && primaryTodoId) {
+    return {todo_id: primaryTodoId, source: "task_orchestration_contract.primary_todo_id"};
+  }
+  return jsonObject(packet.selected_todo) ?? {};
+}
+
+export function selectedWorkTodoId(packet: JsonObject): string | null {
+  const selected = selectedWorkTodo(packet);
   return typeof selected.todo_id === "string" && selected.todo_id.trim() ? selected.todo_id : null;
 }
 

tests/control_plane/test_turn_work_selection.py

@@ -0,0 +1,62 @@
+"""Historical Turn selection retains original JSON and Python text identity."""
+from __future__ import annotations
+
+from copy import deepcopy
+import json
+from typing import Any
+
+import pytest
+
+from loopx.control_plane.turn_driver.driver import selected_turn_todo
+
+
+_SELECTED = {"todo_id": "fallback", "text_ref": "action.recommended_action",
+             "unknown": [None, False, 1.0, -0.0, 9007199254740993]}
+
+
+@pytest.mark.parametrize(("primary", "expected"), [
+    (" primary ", "primary"),
+    ("\u001cprimary\u001f", "primary"),
+    ("\ufeffprimary\ufeff", "\ufeffprimary\ufeff"),
+    ("", None), ("  ", None), (None, None), (False, None), (0, None),
+    (True, "True"), (7, "7"), (1.0, "1.0"),
+    ([], None), ({}, None), (["x"], "['x']"), ({"x": 1}, "{'x': 1}"),
+])
+def test_historical_primary_text_identity(primary: Any, expected: str | None) -> None:
+    envelope = {"action": {"selected_todo": deepcopy(_SELECTED)},
+                "task_orchestration_contract": {
+                    "schema_version": "task_orchestration_contract_v2",
+                    "mode": "adaptive", "primary_todo_id": primary}}
+    before = json.dumps(envelope, ensure_ascii=False, sort_keys=True)
+    wanted = (_SELECTED if expected is None else {
+        "todo_id": expected, "source": "task_orchestration_contract.primary_todo_id"})
+    assert json.dumps(selected_turn_todo(envelope), sort_keys=True) == json.dumps(wanted, sort_keys=True)
+    assert json.dumps(envelope, ensure_ascii=False, sort_keys=True) == before
+
+
+@pytest.mark.parametrize("orchestration", [
+    None, [], {}, {"schema_version": "task_orchestration_contract_v1", "mode": "adaptive", "primary_todo_id": "other"},
+    {"schema_version": "task_orchestration_contract_v2", "mode": "serial", "primary_todo_id": "other"},
+])
+def test_only_registered_adaptive_contract_overrides_selected_record(orchestration: Any) -> None:
+    envelope = {"action": {"selected_todo": deepcopy(_SELECTED)},
+                "task_orchestration_contract": orchestration}
+    assert json.dumps(selected_turn_todo(envelope), sort_keys=True) == json.dumps(_SELECTED, sort_keys=True)
+
+
+@pytest.mark.parametrize("action", [None, [], {}, {"selected_todo": None}, {"selected_todo": []}])
+def test_absent_or_nonobject_selected_work_stays_empty(action: Any) -> None:
+    assert selected_turn_todo({"action": action}) == {}
+
+
+def test_selected_result_is_detached_from_original_record() -> None:
+    envelope = {"action": {"selected_todo": deepcopy(_SELECTED)}}
+    selected = selected_turn_todo(envelope)
+    selected["todo_id"] = "changed"
+    assert envelope["action"]["selected_todo"]["todo_id"] == "fallback"
+
+
+def test_large_selected_record_retains_original_number_tokens_through_snapshot_transport() -> None:
+    selected = {**deepcopy(_SELECTED), "text": '"\\' * 1_100_000}
+    envelope = {"action": {"selected_todo": selected}}
+    assert json.dumps(selected_turn_todo(envelope), sort_keys=True) == json.dumps(selected, sort_keys=True)

tests/control_plane_ts/interaction_contract.test.ts

@@ -9,6 +9,8 @@ import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
 import { interpretQuotaShouldRunPacket } from "../../loopx/control_plane/effect_program.ts";
 import {
   decodeInteractionContract,
+  selectedWorkTodo,
+  selectedWorkTodoId,
   projectInteractionExecution,
   INTERACTION_EXECUTION_FACTS_SCHEMA_VERSION,
   projectInteractionRequiredReads,
@@ -19,6 +21,23 @@ import {
 } from "../../loopx/control_plane/work_items/interaction_contract.ts";
 import { evaluateTodoResumeConditions } from "../../loopx/control_plane/todos/resume_condition.ts";
 
+test("shared selection keeps current typed identity separate from historical decoding", () => {
+  const selected = {todo_id: "fallback", future: {keep: true}};
+  const packet = {selected_todo: selected, task_orchestration_contract: {
+    schema_version: "task_orchestration_contract_v2", mode: "adaptive", primary_todo_id: " primary "}};
+  assert.deepEqual(selectedWorkTodo(packet),
+    {todo_id: "primary", source: "task_orchestration_contract.primary_todo_id"});
+  assert.equal(selectedWorkTodoId(packet), "primary");
+  assert.deepEqual(selectedWorkTodo(packet, "\uFEFFprimary\uFEFF"),
+    {todo_id: "\uFEFFprimary\uFEFF", source: "task_orchestration_contract.primary_todo_id"});
+  assert.strictEqual(selectedWorkTodo(packet, ""), selected);
+  const numeric = {...packet, task_orchestration_contract: {...packet.task_orchestration_contract, primary_todo_id: 7}};
+  assert.strictEqual(selectedWorkTodo(numeric), selected);
+  assert.equal(selectedWorkTodoId(numeric), "fallback");
+  assert.deepEqual(selectedWorkTodo(numeric, "7"),
+    {todo_id: "7", source: "task_orchestration_contract.primary_todo_id"});
+});
+
 test("only a positively qualified unbound scheduled wait can omit observation", () => {
   const todo = {todo_id: "todo_scheduled", status: "deferred", role: "agent",
     task_class: "advancement_task", resume_when: "resume_at:2099-01-01T00:00:00Z"};