From 198305565d7df3690e041095a71b09a9401b59ed Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:03:04 +0800 Subject: [PATCH 1/6] refactor(scheduler): compose native arbitration with interaction reads Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- loopx/control_plane/effect_runtime.py | 1 + .../control_plane/effect_runtime_handlers.ts | 1 - loopx/control_plane/effect_runtime_server.ts | 1 + loopx/control_plane/quota/should_run.py | 8 +- .../control_plane/quota/should_run_packet.py | 13 +- loopx/control_plane/scheduler/arbitration.py | 184 ++++-------------- .../control_plane/scheduler/scheduler_hint.py | 12 +- .../work_items/interaction_contract.py | 43 +++- .../work_items/interaction_contract.ts | 70 +++++-- .../scheduler_arbitration_fixture.py | 34 ++++ .../test_blocked_retry_wait_context.py | 4 +- .../test_scheduler_arbitration_native_cli.py | 167 ++++++++++++++++ .../test_scheduler_arbitration_transport.py | 109 +++++++++++ .../test_scheduler_backoff_convergence.py | 2 +- .../test_scheduler_execution_context.py | 24 ++- .../test_scheduler_fallback_hint.py | 2 +- .../test_scheduler_host_failure_cache.py | 2 +- .../test_scheduler_interaction_arbitration.py | 23 ++- .../test_settled_replay_construction.py | 21 +- .../test_user_gate_lane_progress.py | 2 +- .../interaction_contract.test.ts | 26 +-- .../scheduler_arbitration.test.ts | 143 ++++++++++++++ tests/test_automation_cadence.py | 2 +- tests/test_host_loop_runtime_parity.py | 4 +- tsconfig.control-plane.json | 1 + 25 files changed, 657 insertions(+), 242 deletions(-) create mode 100644 tests/control_plane/scheduler_arbitration_fixture.py create mode 100644 tests/control_plane/test_scheduler_arbitration_native_cli.py create mode 100644 tests/control_plane/test_scheduler_arbitration_transport.py create mode 100644 tests/control_plane_ts/scheduler_arbitration.test.ts diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 823c224e4d..0eaf21ded6 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -45,6 +45,7 @@ "coordination.local_authority.todo_source", "todo.context.page", "work_item.context.project", + "work_item.interaction_reads.project", "work_item.replan_semantics.project", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index b4ccd3c6c7..bdbb4e08c9 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -768,7 +768,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)], diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 7f024eed59..263f57d2a9 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -31,6 +31,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.replan_semantics.project", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py index c9f9cc90b0..3bcd265861 100644 --- a/loopx/control_plane/quota/should_run.py +++ b/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, diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index f81bda1c3b..e983d56663 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/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,7 +197,7 @@ 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, @@ -204,7 +205,7 @@ def _scheduler_hint( 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, @@ -1648,8 +1649,8 @@ 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, @@ -1657,7 +1658,7 @@ def _build_quota_should_run_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=( diff --git a/loopx/control_plane/scheduler/arbitration.py b/loopx/control_plane/scheduler/arbitration.py index 22893d1936..d52e875f92 100644 --- a/loopx/control_plane/scheduler/arbitration.py +++ b/loopx/control_plane/scheduler/arbitration.py @@ -5,9 +5,6 @@ from enum import Enum from typing import Any -from ..work_items.interaction_contract import INTERACTION_CONTRACT_SCHEMA_VERSION - - SCHEDULER_ARBITRATION_SCHEMA_VERSION = "scheduler_arbitration_v0" @@ -50,151 +47,38 @@ def consistency_error(self) -> dict[str, Any] | None: ), } - -def _mapping(value: Any) -> Mapping[str, Any]: - return value if isinstance(value, Mapping) else {} - - -def _required_bool( - source: Mapping[str, Any], - field: str, - *, - error_prefix: str, - errors: list[str], -) -> bool: - value = source.get(field) - if type(value) is not bool: - errors.append(f"{error_prefix}.{field}_must_be_boolean") - return False - return value - - -def _classify_disposition( - *, - mode: str, - user_required: bool, - must_attempt: bool, - quiet_noop_allowed: bool, - agent_scope_frontier_actions: Collection[str], - non_gating_notice_retry: bool = False, -) -> tuple[SchedulerDisposition, str]: - agent_scope_modes = {str(value) for value in agent_scope_frontier_actions} - if mode == "terminal_no_followup": - return SchedulerDisposition.TERMINAL_STOP, mode - if mode == "peer_coordination_blocked": - return SchedulerDisposition.PEER_COORDINATION_WAIT, mode - if mode == "agent_monitor_only": - return SchedulerDisposition.AGENT_MONITOR_ONLY_WAIT, mode - if user_required and not must_attempt and non_gating_notice_retry: - return SchedulerDisposition.QUIET_WAIT, "blocked_wait" - if user_required and not must_attempt: - return SchedulerDisposition.HUMAN_GATE, "interaction_blocking_user_gate" - if mode == "monitor_quiet_skip": - return SchedulerDisposition.MONITOR_WAIT, "interaction_monitor_quiet_wait" - if mode == "heartbeat_settled_skip": - return SchedulerDisposition.SETTLED_REPLAY, "interaction_heartbeat_settled_replay" - if mode == "successor_replan_required" and must_attempt: - return SchedulerDisposition.ACTIVE_WORK, "interaction_successor_replan_required" - if mode in agent_scope_modes: - return SchedulerDisposition.AGENT_SCOPE_WAIT, "interaction_agent_scope_wait" - if mode == "mapped_noop_if_unchanged": - return SchedulerDisposition.UNCHANGED_WAIT, "interaction_unchanged_wait" - if must_attempt: - return SchedulerDisposition.ACTIVE_WORK, "interaction_agent_attempt_required" - if quiet_noop_allowed: - return SchedulerDisposition.QUIET_WAIT, "interaction_quiet_noop_allowed" - return SchedulerDisposition.QUIET_WAIT, "interaction_delivery_not_allowed" - - -def build_scheduler_arbitration( - payload: Mapping[str, Any], - *, - agent_scope_frontier_actions: Collection[str] = (), -) -> SchedulerArbitration: - """Derive scheduler authority from the final interaction contract. - - Lower-level quota fields never participate in branch selection. Structural - contradictions inside the final contract fail closed to projection repair. - """ - - errors: list[str] = [] - contract = _mapping(payload.get("interaction_contract")) - schema_version = contract.get("schema_version") - if schema_version != INTERACTION_CONTRACT_SCHEMA_VERSION: - errors.append("interaction_contract.schema_version_mismatch") - - mode_value = contract.get("mode") - mode = str(mode_value or "").strip() - if not mode: - errors.append("interaction_contract.mode_missing") - - user_channel = _mapping(contract.get("user_channel")) - agent_channel = _mapping(contract.get("agent_channel")) - if not user_channel: - errors.append("interaction_contract.user_channel_missing") - if not agent_channel: - errors.append("interaction_contract.agent_channel_missing") - - user_required = _required_bool( - user_channel, - "action_required", - error_prefix="interaction_contract.user_channel", - errors=errors, - ) - must_attempt = _required_bool( - agent_channel, - "must_attempt", - error_prefix="interaction_contract.agent_channel", - errors=errors, - ) - delivery_allowed = _required_bool( - agent_channel, - "delivery_allowed", - error_prefix="interaction_contract.agent_channel", - errors=errors, - ) - quiet_noop_allowed = _required_bool( - agent_channel, - "quiet_noop_allowed", - error_prefix="interaction_contract.agent_channel", - errors=errors, - ) - - from ..effect_runtime import effect_runtime_result - - cli_channel = _mapping(contract.get("cli_channel")) - spend_flags = {field: _required_bool( - cli_channel, field, error_prefix="interaction_contract.cli_channel", errors=errors, - ) if field in cli_channel else False for field in ("spend_allowed_now", "spend_after_validation")} - validation = effect_runtime_result("work_item.interaction_consistency.validate", { - "mode": mode, "user_required": user_required, "must_attempt": must_attempt, - "delivery_allowed": delivery_allowed, "quiet_noop_allowed": quiet_noop_allowed, - **spend_flags, - }) - if not isinstance(validation, dict) or not isinstance(validation.get("errors"), list) or any( - not isinstance(error, str) for error in validation["errors"] + @classmethod + def from_projection(cls, projection: Any) -> SchedulerArbitration: + if (not isinstance(projection, Mapping) + or any(not isinstance(projection.get(key), str) for key in ("disposition", "reason_code", "mode")) + or not isinstance(projection.get("errors"), list) + or any(not isinstance(error, str) for error in projection["errors"])): + raise TypeError("TypeScript scheduler arbitration shape mismatch") + return cls(SchedulerDisposition(projection["disposition"]), projection["reason_code"], + projection["mode"], tuple(projection["errors"])) + + +def _scheduler_arbitration_facts( + contract: Any, agent_scope_frontier_actions: Collection[str] = (), +) -> dict[str, Any]: + """Keep legacy Python scalar/Mapping codecs; transport only consumed facts.""" + contract = contract if isinstance(contract, Mapping) else {} + channels = {} + for name, fields in ( + ("user_channel", ("action_required", "non_gating_notice")), + ("agent_channel", ("must_attempt", "delivery_allowed", "quiet_noop_allowed")), + ("cli_channel", ("spend_allowed_now", "spend_after_validation")), ): - raise TypeError("interaction consistency validation returned invalid diagnostics") - errors.extend(validation["errors"]) - disposition, reason_code = _classify_disposition( - mode=mode, - user_required=user_required, - must_attempt=must_attempt, - quiet_noop_allowed=quiet_noop_allowed, - agent_scope_frontier_actions=agent_scope_frontier_actions, - non_gating_notice_retry=(user_channel.get("non_gating_notice") is True and - _mapping(agent_channel.get("wait")).get("source") == "turn_settlement"), - ) - - if errors: - return SchedulerArbitration( - disposition=SchedulerDisposition.CONSISTENCY_REPAIR, - reason_code="scheduler_interaction_contract_inconsistent", - mode=mode, - errors=tuple(dict.fromkeys(errors)), - ) - return SchedulerArbitration( - disposition=disposition, - reason_code=reason_code, - mode=mode, - ) + source = contract.get(name) + source = source if isinstance(source, Mapping) else {} + channels[name] = {"present": bool(source), **{ + field: source[field] if type(source[field]) is bool else None + for field in fields if field in source + }} + agent = contract.get("agent_channel") + wait = agent.get("wait") if isinstance(agent, Mapping) else None + source = wait.get("source") if isinstance(wait, Mapping) else None + channels["agent_channel"]["wait_source"] = source if isinstance(source, str) and len(source) <= 64 else None + return {"schema_version": contract.get("schema_version"), + "mode": str(contract.get("mode") or "").strip(), **channels, + "agent_scope_frontier_actions": [str(action) for action in agent_scope_frontier_actions]} diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index 857b8fe114..5e77275e35 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -7,7 +7,7 @@ import json import re import zlib -from collections.abc import Collection, Mapping +from collections.abc import Mapping from dataclasses import dataclass from datetime import timedelta from typing import Any @@ -20,7 +20,6 @@ from .arbitration import ( SchedulerArbitration, SchedulerDisposition, - build_scheduler_arbitration, ) from .app_automation_compat import ( CODEX_APP_SCHEDULER_ACK_HINT_SCHEMA_VERSION, @@ -1296,7 +1295,7 @@ def build_scheduler_hint( payload: dict[str, Any], *, user_action_required: bool = False, - agent_scope_frontier_actions: Collection[str] = (), + arbitration: SchedulerArbitration | None = None, include_detail: bool = False, codex_app_scheduler_state: dict[str, Any] | None = None, available_capabilities: Any = None, @@ -1440,11 +1439,8 @@ def build_scheduler_hint( if isinstance(capability_gate.get("available"), list) else [] ) - agent_scope_action_set = {str(value) for value in agent_scope_frontier_actions} - arbitration = build_scheduler_arbitration( - payload, - agent_scope_frontier_actions=agent_scope_action_set, - ) + if arbitration is None: + raise ValueError("scheduler arbitration must come from the interaction projection") automatic_cadence_wait = _automatic_cadence_wait_eligibility(payload) if ( automatic_cadence_wait is not None diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index f0e44342a4..8456138926 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -4,7 +4,7 @@ from ..effect_runtime import effect_runtime_result import shlex import typing -from collections.abc import Mapping +from collections.abc import Collection, Mapping from typing import Any from ..agents.agent_scope_frontier import AgentScopeFrontierAction, agent_scope_frontier_action as _agent_scope_frontier_action @@ -21,6 +21,7 @@ host_goal_turn_reentry_action, quota_spend_source_for_execution_context, ) +from ..scheduler.arbitration import SchedulerArbitration, _scheduler_arbitration_facts from ..scheduler.execution_context import ( APP_HEARTBEAT_SETTLEMENT_RUNTIME_PROFILES, SchedulerExecutionContextResolution, @@ -1087,10 +1088,11 @@ def _admitted_interaction_next_cli_actions( ) -def _interaction_required_reads( +def _interaction_reads_and_arbitration( payload: dict[str, Any], contract: dict[str, Any], *, runtime_root: str | None, registry_path: str | None, -) -> list[dict[str, Any]]: + agent_scope_frontier_actions: Collection[str], +) -> tuple[list[dict[str, Any]], SchedulerArbitration]: """Transport source/admission facts to the existing typed interaction owner.""" reads = payload.get("required_reads") summary = payload.get("agent_todo_summary") or {} @@ -1099,6 +1101,7 @@ def _interaction_required_reads( selected = payload.get("selected_todo") or {} orchestration = payload.get("task_orchestration_contract") or {} result = effect_runtime_result("work_item.interaction_reads.project", { + "arbitration_facts": _scheduler_arbitration_facts(contract, agent_scope_frontier_actions), "required_reads": [dict(item) for item in reads if isinstance(item, dict) and item.get("command")] if isinstance(reads, list) else [], "goal_id": payload.get("goal_id"), @@ -1115,13 +1118,13 @@ def _interaction_required_reads( "command_prefix": selection.render_cli_command_prefix( runtime_root=runtime_root or payload.get("runtime_root"), registry_path=registry_path or payload.get("registry")), - }) + }, large_local_snapshot=True) projected = result.get("required_reads") if not isinstance(projected, list) or any( not isinstance(item, dict) or not item.get("command") for item in projected ): raise RuntimeError("TypeScript interaction required reads are malformed") - return projected + return projected, SchedulerArbitration.from_projection(result.get("arbitration")) def _interaction_spend_policy( @@ -1704,9 +1707,10 @@ def unadmitted_action_selection(payload: dict[str, Any]) -> bool: ) -def build_interaction_contract( +def _build_interaction_contract_projection( payload: dict[str, Any], *, + agent_scope_frontier_actions: Collection[str], available_capabilities: Any = None, scheduler_execution_context: ( Mapping[str, Any] | SchedulerExecutionContextResolution | None @@ -1714,7 +1718,7 @@ def build_interaction_contract( turn_instance_id: str | None = None, runtime_root: str | None = None, registry_path: str | None = None, -) -> InteractionContractPacket: +) -> tuple[InteractionContractPacket, SchedulerArbitration]: execution_obligation = ( payload.get("execution_obligation") if isinstance(payload.get("execution_obligation"), dict) @@ -1814,11 +1818,30 @@ def build_interaction_contract( ) if response_plan is not None: contract["response_plan"] = response_plan - _attach_interaction_required_reads(contract, _interaction_required_reads( - payload, contract, runtime_root=runtime_root, registry_path=registry_path)) + required_reads, arbitration = _interaction_reads_and_arbitration( + payload, contract, runtime_root=runtime_root, registry_path=registry_path, + agent_scope_frontier_actions=agent_scope_frontier_actions) + _attach_interaction_required_reads(contract, required_reads) _attach_interaction_post_writeback_actions(contract, payload) _attach_interaction_vision_continuation_audit(contract, payload) _attach_interaction_vision_wait_state(contract, payload) if _interaction_fallback_policy_required(payload, mode=mode): contract["fallback_policy"] = {"do_not_cancel_on_block": True} - return typing.cast(InteractionContractPacket, contract) + return typing.cast(InteractionContractPacket, contract), arbitration + + +def build_interaction_contract( + payload: dict[str, Any], + *, + available_capabilities: Any = None, + scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution | None = None, + turn_instance_id: str | None = None, + runtime_root: str | None = None, + registry_path: str | None = None, +) -> InteractionContractPacket: + contract, _arbitration = _build_interaction_contract_projection( + payload, agent_scope_frontier_actions=[action.value for action in AgentScopeFrontierAction], + available_capabilities=available_capabilities, scheduler_execution_context=scheduler_execution_context, + turn_instance_id=turn_instance_id, runtime_root=runtime_root, registry_path=registry_path, + ) + return contract diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index f331b50742..fd8bb5c675 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -122,18 +122,58 @@ function interactionConsistencyIssues(facts: InteractionConsistencyFacts) { return issues; } -/** Python normalizes its legacy structural diagnostics, then asks this owner. */ -export function projectInteractionConsistencyValidation(request: JsonObject): JsonObject { - if (typeof request.mode !== "string") throw new EffectRuntimeRequestError("interaction consistency mode must be a string"); - return {errors: interactionConsistencyIssues({ - mode: request.mode, - user_required: requireBoolean(request.user_required, "user_required"), - must_attempt: requireBoolean(request.must_attempt, "must_attempt"), - delivery_allowed: requireBoolean(request.delivery_allowed, "delivery_allowed"), - quiet_noop_allowed: requireBoolean(request.quiet_noop_allowed, "quiet_noop_allowed"), - spend_allowed_now: requireBoolean(request.spend_allowed_now, "spend_allowed_now"), - spend_after_validation: requireBoolean(request.spend_after_validation, "spend_after_validation"), - }).map(issue => issue.code)}; +function schedulerDisposition(facts: InteractionConsistencyFacts, modes: readonly string[], noticeRetry: boolean): [string, string] { + const {mode, user_required, must_attempt, quiet_noop_allowed} = facts; + if (mode === "terminal_no_followup") return ["terminal_stop", mode]; + if (mode === "peer_coordination_blocked") return ["peer_coordination_wait", mode]; + if (mode === "agent_monitor_only") return ["agent_monitor_only_wait", mode]; + if (user_required && !must_attempt && noticeRetry) return ["quiet_wait", "blocked_wait"]; + if (user_required && !must_attempt) return ["human_gate", "interaction_blocking_user_gate"]; + if (mode === "monitor_quiet_skip") return ["monitor_wait", "interaction_monitor_quiet_wait"]; + if (mode === "heartbeat_settled_skip") return ["settled_replay", "interaction_heartbeat_settled_replay"]; + if (mode === "successor_replan_required" && must_attempt) return ["active_work", "interaction_successor_replan_required"]; + if (modes.includes(mode)) return ["agent_scope_wait", "interaction_agent_scope_wait"]; + if (mode === "mapped_noop_if_unchanged") return ["unchanged_wait", "interaction_unchanged_wait"]; + if (must_attempt) return ["active_work", "interaction_agent_attempt_required"]; + return ["quiet_wait", quiet_noop_allowed ? "interaction_quiet_noop_allowed" : "interaction_delivery_not_allowed"]; +} + +/** Preserve accumulated structural diagnostics before shared consistency rules; + * malformed public contracts produce repair advice, never scheduler authority. */ +export function projectSchedulerArbitration(value: unknown): JsonObject { + const request = requireJsonObject(value, "scheduler arbitration facts"); + if (typeof request.mode !== "string") throw new EffectRuntimeRequestError("scheduler arbitration mode must be a string"); + const mode = request.mode; + const user = requireJsonObject(request.user_channel, "user channel facts"); + const agent = requireJsonObject(request.agent_channel, "agent channel facts"); + const cli = requireJsonObject(request.cli_channel, "CLI channel facts"); + const userPresent = requireBoolean(user.present, "user channel presence"); + const agentPresent = requireBoolean(agent.present, "agent channel presence"); + const modes = requireStringArray(request.agent_scope_frontier_actions, "agent scope frontier actions"); + const errors: string[] = []; + if (request.schema_version !== INTERACTION_CONTRACT_SCHEMA_VERSION) errors.push("interaction_contract.schema_version_mismatch"); + if (!mode) errors.push("interaction_contract.mode_missing"); + if (!userPresent) errors.push("interaction_contract.user_channel_missing"); + if (!agentPresent) errors.push("interaction_contract.agent_channel_missing"); + const flag = (channel: JsonObject, field: string, name: string): boolean => { + if (typeof channel[field] === "boolean") return channel[field]; + errors.push(`interaction_contract.${name}.${field}_must_be_boolean`); + return false; + }; + const facts: InteractionConsistencyFacts = { + mode, user_required: flag(user, "action_required", "user_channel"), + must_attempt: flag(agent, "must_attempt", "agent_channel"), + delivery_allowed: flag(agent, "delivery_allowed", "agent_channel"), + quiet_noop_allowed: flag(agent, "quiet_noop_allowed", "agent_channel"), + spend_allowed_now: Object.hasOwn(cli, "spend_allowed_now") ? flag(cli, "spend_allowed_now", "cli_channel") : false, + spend_after_validation: Object.hasOwn(cli, "spend_after_validation") ? flag(cli, "spend_after_validation", "cli_channel") : false, + }; + errors.push(...interactionConsistencyIssues(facts).map(issue => issue.code)); + const [disposition, reason] = schedulerDisposition(facts, modes, + user.non_gating_notice === true && agent.wait_source === "turn_settlement"); + return {mode, errors: [...new Set(errors)], + disposition: errors.length ? "consistency_repair" : disposition, + reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason}; } function decodeUserChannel(value: unknown): UserInteractionChannel { @@ -273,7 +313,7 @@ export function selectedWorkTodoId(packet: JsonObject): string | null { * list; transport projections must not independently infer requirements. * Python supplies registered source routing, never a second admission rule. */ -export function projectInteractionRequiredReads(request: JsonObject): JsonObject { +function interactionRequiredReads(request: JsonObject): JsonObject { if (!Array.isArray(request.required_reads)) { throw new EffectRuntimeRequestError("required_reads must be an array"); } @@ -309,6 +349,10 @@ export function projectInteractionRequiredReads(request: JsonObject): JsonObject return {required_reads: reads}; } +export function projectInteractionRequiredReads(request: JsonObject): JsonObject { + return {...interactionRequiredReads(request), arbitration: projectSchedulerArbitration(request.arbitration_facts)}; +} + /** Context routing reuses the admitted work identity; User scope remains owned * by the canonical Todo reader. No content confers execution authority. */ export function planInteractionWorkContext(request: JsonObject): JsonObject { diff --git a/tests/control_plane/scheduler_arbitration_fixture.py b/tests/control_plane/scheduler_arbitration_fixture.py new file mode 100644 index 0000000000..cb9f8427b4 --- /dev/null +++ b/tests/control_plane/scheduler_arbitration_fixture.py @@ -0,0 +1,34 @@ +"""Real native arbitration for tests of the unchanged scheduler projections.""" +from collections.abc import Collection +from typing import Any + +from loopx.control_plane.effect_runtime import effect_runtime_result +from loopx.control_plane.scheduler.arbitration import ( + SchedulerArbitration, + _scheduler_arbitration_facts, +) +from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + + +def project_native_arbitration( + payload: dict[str, Any], *, agent_scope_frontier_actions: Collection[str] = (), +) -> SchedulerArbitration: + result = effect_runtime_result("work_item.interaction_reads.project", { + "required_reads": [], "goal_id": None, "goal_state_file": None, + "goal_acceptance_enabled": False, "should_run": False, + "delivery_allowed": False, "selection_required": False, + "has_replan": False, "settlement_only": False, + "arbitration_facts": _scheduler_arbitration_facts( + payload.get("interaction_contract"), agent_scope_frontier_actions, + ), + }) + return SchedulerArbitration.from_projection(result["arbitration"]) + + +def build_native_scheduler_hint( + payload: dict[str, Any], *, agent_scope_frontier_actions: Collection[str] = (), + **kwargs: Any, +) -> dict[str, Any]: + return build_scheduler_hint(payload, arbitration=project_native_arbitration( + payload, agent_scope_frontier_actions=agent_scope_frontier_actions, + ), **kwargs) diff --git a/tests/control_plane/test_blocked_retry_wait_context.py b/tests/control_plane/test_blocked_retry_wait_context.py index 2ba2bf27d4..bfacf1b503 100644 --- a/tests/control_plane/test_blocked_retry_wait_context.py +++ b/tests/control_plane/test_blocked_retry_wait_context.py @@ -94,7 +94,7 @@ def packet(monkeypatch, *, seconds=120, user_class="user_action", newer=False, r scheduler_execution_context=scheduler_execution_context_for_runtime_profile( runtime or SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT)) if floor: - from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint result["automation_cadence"] = {"schema_version": "automation_cadence_result_v1", "ok": True, "enabled": True, "min_interval_minutes": floor} result["scheduler_hint"] = build_scheduler_hint(result, @@ -168,7 +168,7 @@ def test_newer_run_supersedes_retry(monkeypatch): def test_real_user_gate_survives_retry_clock(monkeypatch): # A real final gate contract wins over an independently pending retry. # Source Todo copy alone is not a final gate/admission contract. - from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint result = packet(monkeypatch) contract = result["interaction_contract"] contract["mode"] = "user_gate" diff --git a/tests/control_plane/test_scheduler_arbitration_native_cli.py b/tests/control_plane/test_scheduler_arbitration_native_cli.py new file mode 100644 index 0000000000..707df34a25 --- /dev/null +++ b/tests/control_plane/test_scheduler_arbitration_native_cli.py @@ -0,0 +1,167 @@ +"""Real quota entrypoints preserve scheduler authority and read-only boundaries.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from loopx.control_plane.coordination.canonical_snapshot import read_canonical_snapshot +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.effect_runtime import effect_runtime_result, restart_effect_runtime +from loopx.control_plane.scheduler.state import APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY, load_scheduler_state +from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos +from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + +GOAL = "scheduler-arbitration-cli-fixture" +AGENT = "scheduler-arbitration-agent" +TODO = "todo_scheduler_work" +TURN = "turn-scheduler-arbitration" + + +def make_scheduler_fixture(root: Path, provider: str, scenario: str) -> dict: + """Keep this synthetic source usable by source and installed-package replay.""" + project, runtime = root / "project", root / "runtime" + project.mkdir() + state = project / "ACTIVE_GOAL_STATE.md" + status = "active" + row = ( + "- [ ] [P1] Validate the bounded source artifact.\n" + f" \n" + ) + if scenario == "monitor": + row = ( + "- [ ] [P2-monitor] Observe the bounded source after its deadline.\n" + f" \n" + ) + gate = ( + "## User Todo\n\n- [ ] [P0-user] Approve the bounded delivery.\n" + " \n\n" + if scenario == "human_gate" else "" + ) + state.write_text(f"---\nstatus: {status}\n---\n\n# Synthetic scheduler Goal\n\n" + + gate + "## Agent Todo\n\n" + row) + registry = root / "registry.json" + registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{ + "id": GOAL, "status": status, "repo": str(project), "state_file": state.name, + "activation_state": "stopped" if scenario == "stopped" else "active", + "domain": "synthetic-scheduler", + "adapter": {"kind": "fixture_connected_delivery_v0", "status": "connected-delivery"}, + "quota": {"compute": 0.0 if scenario == "paused" else 1.0, "window_hours": 24}, + "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT]}, + }]})) + fields = parse_active_state_todos(state.read_text(), item_limit=None) + initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection( + goal_id=GOAL, todos=fields["agent_todos"]["items"] + fields.get("user_todos", {}).get("items", []), + handoff_mode="soft_claim", leases=[], + ), state_path=state, provider=provider) + context = ["--host-surface", "codex_app"] if scenario == "invalid_context" else ["--codex-app"] + guard = ["quota", "should-run", *context, "--goal-id", GOAL, "--agent-id", AGENT, + "--turn-instance-id", TURN, "--scan-path", str(project)] + return {"project": project, "runtime": runtime, "registry": registry, + "state": state, "guard": guard} + + +def call_scheduler_cli(fixture: dict, *args: str, expected_code: int = 0) -> dict: + result = subprocess.run([ + sys.executable, "-m", "loopx.cli", "--registry", str(fixture["registry"]), + "--runtime-root", str(fixture["runtime"]), "--format", "json", *args, + ], cwd=fixture["project"], env={**os.environ, "LOOPX_USAGE_PING": "0"}, + capture_output=True, text=True, timeout=60) + assert result.returncode == expected_code, (result.stdout, result.stderr) + return json.loads(result.stdout) + + +def scheduler_authority_snapshot(fixture: dict) -> dict: + result = read_canonical_snapshot(rpc=effect_runtime_result, runtime_root=str(fixture["runtime"]), + goal_id=GOAL, include_leases=True, projection_readback=None, timeout=20) + assert result["status"] == "loaded", result + return result + + +def _assert_no_private_arbitration(value: object) -> None: + if isinstance(value, dict): + assert not {"arbitration", "scheduler_arbitration", "_scheduler_arbitration"}.intersection(value) + for item in value.values(): + _assert_no_private_arbitration(item) + elif isinstance(value, list): + for item in value: + _assert_no_private_arbitration(item) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("scenario,mode,action,cadence,reason", [ + ("active", "bounded_delivery", "run_now", "active_work", "interaction_agent_attempt_required"), + ("human_gate", "user_gate", "backoff_waiting_for_user", "human_gate", "interaction_blocking_user_gate"), + ("monitor", "monitor_quiet_skip", "backoff_until_material_transition", "monitor_wait", "interaction_monitor_quiet_wait"), + ("paused", "skip", "stop_until_explicit_resume", "quota_paused", "quota_paused"), + ("stopped", "skip", "stop_until_explicit_resume", "goal_stopped", "goal_stopped"), + ("invalid_context", "bounded_delivery", "repair_scheduler_execution_context", "control_plane_repair", "invalid_scheduler_execution_context"), +]) +def test_quota_cli_keeps_scheduler_authority_without_premature_effects( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, + scenario: str, mode: str, action: str, cadence: str, reason: str, +) -> None: + isolate_sqlite_runtime(tmp_path, monkeypatch) + fixture = make_scheduler_fixture(tmp_path, provider, scenario) + try: + before = {key: fixture[key].read_bytes() for key in ("registry", "state")} + authority = scheduler_authority_snapshot(fixture) + packet = call_scheduler_cli(fixture, *fixture["guard"]) + hint = packet["scheduler_hint"] + assert packet["interaction_contract"]["mode"] == mode, packet + assert (hint["schema_version"], hint["action"], hint["cadence_class"], hint["reason_code"]) == ( + "scheduler_hint_v0", action, cadence, reason, + ) + _assert_no_private_arbitration(packet) + assert {key: fixture[key].read_bytes() for key in before} == before + assert scheduler_authority_snapshot(fixture) == authority + assert load_scheduler_state(fixture["runtime"], goal_id=GOAL, agent_id=AGENT, + state_key=APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY) is None + events = load_rollout_events(rollout_event_log_path(fixture["runtime"], GOAL)) + assert not {"refresh_state", "quota_spend", "scheduler_ack"}.intersection( + event["event_kind"] for event in events) + index = fixture["runtime"] / "goals" / GOAL / "runs" / "index.jsonl" + if scenario == "monitor": + rows = [json.loads(line) for line in index.read_text().splitlines()] + assert [(row["classification"], row["turn_instance_id"]) for row in rows] == [ + ("quota_monitor_poll", TURN)] + else: + assert not index.exists() + if scenario == "invalid_context": + assert hint["execution_phase"]["disposition"] == "contract_error" + assert hint["codex_app"]["apply"] == "none" + assert hint["consistency_error"]["source"] == "scheduler_execution_context" + elif scenario in {"paused", "stopped"}: + assert packet["should_run"] is False + assert hint["codex_app"]["ack_required"] is False + assert "recommended_rrule" not in hint["codex_app"] + if scenario == "active": + assert packet["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO + refresh = call_scheduler_cli(fixture, "refresh-state", "--goal-id", GOAL, + "--agent-id", AGENT, "--todo-id", TODO, "--turn-instance-id", TURN, + "--classification", "validated_intermediate_progress", "--delivery-batch-scale", "implementation", + "--delivery-outcome", "outcome_progress", "--delivery-boundary", "in_flight_continuation", + "--delivery-workspace-path", str(fixture["project"]), "--no-global-sync", "--suppress-external-sinks") + assert refresh["settlement_result"]["ok"] is True + spent = call_scheduler_cli(fixture, "quota", "spend-slot", "--goal-id", GOAL, + "--agent-id", AGENT, "--todo-id", TODO, "--turn-instance-id", TURN, + "--slots", "1", "--source", "heartbeat", "--execute") + assert spent["settlement_progress"]["state"] == "settled" + restart_effect_runtime(wait_for_exit=True) + replay = call_scheduler_cli(fixture, *fixture["guard"]) + assert replay["interaction_contract"]["mode"] == "heartbeat_settled_skip" + assert replay["should_run"] is False + assert replay["scheduler_hint"]["reason_code"] == "interaction_heartbeat_settled_replay" + assert scheduler_authority_snapshot(fixture) == authority + assert fixture["registry"].read_bytes() == before["registry"] + finally: + restart_effect_runtime(wait_for_exit=True) diff --git a/tests/control_plane/test_scheduler_arbitration_transport.py b/tests/control_plane/test_scheduler_arbitration_transport.py new file mode 100644 index 0000000000..73c725516b --- /dev/null +++ b/tests/control_plane/test_scheduler_arbitration_transport.py @@ -0,0 +1,109 @@ +"""Fused scheduler facts must retain complete reads at the ordinary wire boundary.""" +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest + +from loopx.control_plane import effect_runtime +from loopx.control_plane.scheduler.arbitration import ( + SchedulerArbitration, + SchedulerDisposition, + _scheduler_arbitration_facts, +) +from loopx.control_plane.work_items.interaction_contract import _interaction_reads_and_arbitration + + +def contract() -> dict: + return {"schema_version": "loopx_interaction_contract_v0", "mode": "bounded_delivery", + "user_channel": {"action_required": False}, + "agent_channel": {"must_attempt": True, "delivery_allowed": True, "quiet_noop_allowed": False}, + "cli_channel": {"spend_allowed_now": False, "spend_after_validation": False}} + + +def encoded_size(value: dict) -> int: + return len(json.dumps(value, separators=(",", ":")).encode()) + + +def assert_private_files_removed(root: Path) -> None: + assert list(root.glob("loopx-effect-*/params.json")) == [] + assert list(root.glob("loopx-effect-*/response.json")) == [] + + +@pytest.mark.parametrize("reject", [False, True], ids=["complete", "rejected"]) +def test_near_limit_read_survives_fused_arbitration_and_cleans_private_files( + isolated_effect_runtime, reject: bool, +) -> None: + root = Path(tempfile.gettempdir()) + read = {"command": "bad\0command" if reject else "read-complete-source", + "source": "existing", "evidence": ""} + # Match the legacy carrier and envelope sizes; nonce values themselves do not matter. + params = {"required_reads": [read], "goal_id": None, "goal_state_file": None, + "goal_acceptance_enabled": False, "selected_todo": {"todo_id": None}, + "task_orchestration_contract": {"schema_version": None, "mode": None, "primary_todo_id": None}, + "should_run": False, "delivery_allowed": True, "selection_required": False, + "has_replan": False, "settlement_only": False, "effective_action": None, + "command_prefix": "loopx"} + envelope = {"schema_version": effect_runtime.EFFECT_RUNTIME_REQUEST_SCHEMA_VERSION, + "token": "t" * 43, "request_id": "0" * 36, + "method": "work_item.interaction_reads.project", "params": params} + read["evidence"] = "x" * (effect_runtime.MAX_REQUEST_BYTES - encoded_size(envelope) - 128) + assert encoded_size(envelope) + 1 < effect_runtime.MAX_REQUEST_BYTES + fused = {**params, "arbitration_facts": _scheduler_arbitration_facts(contract())} + assert encoded_size({**envelope, "params": fused}) > effect_runtime.MAX_REQUEST_BYTES + try: + if reject: + with pytest.raises(effect_runtime.EffectRuntimeRejected, match="required read command contains NUL"): + _interaction_reads_and_arbitration({"required_reads": [read]}, contract(), + runtime_root=None, registry_path=None, agent_scope_frontier_actions=()) + else: + reads, arbitration = _interaction_reads_and_arbitration({"required_reads": [read]}, contract(), + runtime_root=None, registry_path=None, agent_scope_frontier_actions=()) + assert reads == [read] + assert arbitration == SchedulerArbitration(SchedulerDisposition.ACTIVE_WORK, + "interaction_agent_attempt_required", "bounded_delivery") + finally: + assert_private_files_removed(root) + assert effect_runtime.effect_runtime_result("runtime.ping", {})["ready"] is True + + +@pytest.mark.parametrize("reject", [False, True], ids=["complete", "rejected"]) +def test_small_request_retains_large_generated_read_reply_and_cleans_private_files( + isolated_effect_runtime, reject: bool, +) -> None: + root = Path(tempfile.gettempdir()) + runtime_root = "/" + "r" * (effect_runtime.MAX_RESPONSE_BYTES // 2) + prefix = "loopx --runtime-root " + runtime_root + payload = {"goal_id": "goal-wire", "should_run": True, + "agent_todo_summary": {"goal_acceptance_contract": {"enabled": True}}, + "selected_todo": {"todo_id": "todo_selected"}, + "required_reads": [{"command": "bad\0command"}] if reject else []} + # The input carries this prefix once; generated acceptance/Todo reads each retain it. + assert encoded_size({**payload, "command_prefix": prefix, + "arbitration_facts": _scheduler_arbitration_facts(contract())}) < effect_runtime.MAX_REQUEST_BYTES - 4096 + expected = [{ + "command": prefix + " --format json goal-acceptance inspect --goal-id goal-wire", + "source": "goal_acceptance", "ordering": "before_work", + "reason": "Read full configured objective, non-goals and criteria with scope/revision. Scoped acceptance does not replace Goal intent or prove completion; disabled/changed criteria require a fresh guard.", + }, { + "command": prefix + " --format json todo list --goal-id goal-wire --todo-id todo_selected", + "source": "selected_todo", "ordering": "before_work", + "reason": "Read full current requirements and status/claim. Require one matching active Todo; missing, ambiguous or changed work requires a fresh guard. A summary cannot replace this read.", + }] + assert encoded_size({"required_reads": expected}) > effect_runtime.MAX_RESPONSE_BYTES + try: + if reject: + with pytest.raises(effect_runtime.EffectRuntimeRejected, match="required read command contains NUL"): + _interaction_reads_and_arbitration(payload, contract(), runtime_root=runtime_root, + registry_path=None, agent_scope_frontier_actions=()) + else: + reads, arbitration = _interaction_reads_and_arbitration(payload, contract(), runtime_root=runtime_root, + registry_path=None, agent_scope_frontier_actions=()) + assert reads == expected + assert arbitration == SchedulerArbitration(SchedulerDisposition.ACTIVE_WORK, + "interaction_agent_attempt_required", "bounded_delivery") + finally: + assert_private_files_removed(root) + assert effect_runtime.effect_runtime_result("runtime.ping", {})["ready"] is True diff --git a/tests/control_plane/test_scheduler_backoff_convergence.py b/tests/control_plane/test_scheduler_backoff_convergence.py index b72e39c635..9f76eaf5b6 100644 --- a/tests/control_plane/test_scheduler_backoff_convergence.py +++ b/tests/control_plane/test_scheduler_backoff_convergence.py @@ -13,7 +13,7 @@ from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint GOAL_ID = "scheduler-backoff-convergence" AGENT_ID = "codex-fixture" diff --git a/tests/control_plane/test_scheduler_execution_context.py b/tests/control_plane/test_scheduler_execution_context.py index d2b27f4216..46d672fc79 100644 --- a/tests/control_plane/test_scheduler_execution_context.py +++ b/tests/control_plane/test_scheduler_execution_context.py @@ -22,7 +22,7 @@ from loopx.control_plane.scheduler.automation_liveness import ( build_automation_liveness, ) -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint from loopx.control_plane.testing.quota_fixtures import ( quota_status_payload, quota_todo_item, @@ -251,6 +251,8 @@ def test_scheduler_execution_context_decision_table( def test_partial_scheduler_context_fails_closed_without_app_action() -> None: + from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + hint = build_scheduler_hint( _active_payload(), scheduler_execution_context={"host_surface": "generic_cli"}, @@ -263,6 +265,8 @@ def test_partial_scheduler_context_fails_closed_without_app_action() -> None: def test_missing_scheduler_context_fails_closed() -> None: + from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + hint = build_scheduler_hint(_active_payload()) assert hint["action"] == "repair_scheduler_execution_context" @@ -271,6 +275,24 @@ def test_missing_scheduler_context_fails_closed() -> None: assert hint["execution_phase"]["disposition"] == "contract_error" +def test_valid_scheduler_context_requires_the_supplied_native_arbitration() -> None: + from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + + with pytest.raises(ValueError, match="scheduler arbitration must come from the interaction projection"): + build_scheduler_hint(_active_payload(), scheduler_execution_context= + scheduler_execution_context_for_runtime_profile("codex_app_heartbeat")) + + +@pytest.mark.parametrize("mode", ["goal_stopped", "quota_paused"]) +def test_explicit_shutdown_precedes_the_native_arbitration_requirement(mode: str) -> None: + from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint + + hint = build_scheduler_hint({"heartbeat_recommendation": {"recommended_mode": mode}}, + scheduler_execution_context=scheduler_execution_context_for_runtime_profile("codex_app_heartbeat")) + assert hint["cadence_class"] == mode + assert hint["app_automation"]["host_action"] == "pause_or_delete_current_heartbeat" + + def test_codex_app_runtime_profile_preserves_host_backoff_and_authority() -> None: context = scheduler_execution_context_for_runtime_profile( SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT diff --git a/tests/control_plane/test_scheduler_fallback_hint.py b/tests/control_plane/test_scheduler_fallback_hint.py index 96fac2844f..22f7d4c5c4 100644 --- a/tests/control_plane/test_scheduler_fallback_hint.py +++ b/tests/control_plane/test_scheduler_fallback_hint.py @@ -17,7 +17,7 @@ from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint from loopx.upgrade import ( load_codex_app_automation_manifest, prompt_digest, diff --git a/tests/control_plane/test_scheduler_host_failure_cache.py b/tests/control_plane/test_scheduler_host_failure_cache.py index 3f1c5ea6ca..88a07e0941 100644 --- a/tests/control_plane/test_scheduler_host_failure_cache.py +++ b/tests/control_plane/test_scheduler_host_failure_cache.py @@ -11,7 +11,7 @@ from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint from loopx.control_plane.scheduler.state import ( CODEX_APP_STATEFUL_BACKOFF_STATE_KEY, CODEX_APP_SURFACE, diff --git a/tests/control_plane/test_scheduler_interaction_arbitration.py b/tests/control_plane/test_scheduler_interaction_arbitration.py index 990a7fb8ec..0366419f48 100644 --- a/tests/control_plane/test_scheduler_interaction_arbitration.py +++ b/tests/control_plane/test_scheduler_interaction_arbitration.py @@ -8,9 +8,11 @@ from loopx.control_plane.scheduler import scheduler_hint as scheduler_hint_module from loopx.control_plane.scheduler.arbitration import ( SchedulerDisposition, - build_scheduler_arbitration, ) -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import ( + build_native_scheduler_hint as build_scheduler_hint, + project_native_arbitration, +) from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) @@ -182,7 +184,7 @@ def test_interaction_contract_drives_scheduler( disposition: SchedulerDisposition, cadence: str, ) -> None: - arbitration = build_scheduler_arbitration( + arbitration = project_native_arbitration( payload, agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS, ) @@ -547,3 +549,18 @@ def test_structurally_invalid_contract_fails_closed() -> None: "interaction_contract.agent_channel.quiet_noop_allowed_must_be_boolean" in hint["consistency_error"]["errors"] ) + + +@pytest.mark.parametrize("flag", [None, 0, 1, "false"]) +def test_python_source_codec_does_not_coerce_non_boolean_authority(flag) -> None: + payload = _payload(mode="\u0085bounded_delivery\u001c", should_run=True, + user_required=False, must_attempt=True, delivery_allowed=True, quiet_noop_allowed=False) + payload["interaction_contract"]["user_channel"]["action_required"] = flag + original = deepcopy(payload) + + arbitration = project_native_arbitration(payload) + + assert arbitration.mode == "bounded_delivery" + assert arbitration.disposition is SchedulerDisposition.CONSISTENCY_REPAIR + assert arbitration.errors == ("interaction_contract.user_channel.action_required_must_be_boolean",) + assert payload == original diff --git a/tests/control_plane/test_settled_replay_construction.py b/tests/control_plane/test_settled_replay_construction.py index 619b24c97d..63ebcfd6fc 100644 --- a/tests/control_plane/test_settled_replay_construction.py +++ b/tests/control_plane/test_settled_replay_construction.py @@ -9,7 +9,8 @@ from loopx.control_plane.testing.quota_fixtures import quota_status_payload from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile -from loopx.control_plane.scheduler.arbitration import SchedulerDisposition, build_scheduler_arbitration +from loopx.control_plane.scheduler.arbitration import SchedulerDisposition +from tests.control_plane.scheduler_arbitration_fixture import project_native_arbitration from loopx.control_plane.effect_program import interpret_quota_should_run_packet from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result from loopx.control_plane.quota.turn_envelope import build_turn_envelope @@ -46,20 +47,6 @@ def test_settled_replay_preserves_schedule_without_new_host_effects(profile, obs assert hint["goal_runtime_continuation"]["disposition"] == "continue_now" -@pytest.mark.parametrize("channel,field", [("agent_channel", "must_attempt"), - ("agent_channel", "delivery_allowed"), - ("user_channel", "action_required")]) -def test_settled_contract_cannot_carry_open_execution(channel, field): - payload = build_quota_should_run( - quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"), - goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, - ) - payload["interaction_contract"][channel][field] = True - result = build_scheduler_arbitration(payload) - assert result.disposition is SchedulerDisposition.CONSISTENCY_REPAIR - assert "interaction_contract.settled_conflicts_with_open_action" in result.errors - - @pytest.mark.parametrize("mode", ["heartbeat_settled_skip", "terminal_no_followup", "peer_coordination_blocked", "agent_monitor_only"]) def test_closed_contract_cannot_reopen_work_through_typed_consumers(mode): @@ -70,7 +57,7 @@ def test_closed_contract_cannot_reopen_work_through_typed_consumers(mode): contract = payload["interaction_contract"] contract["mode"] = mode contract["agent_channel"].update(must_attempt=True, delivery_allowed=True, quiet_noop_allowed=False) - assert build_scheduler_arbitration(payload).disposition is SchedulerDisposition.CONSISTENCY_REPAIR + assert project_native_arbitration(payload).disposition is SchedulerDisposition.CONSISTENCY_REPAIR with pytest.raises(EffectRuntimeRejected, match="conflicts with"): interpret_quota_should_run_packet(payload) with pytest.raises(EffectRuntimeRejected, match="conflicts with"): @@ -98,7 +85,7 @@ def test_closed_contract_cannot_authorize_another_spend(mode, error_code, spend_ contract["mode"] = mode contract["agent_channel"]["quiet_noop_allowed"] = mode != "heartbeat_settled_skip" contract["cli_channel"][spend_field] = True - result = build_scheduler_arbitration(payload) + result = project_native_arbitration(payload) assert result.disposition is SchedulerDisposition.CONSISTENCY_REPAIR assert error_code in result.errors with pytest.raises(EffectRuntimeRejected, match="conflicts with"): diff --git a/tests/control_plane/test_user_gate_lane_progress.py b/tests/control_plane/test_user_gate_lane_progress.py index 9cf15405c3..319c9f55ad 100644 --- a/tests/control_plane/test_user_gate_lane_progress.py +++ b/tests/control_plane/test_user_gate_lane_progress.py @@ -17,7 +17,7 @@ from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint from loopx.control_plane.scheduler.state import ( SCHEDULER_HOST_UPDATE_FAILURE_SCHEMA_VERSION, ) diff --git a/tests/control_plane_ts/interaction_contract.test.ts b/tests/control_plane_ts/interaction_contract.test.ts index 4bc5d2963e..d420488254 100644 --- a/tests/control_plane_ts/interaction_contract.test.ts +++ b/tests/control_plane_ts/interaction_contract.test.ts @@ -6,7 +6,6 @@ import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; import { interpretQuotaShouldRunPacket } from "../../loopx/control_plane/effect_program.ts"; import { decodeInteractionContract, - projectInteractionConsistencyValidation, projectInteractionRequiredReads, projectInteractionWorkContext, projectBlockedSuccessorWaitObservation, @@ -68,26 +67,13 @@ test("a closed contract can notify without requiring user action or spending", ( assert.deepEqual(decodeInteractionContract(contract), contract); }); -test("shared consistency validation retains diagnostic order and does not gate independent capability intent", () => { - const invalid = {mode: "heartbeat_settled_skip", user_required: true, - must_attempt: false, delivery_allowed: true, quiet_noop_allowed: true, - spend_allowed_now: false, spend_after_validation: true}; - assert.deepEqual(projectInteractionConsistencyValidation(invalid), {errors: [ - "interaction_contract.delivery_without_attempt", - "interaction_contract.quiet_noop_conflicts_with_required_action", - "interaction_contract.settled_conflicts_with_open_action", - "interaction_contract.settled_conflicts_with_spend", - ]}); - assert.deepEqual(projectInteractionConsistencyValidation({ - mode: "governed_capability_intent", user_required: true, - must_attempt: true, delivery_allowed: true, quiet_noop_allowed: false, - spend_allowed_now: false, spend_after_validation: false, - }), {errors: []}); - assert.throws(() => projectInteractionConsistencyValidation({...invalid, must_attempt: "false"}), /must_attempt.*boolean/); -}); - function readFacts(): JsonObject { - return {required_reads: [{command: "read-existing-evidence", source: "existing"}], + return {arbitration_facts: { + schema_version: "loopx_interaction_contract_v0", mode: "bounded_delivery", + user_channel: {present: true, action_required: false}, + agent_channel: {present: true, must_attempt: true, delivery_allowed: true, quiet_noop_allowed: false}, + cli_channel: {}, agent_scope_frontier_actions: [], + }, required_reads: [{command: "read-existing-evidence", source: "existing"}], goal_id: "goal-requirements", goal_state_file: "/tmp/goal source/state's requirements.md", command_prefix: "loopx --registry '/tmp/goal source/registry.json' --runtime-root '/tmp/goal source/runtime'", goal_acceptance_enabled: true, should_run: true, delivery_allowed: true, selection_required: false, diff --git a/tests/control_plane_ts/scheduler_arbitration.test.ts b/tests/control_plane_ts/scheduler_arbitration.test.ts new file mode 100644 index 0000000000..81b36f7e12 --- /dev/null +++ b/tests/control_plane_ts/scheduler_arbitration.test.ts @@ -0,0 +1,143 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import { + projectInteractionRequiredReads, + projectSchedulerArbitration, +} from "../../loopx/control_plane/work_items/interaction_contract.ts"; + +function facts(mode = "bounded_delivery", user = false, attempt = true, + delivery = true, quiet = false): JsonObject { + return {schema_version: "loopx_interaction_contract_v0", mode, + user_channel: {present: true, action_required: user}, + agent_channel: {present: true, must_attempt: attempt, delivery_allowed: delivery, quiet_noop_allowed: quiet}, + cli_channel: {}, agent_scope_frontier_actions: ["agent_scope_wait", "successor_replan_required", + "terminal_no_followup", "peer_coordination_blocked", "agent_monitor_only", "monitor_quiet_skip", "heartbeat_settled_skip"]}; +} + +for (const [mode, user, attempt, delivery, quiet, disposition, reason] of [ + ["terminal_no_followup", false, false, false, true, "terminal_stop", "terminal_no_followup"], + ["peer_coordination_blocked", false, false, false, true, "peer_coordination_wait", "peer_coordination_blocked"], + ["agent_monitor_only", false, false, false, true, "agent_monitor_only_wait", "agent_monitor_only"], + ["user_gate", true, false, false, false, "human_gate", "interaction_blocking_user_gate"], + ["monitor_quiet_skip", true, false, false, false, "human_gate", "interaction_blocking_user_gate"], + ["monitor_quiet_skip", false, true, true, false, "monitor_wait", "interaction_monitor_quiet_wait"], + ["heartbeat_settled_skip", false, false, false, false, "settled_replay", "interaction_heartbeat_settled_replay"], + ["successor_replan_required", false, true, false, false, "active_work", "interaction_successor_replan_required"], + ["successor_replan_required", false, false, false, true, "agent_scope_wait", "interaction_agent_scope_wait"], + ["agent_scope_wait", false, true, false, false, "agent_scope_wait", "interaction_agent_scope_wait"], + ["mapped_noop_if_unchanged", false, true, true, false, "unchanged_wait", "interaction_unchanged_wait"], + ["automation_prompt_upgrade", false, true, false, false, "active_work", "interaction_agent_attempt_required"], + ["bounded_delivery_with_user_notice", true, true, true, false, "active_work", "interaction_agent_attempt_required"], + ["quiet_skip", false, false, false, true, "quiet_wait", "interaction_quiet_noop_allowed"], + ["unknown_nonempty_mode", false, false, false, false, "quiet_wait", "interaction_delivery_not_allowed"], +] as const) { + test(`${mode}: channel ${user}/${attempt}/${delivery}/${quiet} retains ${disposition} precedence`, () => { + const input = facts(mode, user, attempt, delivery, quiet); + const before = structuredClone(input); + assert.deepEqual(projectSchedulerArbitration(input), {mode, disposition, reason_code: reason, errors: []}); + assert.deepEqual(input, before); + }); +} + +test("a settlement retry notice is non-gating only with both exact source facts", () => { + const source = facts("monitor_quiet_skip", true, false, false, false); + source.user_channel = {...source.user_channel as JsonObject, non_gating_notice: true}; + source.agent_channel = {...source.agent_channel as JsonObject, wait_source: "turn_settlement"}; + assert.deepEqual(projectSchedulerArbitration(source), {mode: "monitor_quiet_skip", errors: [], + disposition: "quiet_wait", reason_code: "blocked_wait"}); + for (const patch of [{non_gating_notice: false}, {non_gating_notice: null}]) { + assert.equal(projectSchedulerArbitration({...source, + user_channel: {...source.user_channel as JsonObject, ...patch}}).disposition, "human_gate"); + } + assert.equal(projectSchedulerArbitration({...source, + agent_channel: {...source.agent_channel as JsonObject, wait_source: "another_source"}}).disposition, "human_gate"); +}); + +test("structural diagnostics precede semantic diagnostics in the legacy stable order", () => { + const invalid = {schema_version: "future", mode: "", user_channel: {present: false}, + agent_channel: {present: false}, cli_channel: {spend_allowed_now: 1, spend_after_validation: "false"}, + agent_scope_frontier_actions: []}; + assert.deepEqual(projectSchedulerArbitration(invalid), {mode: "", disposition: "consistency_repair", + reason_code: "scheduler_interaction_contract_inconsistent", errors: [ + "interaction_contract.schema_version_mismatch", "interaction_contract.mode_missing", + "interaction_contract.user_channel_missing", "interaction_contract.agent_channel_missing", + "interaction_contract.user_channel.action_required_must_be_boolean", + "interaction_contract.agent_channel.must_attempt_must_be_boolean", + "interaction_contract.agent_channel.delivery_allowed_must_be_boolean", + "interaction_contract.agent_channel.quiet_noop_allowed_must_be_boolean", + "interaction_contract.cli_channel.spend_allowed_now_must_be_boolean", + "interaction_contract.cli_channel.spend_after_validation_must_be_boolean", + ]}); + const semantic = facts("heartbeat_settled_skip", true, false, true, true); + semantic.cli_channel = {spend_after_validation: true}; + assert.deepEqual(projectSchedulerArbitration(semantic).errors, [ + "interaction_contract.delivery_without_attempt", + "interaction_contract.quiet_noop_conflicts_with_required_action", + "interaction_contract.settled_conflicts_with_open_action", + "interaction_contract.settled_conflicts_with_spend", + ]); +}); + +for (const [channel, field] of [ + ["user_channel", "action_required"], ["agent_channel", "must_attempt"], + ["agent_channel", "delivery_allowed"], ["agent_channel", "quiet_noop_allowed"], + ["cli_channel", "spend_allowed_now"], ["cli_channel", "spend_after_validation"], +] as const) { + test(`${channel}.${field} cannot coerce a non-boolean into authority`, () => { + for (const invalid of [null, 0, 1, "false", []]) { + const source = facts(); + source[channel] = {...source[channel] as JsonObject, [field]: invalid}; + const projected = projectSchedulerArbitration(source); + assert.equal(projected.disposition, "consistency_repair"); + assert.ok((projected.errors as string[]).includes(`interaction_contract.${channel}.${field}_must_be_boolean`)); + } + }); +} + +for (const [mode, diagnostic] of [ + ["heartbeat_settled_skip", "settled_conflicts_with_spend"], + ["terminal_no_followup", "terminal_conflicts_with_open_action"], + ["peer_coordination_blocked", "terminal_conflicts_with_open_action"], + ["agent_monitor_only", "terminal_conflicts_with_open_action"], +] as const) { + test(`${mode} refuses each spend flag`, () => { + for (const field of ["spend_allowed_now", "spend_after_validation"]) { + const source = facts(mode, false, false, false, mode !== "heartbeat_settled_skip"); + source.cli_channel = {[field]: true}; + const projected = projectSchedulerArbitration(source); + assert.equal(projected.disposition, "consistency_repair"); + assert.ok((projected.errors as string[]).includes(`interaction_contract.${diagnostic}`)); + } + }); +} + +test("settled replay refuses every open-execution channel independently", () => { + for (const [channel, field] of [["user_channel", "action_required"], + ["agent_channel", "must_attempt"], ["agent_channel", "delivery_allowed"]]) { + const source = facts("heartbeat_settled_skip", false, false, false, false); + source[channel] = {...source[channel] as JsonObject, [field]: true}; + const projected = projectSchedulerArbitration(source); + assert.equal(projected.disposition, "consistency_repair"); + assert.ok((projected.errors as string[]).includes("interaction_contract.settled_conflicts_with_open_action")); + } +}); + +test("scope membership is explicit and cannot override a higher-priority disposition", () => { + const source = facts("mapped_noop_if_unchanged", false, false, false, true); + source.agent_scope_frontier_actions = ["mapped_noop_if_unchanged"]; + assert.equal(projectSchedulerArbitration(source).disposition, "agent_scope_wait"); + source.mode = "successor_replan_required"; + assert.equal(projectSchedulerArbitration(source).disposition, "quiet_wait"); + const capability = facts("governed_capability_intent", true, true, true, false); + assert.equal(projectSchedulerArbitration(capability).disposition, "active_work"); +}); + +test("the required-read reply retains its reads and complete scheduler disposition", () => { + const arbitration = facts(); + const result = projectInteractionRequiredReads({required_reads: [], should_run: false, + delivery_allowed: false, selection_required: false, has_replan: false, settlement_only: false, + goal_acceptance_enabled: false, goal_id: null, arbitration_facts: arbitration}); + assert.deepEqual(result, {required_reads: [], arbitration: {mode: "bounded_delivery", + disposition: "active_work", reason_code: "interaction_agent_attempt_required", errors: []}}); +}); diff --git a/tests/test_automation_cadence.py b/tests/test_automation_cadence.py index b1c62304de..c3ff664a7a 100644 --- a/tests/test_automation_cadence.py +++ b/tests/test_automation_cadence.py @@ -8,7 +8,7 @@ ) from loopx.cli_commands.automation_cadence import _render_automation_cadence from loopx.control_plane.testing.canary_harness import run_json_cli_result -from loopx.control_plane.scheduler.scheduler_hint import build_scheduler_hint +from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint as build_scheduler_hint from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) diff --git a/tests/test_host_loop_runtime_parity.py b/tests/test_host_loop_runtime_parity.py index dc2da2cf30..8f07238571 100644 --- a/tests/test_host_loop_runtime_parity.py +++ b/tests/test_host_loop_runtime_parity.py @@ -21,8 +21,8 @@ from loopx.control_plane.scheduler.execution_context import ( # noqa: E402 scheduler_execution_context_for_runtime_profile, ) -from loopx.control_plane.scheduler.scheduler_hint import ( # noqa: E402 - build_scheduler_hint, +from tests.control_plane.scheduler_arbitration_fixture import ( # noqa: E402 + build_native_scheduler_hint as build_scheduler_hint, ) diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index bbe996c54b..4a8ee2cfca 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -38,6 +38,7 @@ "tests/control_plane_ts/action_review_plan.test.ts", "tests/control_plane_ts/goal_workspace_digest.test.ts", "tests/control_plane_ts/interaction_contract.test.ts", + "tests/control_plane_ts/scheduler_arbitration.test.ts", "tests/control_plane_ts/runtime_decode.test.ts", "tests/control_plane_ts/quota_run_decision.test.ts", "tests/control_plane_ts/authority_store.test.ts", From c1e339fa86dd3260cf4c19a7478f53b55006dbdc Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:03:04 +0800 Subject: [PATCH 2/6] docs(architecture): record native scheduler arbitration ownership Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- .../rfcs/typescript-control-plane-migration-v0.md | 12 ++++++++++++ .../typescript-control-plane-migration-v0.zh-CN.md | 9 +++++++++ 2 files changed, 21 insertions(+) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index a08faf325e..2b8dad3281 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1409,6 +1409,18 @@ 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. Replies use the existing +private response sink; requests spool only above the inline frame bound. 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 diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 4a7cc6dfe2..9fdb9e6f13 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -962,6 +962,15 @@ 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 较早的决策。结果始终复用既有私有 response sink； +请求仅在超过内联 frame 上限时写入快照。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 与实际执行 From 54ff013319f191770063e97e81ab003ae7a9d1b5 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:11:08 +0800 Subject: [PATCH 3/6] refactor(scheduler): narrow private projection input types Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- loopx/control_plane/work_items/interaction_contract.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 8456138926..50d7a7a5cc 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1708,12 +1708,12 @@ def unadmitted_action_selection(payload: dict[str, Any]) -> bool: def _build_interaction_contract_projection( - payload: dict[str, Any], + payload: dict[str, object], *, agent_scope_frontier_actions: Collection[str], - available_capabilities: Any = None, + available_capabilities: object = None, scheduler_execution_context: ( - Mapping[str, Any] | SchedulerExecutionContextResolution | None + Mapping[str, object] | SchedulerExecutionContextResolution | None ) = None, turn_instance_id: str | None = None, runtime_root: str | None = None, From 30bbb5718d1ccf042fee5f1da8e63ae5b53e0711 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:05:04 +0800 Subject: [PATCH 4/6] perf(scheduler): keep bounded interaction facts on inline transport Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- .../work_items/interaction_contract.py | 14 ++- .../test_scheduler_arbitration_transport.py | 105 ++++++++++++++++++ 2 files changed, 117 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 50d7a7a5cc..7f94137b8f 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -2,6 +2,7 @@ from ..quota.blocked_transition_notice import blocked_priority_fallback_owner_reason from ..quota.effective_action import EffectiveAction from ..effect_runtime import effect_runtime_result +import json import shlex import typing from collections.abc import Collection, Mapping @@ -1100,7 +1101,7 @@ def _interaction_reads_and_arbitration( replan = payload.get("replan_action_packet") or {} selected = payload.get("selected_todo") or {} orchestration = payload.get("task_orchestration_contract") or {} - result = effect_runtime_result("work_item.interaction_reads.project", { + projection_params = { "arbitration_facts": _scheduler_arbitration_facts(contract, agent_scope_frontier_actions), "required_reads": [dict(item) for item in reads if isinstance(item, dict) and item.get("command")] if isinstance(reads, list) else [], @@ -1118,7 +1119,16 @@ def _interaction_reads_and_arbitration( "command_prefix": selection.render_cli_command_prefix( runtime_root=runtime_root or payload.get("runtime_root"), registry_path=registry_path or payload.get("registry")), - }, large_local_snapshot=True) + } + # With at most three generated reads and shell-quote expansion, replies fit + # 16B + 8 KiB for B ASCII JSON bytes; 64 KiB inputs stay below the 2 MiB frame cap. + try: + large_local_snapshot = len(json.dumps(projection_params, ensure_ascii=True, separators=(",", ":")).encode()) > 64 * 1024 + except (TypeError, ValueError, RecursionError, OverflowError): + large_local_snapshot = True + result = effect_runtime_result( + "work_item.interaction_reads.project", projection_params, large_local_snapshot=large_local_snapshot, + ) projected = result.get("required_reads") if not isinstance(projected, list) or any( not isinstance(item, dict) or not item.get("command") for item in projected diff --git a/tests/control_plane/test_scheduler_arbitration_transport.py b/tests/control_plane/test_scheduler_arbitration_transport.py index 73c725516b..b678cbba26 100644 --- a/tests/control_plane/test_scheduler_arbitration_transport.py +++ b/tests/control_plane/test_scheduler_arbitration_transport.py @@ -2,12 +2,14 @@ from __future__ import annotations import json +import shlex import tempfile from pathlib import Path import pytest from loopx.control_plane import effect_runtime +from loopx.control_plane.work_items import interaction_contract as interaction from loopx.control_plane.scheduler.arbitration import ( SchedulerArbitration, SchedulerDisposition, @@ -32,6 +34,109 @@ def assert_private_files_removed(root: Path) -> None: assert list(root.glob("loopx-effect-*/response.json")) == [] +@pytest.fixture +def snapshot_directories(monkeypatch): + created = [] + original = tempfile.TemporaryDirectory + + def observe(*args, **kwargs): + directory = original(*args, **kwargs) + if kwargs.get("prefix") == "loopx-effect-": + created.append(Path(directory.name)) + return directory + + monkeypatch.setattr(tempfile, "TemporaryDirectory", observe) + return created + + +@pytest.mark.parametrize("params_bytes", [65535, 65536, 65537]) +def test_compact_boundary_keeps_complete_quoted_reads_without_unneeded_snapshot( + isolated_effect_runtime, monkeypatch, snapshot_directories, params_bytes: int, +) -> None: + goal_id = "'" * 50_000 + "界😀" + state_file = "/state/'\"\\界😀" + todo_id = "todo_'\"\\界😀" + read = {"command": "read-complete-source", "source": "existing", "evidence": "", + "unknown": {"opaque": ["界😀\ud800", {"keep": True, "null": None}], + "numbers": [1e20, -1e20, 1e-6, -0.0]}} + payload = {"goal_id": goal_id, "goal_state_file": state_file, "should_run": True, + "agent_todo_summary": {"goal_acceptance_contract": {"enabled": True}}, + "selected_todo": {"todo_id": todo_id}, "required_reads": [read]} + observed = [] + original = interaction.effect_runtime_result + + def observe(method, params, **kwargs): + observed.append((encoded_size(params), kwargs["large_local_snapshot"])) + return original(method, params, **kwargs) + + monkeypatch.setattr(interaction, "effect_runtime_result", observe) + _interaction_reads_and_arbitration(payload, contract(), runtime_root=None, + registry_path=None, agent_scope_frontier_actions=()) + assert observed[-1][0] < params_bytes + assert observed[-1][1] is False + assert snapshot_directories == [] + read["evidence"] = "x" * (params_bytes - observed[-1][0]) + observed.clear() + snapshot_directories.clear() + + reads, arbitration = _interaction_reads_and_arbitration(payload, contract(), + runtime_root=None, registry_path=None, agent_scope_frontier_actions=()) + assert observed == [(params_bytes, params_bytes > 65536)] + assert bool(snapshot_directories) is (params_bytes > 65536) + assert all(not directory.exists() for directory in snapshot_directories) + assert reads == [read, { + "command": "cat -- " + shlex.quote(state_file), "source": "goal_state", "ordering": "before_work", + "reason": "Read complete Goal intent, acceptance and stops. Exact Todo reads own task state/claims. Source failure or changed requirements require a fresh guard; summaries cannot replace this read.", + }, { + "command": "loopx --format json goal-acceptance inspect --goal-id " + shlex.quote(goal_id), + "source": "goal_acceptance", "ordering": "before_work", + "reason": "Read full configured objective, non-goals and criteria with scope/revision. Scoped acceptance does not replace Goal intent or prove completion; disabled/changed criteria require a fresh guard.", + }, { + "command": "loopx --format json todo list --goal-id " + shlex.quote(goal_id) + " --todo-id " + shlex.quote(todo_id), + "source": "selected_todo", "ordering": "before_work", + "reason": "Read full current requirements and status/claim. Require one matching active Todo; missing, ambiguous or changed work requires a fresh guard. A summary cannot replace this read.", + }] + assert arbitration == SchedulerArbitration(SchedulerDisposition.ACTIVE_WORK, + "interaction_agent_attempt_required", "bounded_delivery") + assert_private_files_removed(Path(tempfile.gettempdir())) + + +@pytest.mark.parametrize(("malformed", "error_type"), [ + ("unsupported", TypeError), ("circular", ValueError), ("deep", effect_runtime.EffectRuntimeStartupError), +]) +def test_failed_size_probe_preserves_real_rpc_serialization_error( + isolated_effect_runtime, monkeypatch, snapshot_directories, malformed: str, error_type: type[Exception], +) -> None: + metadata = object() + if malformed == "circular": + metadata = [] + metadata.append(metadata) + elif malformed == "deep": + metadata = [] + # CPython 3.13's C encoder accepts 2,000 levels without a recursion error. + for _ in range(100_000): + metadata = [metadata] + modes = [] + original = interaction.effect_runtime_result + + def observe(method, params, **kwargs): + modes.append(kwargs["large_local_snapshot"]) + return original(method, params, **kwargs) + + monkeypatch.setattr(interaction, "effect_runtime_result", observe) + pid = effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] + with pytest.raises(error_type) as error: + _interaction_reads_and_arbitration({"required_reads": [{"command": "read-source", "metadata": metadata}]}, + contract(), runtime_root=None, registry_path=None, agent_scope_frontier_actions=()) + if malformed == "deep": + assert error.value.diagnostic_code == "runtime_request_failed" + assert isinstance(error.value.__cause__, RecursionError) + assert modes == [True] + assert snapshot_directories == [] + assert_private_files_removed(Path(tempfile.gettempdir())) + assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == pid + + @pytest.mark.parametrize("reject", [False, True], ids=["complete", "rejected"]) def test_near_limit_read_survives_fused_arbitration_and_cleans_private_files( isolated_effect_runtime, reject: bool, From ce2e9635d53c75354adea244642d178d717fbece Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:05:04 +0800 Subject: [PATCH 5/6] docs(architecture): explain the bounded interaction transport path Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- .../rfcs/typescript-control-plane-migration-v0.md | 10 +++++++--- .../typescript-control-plane-migration-v0.zh-CN.md | 8 ++++++-- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 2b8dad3281..a6bbe86f82 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1415,9 +1415,13 @@ 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. Replies use the existing -private response sink; requests spool only above the inline frame bound. The -2 MiB wire and 64 MiB snapshot limits stay unchanged. Python retains scalar/Mapping codecs, +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. diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 9fdb9e6f13..2bdf62148b 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -966,8 +966,12 @@ argv；同一 interaction packet 复用结果，重入投影本身在健康路 结构诊断、共享一致性检查和 disposition 选择由同一个原生 owner 负责。Quota 消费私有 `{required_reads, arbitration}` 结果，不增加公共 contract 字段。Python 选择规则与独立 consistency RPC 已删除。后续 cooldown 与 recovery 构造仍返回 dict，并丢弃未消费的 -仲裁；它们不覆盖 scheduler hint 较早的决策。结果始终复用既有私有 response sink； -请求仅在超过内联 frame 上限时写入快照。2 MiB wire 与 64 MiB snapshot 上限不变。Python 暂留 scalar/Mapping codec、 +仲裁；它们不覆盖 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 或正式安装采用已完成。 From 20233ea157da520fc115469d051f79e4f6e99dfe Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 23:42:11 +0800 Subject: [PATCH 6/6] Migrate public scheduler smokes to native arbitration Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../interaction-contract-state-machine-smoke.py | 8 ++++---- .../quota-scheduler-hint-compaction-smoke.py | 16 ++++++++++++++++ .../quota-scheduler-policy-smoke.py | 17 +++++++++++++++-- 3 files changed, 35 insertions(+), 6 deletions(-) diff --git a/examples/control_plane/interaction-contract-state-machine-smoke.py b/examples/control_plane/interaction-contract-state-machine-smoke.py index 0ff520e273..e0ade11ddc 100644 --- a/examples/control_plane/interaction-contract-state-machine-smoke.py +++ b/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 diff --git a/examples/control_plane/quota-scheduler-hint-compaction-smoke.py b/examples/control_plane/quota-scheduler-hint-compaction-smoke.py index 91b36f2b36..0b8be1c58a 100644 --- a/examples/control_plane/quota-scheduler-hint-compaction-smoke.py +++ b/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,6 +282,14 @@ 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", @@ -284,15 +297,18 @@ def assert_compact_scheduler(name: str, source_payload: dict) -> None: ): 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, diff --git a/examples/control_plane/quota-scheduler-policy-smoke.py b/examples/control_plane/quota-scheduler-policy-smoke.py index 88edfe85c6..3c460f0abd 100644 --- a/examples/control_plane/quota-scheduler-policy-smoke.py +++ b/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, )