refactor(scheduler): compose TypeScript arbitration with interaction reads

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -1205,6 +1205,22 @@ eligible fallback recommendation cannot hide a blocked task's real capability ch
 explicit selection. Success re-enters the same Turn; failure still allows explicit fallback
 selection. A committed receipt's Todo remains bound.
 
+Scheduler arbitration now composes with required-read projection in
+`work_items/interaction_contract.ts`. Structural diagnostics, shared consistency
+checks and disposition selection use one native owner. Quota consumes the
+private `{required_reads, arbitration}` result without adding public contract
+fields. The Python selector and standalone consistency RPC are removed. Later
+cooldown and recovery builds retain the dict API and discard unused arbitration;
+they do not replace the scheduler hint's earlier decision. Complete fact packets
+up to 64 KiB of compact ASCII JSON use inline transport. Passthrough reads,
+at most three generated reads, shell quoting and bounded diagnostics keep the
+reply within `16B + 8 KiB`, where B is the complete compact ASCII JSON byte
+count. This stays below 2 MiB for `B <= 64 KiB`. Larger facts use the existing
+private snapshot path without truncation. The 2 MiB wire and 64 MiB
+snapshot limits stay unchanged. Python retains scalar/Mapping codecs,
+the Enum/dataclass result adapter, scheduler cadence and Host IO until those
+consumers migrate. This does not close whole-scheduler or installed adoption.
+
 `agents/capability_memory.ts` now owns host-local runtime declarations, validation,
 atomic merge and idempotent persistence through the existing mutation lock and durable
 JSON writer. `capability_gate.ts` owns Goal/Agent/invocation precedence and local

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -856,6 +856,19 @@ argv；同一 interaction packet 复用结果，重入投影本身在健康路
 选择前，可执行的低优先级推荐不能隐藏受阻任务的真实能力验证；验证成功在原 Turn 重入，
 失败后仍可显式选择其他工作。已提交 receipt 的 Todo 不变。
 
+调度仲裁现在与 required-read 投影在 `work_items/interaction_contract.ts` 内组合。
+结构诊断、共享一致性检查和 disposition 选择由同一个原生 owner 负责。Quota 消费私有
+`{required_reads, arbitration}` 结果，不增加公共 contract 字段。Python 选择规则与独立
+consistency RPC 已删除。后续 cooldown 与 recovery 构造仍返回 dict，并丢弃未消费的
+仲裁；它们不覆盖 scheduler hint 较早的决策。完整 fact 的 compact ASCII JSON
+不超过 64 KiB 时使用 inline 通道。原 reads 透传、最多三条生成 read、shell 引号展开和
+有界诊断给出 `16B + 8 KiB` 保守结果上界；B 为完整 compact ASCII JSON 的字节数。
+当 `B <= 64 KiB` 时，该上界低于 2 MiB。
+更大的 fact 继续走既有私有快照，不截断数据。2 MiB wire 与 64 MiB snapshot 上限不变。
+Python 暂留 scalar/Mapping codec、
+Enum/dataclass 结果适配、调度 cadence 和 Host IO，直到对应 consumer 迁移。
+这不代表整个 scheduler 或正式安装采用已完成。
+
 `agents/capability_memory.ts` 现在持有本机 Agent 运行时能力声明、校验与幂等合并，
 复用现有文件锁及 durable JSON writer；`capability_gate.ts` 统一 Goal、Agent、本次
 调用的继承规则及不可用覆盖，删除 Python 的重复 union 规则。Live quota 与实际执行

examples/control_plane/interaction-contract-state-machine-smoke.py

@@ -23,6 +23,7 @@
     build_execution_obligation,
 )
 from loopx.control_plane.work_items.interaction_contract import (  # noqa: E402
+    _build_interaction_contract_projection,
     build_interaction_contract,
     user_channel_action_required,
 )
@@ -182,16 +183,15 @@ def base_payload(
 
 
 def finalize(payload: dict[str, Any]) -> dict[str, Any]:
-    payload["interaction_contract"] = build_interaction_contract(
+    payload["interaction_contract"], arbitration = _build_interaction_contract_projection(
         payload,
+        agent_scope_frontier_actions=[action.value for action in AgentScopeFrontierAction],
         scheduler_execution_context=APP_SCHEDULER_CONTEXT,
     )
     payload["scheduler_hint"] = build_scheduler_hint(
         payload,
         user_action_required=user_channel_action_required(payload),
-        agent_scope_frontier_actions=[
-            action.value for action in AgentScopeFrontierAction
-        ],
+        arbitration=arbitration,
         scheduler_execution_context=APP_SCHEDULER_CONTEXT,
     )
     return payload

examples/control_plane/quota-scheduler-hint-compaction-smoke.py

@@ -24,6 +24,10 @@
 )
 from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint  # noqa: E402
 from loopx.control_plane.quota.should_run_packet import _scheduler_hint  # noqa: E402
+from loopx.control_plane.agents.agent_scope_frontier import AgentScopeFrontierAction  # noqa: E402
+from loopx.control_plane.work_items.interaction_contract import (  # noqa: E402
+    _interaction_reads_and_arbitration,
+)
 
 
 RUNTIME_KEYS = (
@@ -81,6 +85,7 @@ def payload(*, should_run: bool, recommended_mode: str = "", user_required: bool
                 "delivery_allowed": should_run,
                 "quiet_noop_allowed": not user_required and not should_run,
             },
+            "cli_channel": {},
         },
     }
 
@@ -277,22 +282,33 @@ def assert_compact_runtime_policy_complete(
 
 
 def assert_compact_scheduler(name: str, source_payload: dict) -> None:
+    source_payload = deepcopy(source_payload)
+    _, arbitration = _interaction_reads_and_arbitration(
+        source_payload,
+        source_payload["interaction_contract"],
+        runtime_root=None,
+        registry_path=None,
+        agent_scope_frontier_actions=[action.value for action in AgentScopeFrontierAction],
+    )
     fixed_now = datetime(2026, 9, 12, 0, 0, tzinfo=timezone.utc)
     with patch(
         "loopx.control_plane.scheduler.scheduler_hint.now_utc",
         return_value=fixed_now,
     ):
         compact = build_scheduler_hint(
             deepcopy(source_payload),
+            arbitration=arbitration,
             user_action_required=False,
             scheduler_execution_context=APP_SCHEDULER_CONTEXT,
         )
         wrapper = _scheduler_hint(
             deepcopy(source_payload),
+            arbitration=arbitration,
             scheduler_execution_context=APP_SCHEDULER_CONTEXT,
         )
         detailed = build_scheduler_hint(
             deepcopy(source_payload),
+            arbitration=arbitration,
             user_action_required=False,
             include_detail=True,
             scheduler_execution_context=APP_SCHEDULER_CONTEXT,

examples/control_plane/quota-scheduler-policy-smoke.py

@@ -19,6 +19,9 @@
 from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint  # noqa: E402
 from loopx.control_plane.agents.agent_scope_frontier import AgentScopeFrontierAction  # noqa: E402
 from loopx.control_plane.quota.should_run_packet import _scheduler_hint  # noqa: E402
+from loopx.control_plane.work_items.interaction_contract import (  # noqa: E402
+    _interaction_reads_and_arbitration,
+)
 
 
 AGENT_SCOPE_ACTIONS = [action.value for action in AgentScopeFrontierAction]
@@ -72,6 +75,7 @@ def payload(
                 "delivery_allowed": must_attempt,
                 "quiet_noop_allowed": not user_required and not must_attempt,
             },
+            "cli_channel": {},
         },
     }
 
@@ -85,8 +89,17 @@ def assert_policy_case(
     expected_progression: list[int] | None = None,
     expected_same_identity_action: str = "advance_index_after_applied_interval_elapsed",
 ) -> None:
+    base_payload = deepcopy(base_payload)
+    _, arbitration = _interaction_reads_and_arbitration(
+        base_payload,
+        base_payload["interaction_contract"],
+        runtime_root=None,
+        registry_path=None,
+        agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
+    )
     quota_wrapper = _scheduler_hint(
         deepcopy(base_payload),
+        arbitration=arbitration,
         scheduler_execution_context=CODEX_APP_SCHEDULER_CONTEXT,
     )
     extracted = build_scheduler_hint(
@@ -96,7 +109,7 @@ def assert_policy_case(
             .get("user_channel", {})
             .get("action_required")
         ),
-        agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
+        arbitration=arbitration,
         scheduler_execution_context=CODEX_APP_SCHEDULER_CONTEXT,
     )
     detailed = build_scheduler_hint(
@@ -106,7 +119,7 @@ def assert_policy_case(
             .get("user_channel", {})
             .get("action_required")
         ),
-        agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
+        arbitration=arbitration,
         include_detail=True,
         scheduler_execution_context=CODEX_APP_SCHEDULER_CONTEXT,
     )

loopx/control_plane/effect_runtime.py

@@ -46,6 +46,7 @@
     "coordination.local_authority.todo_source",
     "todo.context.page",
     "work_item.context.project",
+    "work_item.interaction_reads.project",
     "work_item.task_graph.goal_topology",
     "work_item.replan_semantics.project",
     "quota.settlement.read",

loopx/control_plane/effect_runtime_handlers.ts

@@ -783,7 +783,6 @@ export function createEffectRuntimeHandlers(
     ["turn.settlement.reduce", lazyHandler(() => import("./turn_driver/settlement.ts"), ({reduceTurnSettlementTransaction}) => reduceTurnSettlementTransaction)],
     ["turn.host_todo_completion.evaluate", lazyHandler(() => import("./turn_driver/host_todo_completion.ts"), ({evaluateHostTodoCompletion}) => evaluateHostTodoCompletion)],
     ["work_item.interaction_reads.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionRequiredReads}) => projectInteractionRequiredReads)],
-    ["work_item.interaction_consistency.validate", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionConsistencyValidation}) => projectInteractionConsistencyValidation)],
     ["work_item.blocked_wait_observation.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectBlockedSuccessorWaitObservation}) => projectBlockedSuccessorWaitObservation)],
     ["work_item.context.plan", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({planInteractionWorkContext}) => planInteractionWorkContext)],
     ["work_item.context.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionWorkContext}) => projectInteractionWorkContext)],

loopx/control_plane/effect_runtime_server.ts

@@ -32,6 +32,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([
   "coordination.local_authority.todo_source",
   "todo.context.page",
   "work_item.context.project",
+  "work_item.interaction_reads.project",
   "work_item.task_graph.goal_topology",
   "work_item.replan_semantics.project",
   "quota.settlement.read",

loopx/control_plane/quota/should_run.py

@@ -44,7 +44,7 @@
 )
 from ..todos.write_hint import build_todo_write_hint
 from ..work_items.interaction_contract import (
-    build_interaction_contract,
+    _build_interaction_contract_projection,
 )
 from .effect_program import (
     ReceiptBoundMonitorPhase,
@@ -283,8 +283,8 @@ def build_quota_paused_should_run_payload(
         resolved_scheduler_context=resolved_scheduler_context,
     )
     payload["automation_liveness"] = build_automation_liveness(payload)
-    payload["interaction_contract"] = build_interaction_contract(
-        payload,
+    payload["interaction_contract"], arbitration = _build_interaction_contract_projection(
+        payload, agent_scope_frontier_actions=[action.value for action in AgentScopeFrontierAction],
         available_capabilities=None,
         scheduler_execution_context=resolved_scheduler_context,
         runtime_root=(
@@ -296,7 +296,7 @@ def build_quota_paused_should_run_payload(
         ),
     )
     payload["scheduler_hint"] = _scheduler_hint(
-        payload,
+        payload, arbitration=arbitration,
         include_detail=include_scheduler_detail,
         available_capabilities=None,
         codex_app_current_rrule=codex_app_current_rrule,

loopx/control_plane/quota/should_run_packet.py

@@ -87,6 +87,7 @@
 from ..scheduler.external_evidence_observation import (
     build_external_evidence_observation_obligation,
 )
+from ..scheduler.arbitration import SchedulerArbitration
 from ..scheduler.scheduler_hint import build_scheduler_hint
 from ..scheduler.state import (
     load_app_automation_scheduler_state,
@@ -120,7 +121,7 @@
 )
 from ..work_items.goal_route_hint import build_goal_route_hint
 from ..work_items.interaction_contract import (
-    build_interaction_contract,
+    _build_interaction_contract_projection,
     unadmitted_action_selection,
     finalize_user_gate_notification_cooldown,
 )
@@ -196,15 +197,15 @@ def _compact_autonomous_candidate_context(
 
 
 def _scheduler_hint(
-    payload: dict[str, Any], *, include_detail: bool = False,
+    payload: dict[str, Any], *, arbitration: SchedulerArbitration | None = None, include_detail: bool = False,
     codex_app_scheduler_state: dict[str, Any] | None = None, available_capabilities: Any = None, codex_app_current_rrule: Any = None,
     codex_app_automation_id: Any = None,
     scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution | None = None,
 ) -> dict[str, Any]:
     return build_scheduler_hint(
         payload,
         user_action_required=_user_channel_action_required(payload),
-        agent_scope_frontier_actions=[action.value for action in AgentScopeFrontierAction],
+        arbitration=arbitration,
         include_detail=include_detail,
         codex_app_scheduler_state=codex_app_scheduler_state,
         available_capabilities=available_capabilities, codex_app_current_rrule=codex_app_current_rrule,
@@ -1650,16 +1651,16 @@ def _build_quota_should_run_payload(
         resolved_scheduler_context=prepared.resolved_scheduler_context,
     )
     payload["automation_liveness"] = build_automation_liveness(payload)
-    payload["interaction_contract"] = build_interaction_contract(
-        payload,
+    payload["interaction_contract"], arbitration = _build_interaction_contract_projection(
+        payload, agent_scope_frontier_actions=[action.value for action in AgentScopeFrontierAction],
         available_capabilities=prepared.runtime_available_capabilities,
         scheduler_execution_context=prepared.resolved_scheduler_context,
         turn_instance_id=turn_instance_id,
         runtime_root=_interaction_runtime_root(runtime_root, prepared.status_payload),
         registry_path=prepared.status_payload.get("registry"),
     )
     payload["scheduler_hint"] = _scheduler_hint(
-        payload,
+        payload, arbitration=arbitration,
         include_detail=prepared.include_scheduler_detail,
         available_capabilities=prepared.runtime_available_capabilities,
         codex_app_scheduler_state=(