From 862f988cca3c72557616811dccf1d6cf062a055e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:16:55 +0800 Subject: [PATCH] refactor(turn): share typed route qualification across live callers Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/architecture/rfcs/STATUS.md | 2 +- docs/architecture/rfcs/STATUS.zh-CN.md | 2 +- .../protocols/turn-controller-contract-v0.md | 24 ++- loopx/cli_commands/turn_decision.py | 29 +--- loopx/control_plane/quota/turn_envelope.py | 49 +++++- loopx/control_plane/quota/turn_envelope.ts | 97 ++++++++++- loopx/control_plane/turn_driver/driver.py | 159 +++--------------- .../turn_driver/loop_controller.py | 18 +- .../control_plane/turn_driver/managed_step.py | 7 +- .../turn_driver/turn_contract_generated.py | 17 +- .../turn_driver/turn_contract_generated.ts | 6 +- .../turn_loop_controller_contract_v0.json | 4 +- loopx/semantics/production.py | 9 +- loopx/semantics/vocabulary_v0.json | 8 +- .../test_semantic_vocabulary_drift.py | 20 +++ tests/test_loopx_turn_managed_step.py | 2 +- tests/test_turn_controller_projection.py | 128 ++++++++++++++ .../test_turn_route_action_classification.py | 82 ++++----- 18 files changed, 420 insertions(+), 243 deletions(-) create mode 100644 tests/test_turn_controller_projection.py diff --git a/docs/architecture/rfcs/STATUS.md b/docs/architecture/rfcs/STATUS.md index 76b8757020..67a340c3ce 100644 --- a/docs/architecture/rfcs/STATUS.md +++ b/docs/architecture/rfcs/STATUS.md @@ -68,7 +68,7 @@ appendix may keep dated history, but no dated log heading may precede it. | [RFC: Shared Goal Alignment and Governed Amendment Protocol (v0)](shared-goal-alignment-and-governed-amendment-v0.md) | Accepted | none | [2 entries](ledger/shared-goal-alignment-and-governed-amendment-v0/) | | [RFC: LoopX Shared Control-Plane Authority and Pluggable State Providers (v0)](shared-goal-authority-state-provider-v0.md) | Accepted | none | [24 entries](ledger/shared-goal-authority-state-provider-v0/) | | [RFC: Single-Owner Local Daemon (v0)](single-owner-local-daemon-v0.md) | Accepted | none | — | -| [RFC: TypeScript Control-Plane Migration Direction v0](typescript-control-plane-migration-v0.md) | Accepted | none | [16 entries](ledger/typescript-control-plane-migration-v0/) | +| [RFC: TypeScript Control-Plane Migration Direction v0](typescript-control-plane-migration-v0.md) | Accepted | none | [18 entries](ledger/typescript-control-plane-migration-v0/) | ## Superseded (1) diff --git a/docs/architecture/rfcs/STATUS.zh-CN.md b/docs/architecture/rfcs/STATUS.zh-CN.md index 5686b66d77..7059783a1e 100644 --- a/docs/architecture/rfcs/STATUS.zh-CN.md +++ b/docs/architecture/rfcs/STATUS.zh-CN.md @@ -65,7 +65,7 @@ | [RFC：共享 Goal 对齐与受治理 Amendment 协议（v0）](shared-goal-alignment-and-governed-amendment-v0.zh-CN.md) | 已接受 | 无 | [2 条](ledger/shared-goal-alignment-and-governed-amendment-v0/) | | [RFC：LoopX 共享控制面权威与可插拔状态 Provider（v0）](shared-goal-authority-state-provider-v0.zh-CN.md) | 已接受 | 无 | [24 条](ledger/shared-goal-authority-state-provider-v0/) | | [RFC: Single-Owner Local Daemon (v0)](single-owner-local-daemon-v0.md) | 已接受 | none | — | -| [RFC：LoopX 控制面 TypeScript 渐进迁移方向 v0](typescript-control-plane-migration-v0.zh-CN.md) | 已接受 | 无 | [16 条](ledger/typescript-control-plane-migration-v0/) | +| [RFC：LoopX 控制面 TypeScript 渐进迁移方向 v0](typescript-control-plane-migration-v0.zh-CN.md) | 已接受 | 无 | [18 条](ledger/typescript-control-plane-migration-v0/) | ## 已被替代 (1) diff --git a/docs/reference/protocols/turn-controller-contract-v0.md b/docs/reference/protocols/turn-controller-contract-v0.md index 0a58f023c6..3265a7b6b2 100644 --- a/docs/reference/protocols/turn-controller-contract-v0.md +++ b/docs/reference/protocols/turn-controller-contract-v0.md @@ -35,8 +35,10 @@ routes remain failures in the semantic verifier. ## The decision is not a result-kind string map -There is no unary result-kind-to-route projection. The driver qualifies the -fresh quota envelope into a route. The controller then evaluates: +There is no unary result-kind-to-route projection. The existing TypeScript +`quota/turn_envelope.ts::turnEnvelopeRoute` qualifies the fresh quota envelope +into a route; the Python driver and loop controller consume that same decision. +The controller then evaluates: `decide_loop_disposition(receipt, quota_decision, predecessor_turn_key, budget)`. @@ -46,12 +48,18 @@ adapter. A string-to-string result map would discard these inputs and precedence The generated contract contains 29 ordered rows: 10 admission checks and 19 returns. Conditions use finite partitions; the first matching return wins. -The existing Python controller remains the sole evaluator. TypeScript exports -shared data/types and the projection data, not a second decision implementation. - -Receipt settlement/completion qualification, budget construction, envelope -signature validation, lineage and retry normalization remain their existing -Python-owned primitives. The JSON specifies when those checks run; it is not a +The existing Python controller remains the sole evaluator of these ordered +disposition rows. TypeScript owns envelope route qualification and the shared +data/types; it does not implement another receipt/budget disposition engine. + +Receipt settlement/completion qualification, budget construction and retry +normalization retain their existing owners. Python preserves the historical +scalar/numeric codecs used by persisted Turn identities; TypeScript owns route +and envelope-signature interpretation. Plan projection returns route and +signature together, and managed-step consumes an immutable local envelope/route +result from one producer call. An extra field in a wire dictionary cannot carry +that qualification; ordinary or historical dictionaries still use the shared +route reader. The JSON specifies when those checks run; it is not a self-contained validator for arbitrary raw objects. Rejection is part of the function contract, not another enum value. No host launch, quota spend or state write is authorized by a controller output. diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 81f79c89c7..4e36ed4825 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/loopx/cli_commands/turn_decision.py @@ -32,7 +32,9 @@ ) from ..control_plane.quota.live_decision import build_live_quota_should_run_decision from ..control_plane.quota.effective_action import EffectiveAction -from ..control_plane.quota.turn_envelope import build_turn_envelope +from ..control_plane.quota.turn_envelope import ( + TurnControllerInput, build_turn_controller_input, build_turn_envelope, +) from ..control_plane.quota.settlement import SettlementIdentity, build_turn_scoped_cli_settlement_plan from ..control_plane.quota.heartbeat_receipt import ( find_heartbeat_receipt, @@ -447,32 +449,14 @@ def current_heartbeat_host_launch_decision( else current.get("effective_action"))} -def fresh_turn_envelope( - decision: Mapping[str, Any], - *, - scheduler_execution_context: Any, -) -> dict[str, Any]: - """Sign the decision into the envelope the loop controller consumes. - - The causal link to a failed predecessor is *not* written here: the envelope - schema is a signed contract, so the managed step passes its predecessor - beside the envelope instead of mutating it. - """ - - return build_turn_envelope( - decision, - scheduler_execution_context=scheduler_execution_context, - ) - - def build_fresh_envelope_for_managed_step( args: argparse.Namespace, *, registry_path: Path, runtime_root: Path, runtime_root_arg: str | None, -) -> dict[str, Any]: - """Project the current decision as the managed step's fresh envelope. +) -> TurnControllerInput: + """Project the current envelope and its route together for managed-step. Read-only: no Turn-start hooks are dispatched and no Turn instance is minted, because the managed step only decides whether the outer scheduler @@ -485,7 +469,7 @@ def build_fresh_envelope_for_managed_step( runtime_root=runtime_root, runtime_root_arg=runtime_root_arg, ) - return fresh_turn_envelope( + return build_turn_controller_input( owner.resolve(), scheduler_execution_context=owner.scheduler_execution_context, ) @@ -498,5 +482,4 @@ def build_fresh_envelope_for_managed_step( "build_fresh_envelope_for_managed_step", "build_fresh_turn_decision_owner", "collect_turn_status_payload", - "fresh_turn_envelope", ] diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 01de9f1b73..7fe75dc784 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -2,7 +2,9 @@ from __future__ import annotations +import json from collections.abc import Mapping +from dataclasses import dataclass from typing import Any from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result @@ -26,6 +28,26 @@ PLANNING_HORIZON_DETAIL_REFS_REF = "$.detail_ref" +@dataclass(frozen=True, init=False, repr=False, slots=True) +class TurnControllerInput: + """Local producer result; never an agent wire or a retained authority cache. + + The envelope and its typed route originate in one projection. Store the + envelope as immutable bytes so a caller cannot mutate it after routing; + each consumer receives its own decoded object. Ordinary wire dictionaries + cannot supply this qualification through an extra JSON field. + """ + + envelope_json: str + route: str + + def __init__(self) -> None: + raise TypeError("Use build_turn_controller_input with the original quota source") + + def envelope(self) -> dict[str, Any]: + return json.loads(self.envelope_json) + + def _turn_envelope_result( operation: str, **params: object, @@ -91,6 +113,31 @@ def build_turn_envelope( ) -> dict[str, Any]: """Project one full quota decision through the canonical TS transaction.""" + return _build_turn_projection(payload, scheduler_execution_context=scheduler_execution_context, + captured_decision_path=captured_decision_path) + + +def build_turn_controller_input( + payload: Mapping[str, Any], *, scheduler_execution_context: object = None, +) -> TurnControllerInput: + """Build the managed controller's envelope and route in one existing RPC.""" + result = _build_turn_projection(payload, + scheduler_execution_context=scheduler_execution_context, + operation="build_controller_input") + envelope, route = result.get("envelope"), result.get("route") + if not isinstance(envelope, Mapping) or not isinstance(route, str): + raise RuntimeError("TypeScript controller input must include an envelope and route") + projected = object.__new__(TurnControllerInput) + object.__setattr__(projected, "envelope_json", json.dumps(dict(envelope), ensure_ascii=False, + separators=(",", ":"))) + object.__setattr__(projected, "route", route) + return projected + + +def _build_turn_projection( + payload: Mapping[str, Any], *, scheduler_execution_context: object = None, + captured_decision_path: str | None = None, operation: str = "build", +) -> dict[str, Any]: resolution = resolve_scheduler_execution_context(scheduler_execution_context) scheduler_execution_args = "" if resolution.ok and resolution.context is not None: @@ -98,7 +145,7 @@ def build_turn_envelope( scheduler_execution_context=resolution, ) return _turn_envelope_result( - "build", + operation, payload=dict(payload), scheduler_execution_args=scheduler_execution_args, **({"captured_decision_path": captured_decision_path} diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 6441f5c62d..6d5a63d21b 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1,4 +1,5 @@ -import { EffectiveAction } from "./effective_action.generated.ts"; +import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts"; +import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; import { createHash } from "node:crypto"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; import {omitUnboundExploreResultWriteback, referenceExploreResultWriteback} from "../capabilities/explore_turn_context.ts"; @@ -1232,6 +1233,88 @@ function bindTurnDecisionCapture(request: JsonObject): JsonObject { return bound; } +// Explicit coverage of the registered decision/frontier union. New actions +// require a deliberate classification; unknown historical values retain Host routing. +export const TURN_ACTION_ROUTES = { + [EffectiveAction.AGENT_WORKSPACE_REPAIR]: "repair_required", + [EffectiveAction.BOUNDARY_PROJECTION_REPAIR]: "repair_required", + [EffectiveAction.CAPABILITY_BRIDGE_REPAIR]: "repair_required", + [EffectiveAction.CONTROL_PLANE_HEALTH_REPAIR]: "repair_required", + [EffectiveAction.CONTROL_PLANE_PROJECTION_REPAIR]: "repair_required", + [EffectiveAction.CONTROL_PLANE_REPAIR]: "repair_required", + [EffectiveAction.RUNTIME_USER_GATE_PROJECTION_REPAIR]: "repair_required", + [EffectiveAction.STATE_PROJECTION_GAP_REPAIR]: "repair_required", + [EffectiveAction.TODO_DECISION_SCOPE_PROJECTION_REPAIR]: "repair_required", + [EffectiveAction.AGENT_MONITOR_ONLY]: "ready_for_host", + [EffectiveAction.AUTOMATION_PROMPT_UPGRADE_REQUIRED]: "ready_for_host", + [EffectiveAction.BLOCKED_HEALTH]: "ready_for_host", + [EffectiveAction.BLOCKED_WAIT]: "ready_for_host", + [EffectiveAction.COORDINATE_TASK_BUNDLE]: "ready_for_host", + [EffectiveAction.EXTERNAL_EVIDENCE_OBSERVE]: "ready_for_host", + [EffectiveAction.HEARTBEAT_RECEIPT_WRITE_FAILED]: "ready_for_host", + [EffectiveAction.HEARTBEAT_SETTLED_SKIP]: "ready_for_host", + [EffectiveAction.LARK_INBOX_REPLY_DUE]: "ready_for_host", + [EffectiveAction.MONITOR_DUE]: "ready_for_host", + [EffectiveAction.MONITOR_QUIET_SKIP]: "ready_for_host", + [EffectiveAction.NORMAL_RUN]: "ready_for_host", + [EffectiveAction.OPERATOR_GATE_NOTIFY]: "ready_for_host", + [EffectiveAction.OPERATOR_INBOX_MATERIAL_REVIEW_DUE]: "ready_for_host", + [EffectiveAction.OUTCOME_FLOOR_RECOVERY]: "ready_for_host", + [EffectiveAction.PEER_COORDINATION_BLOCKED]: "ready_for_host", + [EffectiveAction.QUOTA_SKIP]: "ready_for_host", + [EffectiveAction.SCOPED_USER_GATE_FALLBACK]: "ready_for_host", + [EffectiveAction.TERMINAL_NO_FOLLOWUP]: "ready_for_host", + [EffectiveAction.THROTTLED_SKIP]: "ready_for_host", + [EffectiveAction.UNSETTLED_HOST_TURN_RECOVERY]: "ready_for_host", + [EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED]: "replan_required", + [EffectiveAction.GOVERNED_CAPABILITY_INTENT]: "capability_action_required", + agent_scope_exhausted: "ready_for_host", + agent_scope_wait: "ready_for_host", + reassignment_required: "ready_for_host", + autonomous_replan: "replan_required", + successor_replan_required: "replan_required", +} satisfies Record; + +/** Route one admitted input, independent of controller continuation/budget policy. */ +export function turnEnvelopeRoute(envelope: JsonObject, historicalText?: JsonObject): LoopXTurnRouteValue { + const signature = object(envelope.action_signature); + // The historical Python reader treats false/zero/null as an empty hash and + // renders scalar booleans with Python spelling. Valid hashes remain strings. + const sourceHash = scalarString(historicalText ? historicalText.source_hash : signature.source_hash || "", "route source hash"); + const envelopeHash = scalarString(historicalText ? historicalText.envelope_hash : signature.envelope_hash || "", "route envelope hash"); + if (signature.matches !== true || !sourceHash || sourceHash !== envelopeHash) return "contract_error"; + // Size diagnostics grant no authority and do not veto a valid contract. + const action = object(envelope.action); + if (envelope.should_run === true) { + if (action.must_attempt !== true) return "blocked"; + const name = scalarString(historicalText ? historicalText.effective_action : envelope.effective_action || "", "route effective action"); + const route = Object.hasOwn(TURN_ACTION_ROUTES, name) + ? TURN_ACTION_ROUTES[name as keyof typeof TURN_ACTION_ROUTES] : "ready_for_host"; + if (route === "replan_required") { + try { turnEnvelopeReplanSettlementIdentity(envelope); } + catch (error) { + if (!(error instanceof EffectRuntimeRequestError)) throw error; + return "contract_error"; + } + return "replan_required"; + } + if (action.delivery_allowed !== true) return "blocked"; + if (route === "capability_action_required") { + const intent = object(action.capability_intent), command = intent.command; + // Preserve empty-container falsiness from historical Python inputs. + const hasCommand = Array.isArray(command) ? command.length > 0 + : command !== null && typeof command === "object" ? Object.keys(command).length > 0 : Boolean(command); + if (intent.schema_version !== "pending_capability_intent_projection_v0" + || (intent.goal_id ?? null) !== (envelope.goal_id ?? null) + || (intent.agent_id ?? null) !== (envelope.agent_id ?? null) || !hasCommand) return "contract_error"; + return "capability_action_required"; + } + return route === "repair_required" ? "repair_required" : "ready_for_host"; + } + if (object(envelope.user).action_required === true) return "user_action_required"; + return action.quiet_noop_allowed === true ? "wait" : "blocked"; +} + export function evaluateTurnEnvelope(value: unknown): JsonObject { const request = requireJsonObject(value, "turn envelope request"); const operation = request.operation; @@ -1268,6 +1351,18 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { autonomous_replan_obligation: {...obligation, context_read: reads.review}, replan_action_packet: replanActionPacket(payload)}; } + if (operation === "driver_route") return {route: turnEnvelopeRoute(requireJsonObject(request.envelope, "Turn route envelope"), + request.historical_route_text === undefined ? undefined : requireJsonObject(request.historical_route_text, "historical route text"))}; + if (operation === "driver_lineage") { + const envelope = requireJsonObject(request.envelope, "Turn driver envelope"); + return {route: turnEnvelopeRoute(envelope, request.historical_route_text === undefined + ? undefined : requireJsonObject(request.historical_route_text, "historical route text")), + action_signature: turnEnvelopeActionSignatureDocument(envelope)}; + } + if (operation === "build_controller_input") { + const envelope = buildTurnEnvelope(request); + return {envelope, route: turnEnvelopeRoute(envelope)}; + } if (operation === "consumer_version") { assertTurnEnvelopeConsumerVersion(requireJsonObject(request.envelope, "turn_envelope")); return {ok: true}; diff --git a/loopx/control_plane/turn_driver/driver.py b/loopx/control_plane/turn_driver/driver.py index e80fbf381a..0fc5fefeb6 100644 --- a/loopx/control_plane/turn_driver/driver.py +++ b/loopx/control_plane/turn_driver/driver.py @@ -1,6 +1,5 @@ from __future__ import annotations from .turn_contract_generated import LoopXTurnRoute # compatibility re-export -from ..quota.effective_action import EffectiveAction import json from collections.abc import Mapping @@ -20,7 +19,7 @@ ) from ..quota.turn_envelope import ( assert_turn_envelope_consumer_version, - turn_envelope_action_signature_document, + _turn_envelope_result, turn_envelope_replan_settlement_identity, ) from ..scheduler.execution_context import ( @@ -53,78 +52,6 @@ def session_identity_fields(scope: str) -> tuple[str, ...]: return ("goal_id", "agent_id") -REPLAN_ACTIONS = { - "autonomous_replan", - "autonomous_replan_required", - "successor_replan_required", -} -REPAIR_ACTIONS: frozenset[EffectiveAction] = frozenset( - { - EffectiveAction.AGENT_WORKSPACE_REPAIR, - EffectiveAction.BOUNDARY_PROJECTION_REPAIR, - EffectiveAction.CAPABILITY_BRIDGE_REPAIR, - EffectiveAction.CONTROL_PLANE_HEALTH_REPAIR, - EffectiveAction.CONTROL_PLANE_PROJECTION_REPAIR, - EffectiveAction.CONTROL_PLANE_REPAIR, - EffectiveAction.RUNTIME_USER_GATE_PROJECTION_REPAIR, - EffectiveAction.STATE_PROJECTION_GAP_REPAIR, - EffectiveAction.TODO_DECISION_SCOPE_PROJECTION_REPAIR, - } -) -# Registered actions that run on a Host. The root should-run/Envelope slot is a -# disjoint union of the decision vocabulary and the frontier vocabulary, and -# every value either owner can carry is classified: replan above, repair above, -# capability intent in its own branch, and the rest named here. Host execution -# is therefore a stated classification rather than the absence of a branch, so -# a newly registered action cannot inherit it silently -- the accompanying test -# fails until the new value is classified deliberately. -HOST_EXECUTION_ACTIONS: frozenset[str] = frozenset( - { - # Decision vocabulary. - EffectiveAction.AGENT_MONITOR_ONLY.value, - EffectiveAction.AUTOMATION_PROMPT_UPGRADE_REQUIRED.value, - EffectiveAction.BLOCKED_HEALTH.value, - EffectiveAction.BLOCKED_WAIT.value, - EffectiveAction.COORDINATE_TASK_BUNDLE.value, - EffectiveAction.EXTERNAL_EVIDENCE_OBSERVE.value, - EffectiveAction.HEARTBEAT_RECEIPT_WRITE_FAILED.value, - EffectiveAction.HEARTBEAT_SETTLED_SKIP.value, - EffectiveAction.LARK_INBOX_REPLY_DUE.value, - EffectiveAction.MONITOR_DUE.value, - EffectiveAction.MONITOR_QUIET_SKIP.value, - EffectiveAction.NORMAL_RUN.value, - EffectiveAction.OPERATOR_GATE_NOTIFY.value, - EffectiveAction.OPERATOR_INBOX_MATERIAL_REVIEW_DUE.value, - EffectiveAction.OUTCOME_FLOOR_RECOVERY.value, - EffectiveAction.PEER_COORDINATION_BLOCKED.value, - EffectiveAction.QUOTA_SKIP.value, - EffectiveAction.SCOPED_USER_GATE_FALLBACK.value, - EffectiveAction.TERMINAL_NO_FOLLOWUP.value, - EffectiveAction.THROTTLED_SKIP.value, - EffectiveAction.UNSETTLED_HOST_TURN_RECOVERY.value, - # Frontier vocabulary. These are not EffectiveAction members, so they - # are named by value rather than through the enum. - "agent_scope_exhausted", - "agent_scope_wait", - "reassignment_required", - } -) - - -def unregistered_route(effective_action: str) -> LoopXTurnRoute: - """Classify an action no registered owner declares. - - Legacy payloads and forward-compatible writers can carry a value this - driver does not know. Such a value still runs on a Host: narrowing what the - driver accepts is an admission-domain change that needs its own approval and - version boundary, so this keeps the previous behaviour. Naming the path - separates "this classification was chosen" from "this value was not - recognised", which the earlier bare fallthrough could not express. - """ - - return LoopXTurnRoute.READY_FOR_HOST - - class FailedTurnSessionRecoveryError(ValueError): """Typed public-safe refusal to resume a failed Turn's Host Session.""" @@ -137,65 +64,23 @@ def _mapping(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, Mapping) else {} -def _typed_route(envelope: Mapping[str, Any]) -> LoopXTurnRoute: +def _historical_route_text(envelope: Mapping[str, Any]) -> dict[str, str]: + """Retain Python str/or encoding for historical, non-string JSON scalars. + + This codec selects no route; the TypeScript owner applies every gate. + Complete contracts do not accept these historical decoding hints. + """ signature = _mapping(envelope.get("action_signature")) - source_hash = str(signature.get("source_hash") or "") - envelope_hash = str(signature.get("envelope_hash") or "") - if ( - signature.get("matches") is not True - or not source_hash - or source_hash != envelope_hash - ): - return LoopXTurnRoute.CONTRACT_ERROR - # Packet size is a performance warning, not execution authority. Keep the - # diagnostics in the envelope; schema/signature/lineage remain hard gates. + return {"source_hash": str(signature.get("source_hash") or ""), + "envelope_hash": str(signature.get("envelope_hash") or ""), + "effective_action": str(envelope.get("effective_action") or "")} - action = _mapping(envelope.get("action")) - user = _mapping(envelope.get("user")) - should_run = envelope.get("should_run") is True - effective_action = str(envelope.get("effective_action") or "") - delivery_allowed = action.get("delivery_allowed") is True - must_attempt = action.get("must_attempt") is True - - if should_run: - if not must_attempt: - return LoopXTurnRoute.BLOCKED - # A semantic replan forbids ordinary delivery while requiring the Host - # to produce its bound planning outcome. Keep that route distinct from - # admission to execute the selected work. - if effective_action in REPLAN_ACTIONS: - try: - # Preserve the typed owner's signed Todo-less binding check. - # The plan's lineage gate still holds unbound replans; selected - # Todo replans retain their existing Task settlement identity. - turn_envelope_replan_settlement_identity(envelope, turn_instance_id=None) - except (ValueError, RuntimeError): - return LoopXTurnRoute.CONTRACT_ERROR - return LoopXTurnRoute.REPLAN_REQUIRED - if not delivery_allowed: - return LoopXTurnRoute.BLOCKED - try: - registered_action = EffectiveAction(effective_action) - except ValueError: - registered_action = None - if registered_action is EffectiveAction.GOVERNED_CAPABILITY_INTENT: - intent = _mapping(action.get("capability_intent")) - if (intent.get("schema_version") != "pending_capability_intent_projection_v0" - or intent.get("goal_id") != envelope.get("goal_id") - or intent.get("agent_id") != envelope.get("agent_id") - or not intent.get("command")): - return LoopXTurnRoute.CONTRACT_ERROR - return LoopXTurnRoute.CAPABILITY_ACTION_REQUIRED - if registered_action in REPAIR_ACTIONS: - return LoopXTurnRoute.REPAIR_REQUIRED - if effective_action in HOST_EXECUTION_ACTIONS: - return LoopXTurnRoute.READY_FOR_HOST - return unregistered_route(effective_action) - if user.get("action_required") is True: - return LoopXTurnRoute.USER_ACTION_REQUIRED - if action.get("quiet_noop_allowed") is True: - return LoopXTurnRoute.WAIT - return LoopXTurnRoute.BLOCKED + +def _typed_route(envelope: Mapping[str, Any]) -> LoopXTurnRoute: + """Compatibility transport to the shared TypeScript route owner.""" + return LoopXTurnRoute(_turn_envelope_result( + "driver_route", envelope=dict(envelope), historical_route_text=_historical_route_text(envelope), + )["route"]) def selected_turn_todo(envelope: Mapping[str, Any]) -> dict[str, Any]: @@ -236,8 +121,12 @@ def _turn_lineage( *, selected_todo: Mapping[str, Any], settlement_identity: SettlementIdentity | None = None, + projection: Mapping[str, Any] | None = None, ) -> dict[str, str]: - signature_document = turn_envelope_action_signature_document(envelope) + resolved = projection if projection is not None else _turn_envelope_result( + "driver_lineage", envelope=dict(envelope), historical_route_text=_historical_route_text(envelope), + ) + signature_document = resolved["action_signature"] action = _mapping(signature_document.get("action")) signature_document["action"] = { **action, @@ -512,8 +401,10 @@ def build_loopx_turn_plan( assert_turn_envelope_consumer_version(envelope) except (ValueError, RuntimeError) as exc: contract_error = str(exc) + projection = _turn_envelope_result("driver_lineage", envelope=envelope, + historical_route_text=_historical_route_text(envelope)) route = ( - _typed_route(envelope) + LoopXTurnRoute(projection["route"]) if execution_context.ok and contract_error is None else LoopXTurnRoute.CONTRACT_ERROR ) @@ -532,7 +423,7 @@ def build_loopx_turn_plan( and (settlement_identity is None or session_scope != "agent")): route = LoopXTurnRoute.BLOCKED lineage = _turn_lineage(envelope, selected_todo=selected_todo, - settlement_identity=settlement_identity) + settlement_identity=settlement_identity, projection=projection) session, session_error = _session_plan( route=route, lineage=lineage, diff --git a/loopx/control_plane/turn_driver/loop_controller.py b/loopx/control_plane/turn_driver/loop_controller.py index b258ad32e4..1576187c89 100644 --- a/loopx/control_plane/turn_driver/loop_controller.py +++ b/loopx/control_plane/turn_driver/loop_controller.py @@ -25,6 +25,7 @@ ) from .turn_contract_generated import LoopDisposition # compatibility re-export from ..quota.effective_action import EffectiveAction +from ..quota.turn_envelope import TurnControllerInput from collections.abc import Mapping from typing import Any, Literal @@ -96,7 +97,9 @@ def _decision_lineage(decision: Mapping[str, Any]) -> dict[str, str]: } -def _envelope_route(decision: Mapping[str, Any]) -> LoopXTurnRoute: +def _envelope_route( + decision: Mapping[str, Any], *, projected_route: LoopXTurnRoute | None = None, +) -> LoopXTurnRoute: """Return the shared typed route for a fresh quota/scheduler decision. Reuses the Turn plan driver's ``_typed_route`` contract, which requires a @@ -107,7 +110,7 @@ def _envelope_route(decision: Mapping[str, Any]) -> LoopXTurnRoute: decision. """ - route = _typed_route(decision) + route = _typed_route(decision) if projected_route is None else projected_route if route is LoopXTurnRoute.CONTRACT_ERROR: raise ValueError( "quota decision failed the shared envelope contract " @@ -441,12 +444,15 @@ class _ControllerInputs: def __init__( self, receipt: ValidatedTurnReceipt | None, - decision: Mapping[str, Any], + decision: Mapping[str, Any] | TurnControllerInput, predecessor: str | None, budget: BoundedTurnBudget | None, ) -> None: self.receipt = receipt - self.decision = decision + self.projected_route = ( + LoopXTurnRoute(decision.route) if isinstance(decision, TurnControllerInput) else None + ) + self.decision = decision.envelope() if isinstance(decision, TurnControllerInput) else decision self.predecessor = predecessor self.budget = budget self.route: LoopXTurnRoute @@ -497,7 +503,7 @@ def fact(self, name: str) -> str | bool: def check(self, name: str) -> None: receipt = self.receipt if name == "envelope": - self.route = _envelope_route(self.decision) + self.route = _envelope_route(self.decision, projected_route=self.projected_route) elif name == "decision_actor": self.decision_lineage = _decision_lineage(self.decision) if ( @@ -658,7 +664,7 @@ def render(self, rule: Mapping[str, Any]) -> dict[str, Any]: def decide_loop_disposition( *, turn_receipt: ValidatedTurnReceipt | None, - quota_decision: Mapping[str, Any], + quota_decision: Mapping[str, Any] | TurnControllerInput, predecessor_turn_key: str | None = None, bounded_turn_budget: BoundedTurnBudget | None = None, ) -> dict[str, Any]: diff --git a/loopx/control_plane/turn_driver/managed_step.py b/loopx/control_plane/turn_driver/managed_step.py index 37aa89a4b2..7a2bd1d5bb 100644 --- a/loopx/control_plane/turn_driver/managed_step.py +++ b/loopx/control_plane/turn_driver/managed_step.py @@ -26,6 +26,7 @@ from collections.abc import Mapping from typing import Any, cast +from ..quota.turn_envelope import TurnControllerInput from .host_failure import normalize_host_failure_record, project_host_failure from .loop_controller import ( LOOP_CONTROLLER_DISPOSITION_SCHEMA_VERSION, @@ -270,7 +271,7 @@ def _managed_payload( def decide_managed_step( journal: Mapping[str, Any], - fresh_decision: Mapping[str, Any], + fresh_decision: Mapping[str, Any] | TurnControllerInput, *, goal_id: str, agent_id: str, @@ -311,14 +312,14 @@ def decide_managed_step( agent_id=agent_id, turn_key=turn_key, ) - envelope = _mapping(fresh_decision) + envelope = fresh_decision.envelope() if isinstance(fresh_decision, TurnControllerInput) else _mapping(fresh_decision) if str(envelope.get("goal_id") or "") not in {"", goal_id}: raise ValueError("fresh decision goal_id does not match the managed step") if str(envelope.get("agent_id") or "") not in {"", agent_id}: raise ValueError("fresh decision agent_id does not match the managed step") disposition = decide_loop_disposition( turn_receipt=receipt, - quota_decision=envelope, + quota_decision=fresh_decision if isinstance(fresh_decision, TurnControllerInput) else envelope, predecessor_turn_key=( turn_key if predecessor_turn_key is None else predecessor_turn_key ), diff --git a/loopx/control_plane/turn_driver/turn_contract_generated.py b/loopx/control_plane/turn_driver/turn_contract_generated.py index 385daded48..66afc6c1eb 100644 --- a/loopx/control_plane/turn_driver/turn_contract_generated.py +++ b/loopx/control_plane/turn_driver/turn_contract_generated.py @@ -1,6 +1,6 @@ """Generated by scripts/generate_turn_contract.py; do not edit. Source: loopx/control_plane/turn_loop_controller_contract_v0.json -SHA256: ab92e3888d83977b2b4548d924a7ff800f7fd1f6388f347772a8c3d6211fa3d8""" +SHA256: 1d0bc416332a409159549e88ce700ceef4a4c64026154ae091469ba5b6c4e293""" from __future__ import annotations from enum import Enum @@ -49,10 +49,11 @@ class LoopDisposition(str, Enum): 'check is evaluated only if reached and its finite when partition ' 'matches.', 'result_route_relation': 'No unary result-kind to route projection exists. ' - 'driver._typed_route qualifies quota input; ' - 'decide_loop_disposition combines receipt, quota, ' - 'causal predecessor and budget. Named admission ' - 'primitives remain Python-owned.'}, + 'turn_envelope.turnEnvelopeRoute qualifies quota input ' + 'in TypeScript; decide_loop_disposition combines ' + 'receipt, quota, causal predecessor and budget. ' + 'Receipt and budget admission primitives remain ' + 'Python-owned.'}, 'partitions': {'receipt_kind': {'owner': 'LoopXTurnResultKind', 'additional': ['absent']}, 'route': {'owner': 'LoopXTurnRoute'}, 'receipt_present': [False, True], @@ -62,9 +63,9 @@ class LoopDisposition(str, Enum): 'retry_state': ['available', 'exhausted', 'not_retryable', 'not_applicable'], 'budget_state': ['absent', 'available', 'exhausted', 'not_applicable'], 'validation_recovery': ['absent', 'repair_required', 'replan_required']}, - 'checks': {'envelope': 'Shared _typed_route rejects schema/signature/capability-intent ' - 'contract_error before user projection; user action then overrides a valid ' - 'route.', + 'checks': {'envelope': 'Shared TypeScript turnEnvelopeRoute rejects ' + 'schema/signature/capability-intent contract_error before user projection; ' + 'user action then overrides a valid route.', 'decision_actor': 'Require fresh nonempty goal and agent lineage.', 'receipt_binding': 'Require predecessor_turn_key, then equality to receipt.turn_key, ' 'then matching goal and agent, in that order.', diff --git a/loopx/control_plane/turn_driver/turn_contract_generated.ts b/loopx/control_plane/turn_driver/turn_contract_generated.ts index e650744c63..abf0fa2fb4 100644 --- a/loopx/control_plane/turn_driver/turn_contract_generated.ts +++ b/loopx/control_plane/turn_driver/turn_contract_generated.ts @@ -1,6 +1,6 @@ // Generated by scripts/generate_turn_contract.py; do not edit. // Source: loopx/control_plane/turn_loop_controller_contract_v0.json -// SHA256: ab92e3888d83977b2b4548d924a7ff800f7fd1f6388f347772a8c3d6211fa3d8 +// SHA256: 1d0bc416332a409159549e88ce700ceef4a4c64026154ae091469ba5b6c4e293 export const TURN_RESULT_KINDS = [ "validated_progress", @@ -50,7 +50,7 @@ export const TURN_CONTROLLER_CONTRACT = { "receipt": "ValidatedTurnReceipt.from_execution; material settlement, durable completion and host failure qualification remain admission preconditions", "budget": "BoundedTurnBudget; strict integers, finite ranges and nonempty lineage remain admission preconditions", "rejection": "Checks raise ValueError; rejection is not a LoopDisposition. A check is evaluated only if reached and its finite when partition matches.", - "result_route_relation": "No unary result-kind to route projection exists. driver._typed_route qualifies quota input; decide_loop_disposition combines receipt, quota, causal predecessor and budget. Named admission primitives remain Python-owned." + "result_route_relation": "No unary result-kind to route projection exists. turn_envelope.turnEnvelopeRoute qualifies quota input in TypeScript; decide_loop_disposition combines receipt, quota, causal predecessor and budget. Receipt and budget admission primitives remain Python-owned." }, "partitions": { "receipt_kind": { @@ -99,7 +99,7 @@ export const TURN_CONTROLLER_CONTRACT = { ] }, "checks": { - "envelope": "Shared _typed_route rejects schema/signature/capability-intent contract_error before user projection; user action then overrides a valid route.", + "envelope": "Shared TypeScript turnEnvelopeRoute rejects schema/signature/capability-intent contract_error before user projection; user action then overrides a valid route.", "decision_actor": "Require fresh nonempty goal and agent lineage.", "receipt_binding": "Require predecessor_turn_key, then equality to receipt.turn_key, then matching goal and agent, in that order.", "initial_terminal": "Require terminal_no_followup state for the terminal action.", diff --git a/loopx/control_plane/turn_loop_controller_contract_v0.json b/loopx/control_plane/turn_loop_controller_contract_v0.json index 5f00dec3ea..6417a482da 100644 --- a/loopx/control_plane/turn_loop_controller_contract_v0.json +++ b/loopx/control_plane/turn_loop_controller_contract_v0.json @@ -5,7 +5,7 @@ "receipt": "ValidatedTurnReceipt.from_execution; material settlement, durable completion and host failure qualification remain admission preconditions", "budget": "BoundedTurnBudget; strict integers, finite ranges and nonempty lineage remain admission preconditions", "rejection": "Checks raise ValueError; rejection is not a LoopDisposition. A check is evaluated only if reached and its finite when partition matches.", - "result_route_relation": "No unary result-kind to route projection exists. driver._typed_route qualifies quota input; decide_loop_disposition combines receipt, quota, causal predecessor and budget. Named admission primitives remain Python-owned." + "result_route_relation": "No unary result-kind to route projection exists. turn_envelope.turnEnvelopeRoute qualifies quota input in TypeScript; decide_loop_disposition combines receipt, quota, causal predecessor and budget. Receipt and budget admission primitives remain Python-owned." }, "partitions": { "receipt_kind": { @@ -54,7 +54,7 @@ ] }, "checks": { - "envelope": "Shared _typed_route rejects schema/signature/capability-intent contract_error before user projection; user action then overrides a valid route.", + "envelope": "Shared TypeScript turnEnvelopeRoute rejects schema/signature/capability-intent contract_error before user projection; user action then overrides a valid route.", "decision_actor": "Require fresh nonempty goal and agent lineage.", "receipt_binding": "Require predecessor_turn_key, then equality to receipt.turn_key, then matching goal and agent, in that order.", "initial_terminal": "Require terminal_no_followup state for the terminal action.", diff --git a/loopx/semantics/production.py b/loopx/semantics/production.py index 77011e92b7..5f0a9a894a 100644 --- a/loopx/semantics/production.py +++ b/loopx/semantics/production.py @@ -112,9 +112,14 @@ def _typescript_scan( root: Path, ts_sources: list[SourceFile], field: str | None, returns: list[str], mode: str = 'production', ) -> list[Production]: - if not ts_sources or not field: + if not ts_sources or (not field and not returns): return [] - rows = run_typescript_scan(root, ts_sources, {'field': field, 'return_functions': returns, 'mode': mode}) + # A return-owned vocabulary need not also claim a global field name. + # Omit the field entirely so dynamic object keys cannot match JSON null. + request = {'return_functions': returns, 'mode': mode} + if field: + request['field'] = field + rows = run_typescript_scan(root, ts_sources, request) # The parser names the same blocker vocabulary as the Python scanner; # ``typescript_dynamic`` stays the fallback for a form it cannot classify. return [Production(r['site'], r['line'], r['form'], frozenset(r['values']), r['unresolved'], diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index c7746fa9b2..6057512317 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -264,22 +264,22 @@ "value_notes": { "ready_for_host": "should_run with delivery_allowed and must_attempt set, and an effective action that is neither a replan nor a repair class. Projects to run_now, so a Host may be engaged.", "capability_action_required": "should_run whose effective action is governed_capability_intent and whose capability-intent projection matches the envelope goal and agent and names a command. A capability action runs before any Host; a malformed intent yields contract_error instead.", - "repair_required": "should_run whose effective action is an explicit typed REPAIR_ACTIONS member. Projects to repair; unregistered strings and spelling alone grant no repair authority.", - "replan_required": "should_run whose effective action is in REPLAN_ACTIONS: autonomous_replan, autonomous_replan_required or successor_replan_required. Projects to replan.", + "repair_required": "should_run whose effective action is classified as repair_required by the typed TURN_ACTION_ROUTES owner. Projects to repair; unregistered strings and spelling alone grant no repair authority.", + "replan_required": "should_run whose effective action is classified as replan_required by TURN_ACTION_ROUTES: autonomous_replan, autonomous_replan_required or successor_replan_required. Projects to replan.", "user_action_required": "Not should_run, and the envelope user slot marks an action required. Projects to the disposition of the same name.", "wait": "Not should_run, no user action required, and the action allows a quiet no-op. Projects to wait.", "blocked": "should_run without delivery_allowed or must_attempt, or not should_run with neither a user action nor a quiet no-op. Projects to wait, so a blocked lane waits rather than failing.", "contract_error": "A wrong envelope schema version, an action signature whose source and envelope hashes do not match, or a malformed capability intent. It is the one route with no disposition: the controller rejects instead of dispositioning." }, "producers": [ + "loopx/control_plane/quota/turn_envelope.ts::turnEnvelopeRoute", "loopx/control_plane/turn_driver/driver.py::_typed_route", - "loopx/control_plane/turn_driver/driver.py::unregistered_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan", "loopx/control_plane/turn_driver/loop_controller.py::_envelope_route" ], "return_producers": [ + "loopx/control_plane/quota/turn_envelope.ts::turnEnvelopeRoute", "loopx/control_plane/turn_driver/driver.py::_typed_route", - "loopx/control_plane/turn_driver/driver.py::unregistered_route", "loopx/control_plane/turn_driver/loop_controller.py::_envelope_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan" ], diff --git a/tests/architecture/test_semantic_vocabulary_drift.py b/tests/architecture/test_semantic_vocabulary_drift.py index dcfe4e3723..a662ce960b 100644 --- a/tests/architecture/test_semantic_vocabulary_drift.py +++ b/tests/architecture/test_semantic_vocabulary_drift.py @@ -22,6 +22,26 @@ SMOKE = REPO_ROOT / "examples" / "semantic-vocabulary-drift-smoke.py" +def test_typescript_return_producers_do_not_require_a_global_field() -> None: + from loopx.semantics.inventory import SourceFile + from loopx.semantics.production import collect_production + + path = "loopx/control_plane/quota/route_probe.ts" + source = SourceFile(path, ".ts", ''' +export function route(changed: boolean) { + const unrelated = {[changed ? "x" : "y"]: "not_a_route"}; + return changed ? "run" : "wait"; +} +export function other() { return "unregistered_elsewhere"; } +export function unknown(value: string) { return value; } +''') + vocabulary = {"owners": {}, "return_producers": [path + "::route", path + "::unknown"]} + rows = collect_production(REPO_ROOT, vocabulary, [source]) + assert len(rows) == 2 + assert rows[0].values == frozenset({"run", "wait"}) and not rows[0].unresolved + assert not rows[1].values and rows[1].unresolved + + def test_semantic_vocabulary_registry_matches_the_code() -> None: completed = subprocess.run( [sys.executable, "-B", str(SMOKE)], diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py index be9925b7c5..de214470ba 100644 --- a/tests/test_loopx_turn_managed_step.py +++ b/tests/test_loopx_turn_managed_step.py @@ -475,7 +475,7 @@ def _record_envelope(decision, *, scheduler_execution_context): envelopes.append((decision, scheduler_execution_context)) return {"schema_version": "loopx_turn_envelope_v0"} - monkeypatch.setattr(turn_decision, "fresh_turn_envelope", _record_envelope) + monkeypatch.setattr(turn_decision, "build_turn_controller_input", _record_envelope) turn_decision.build_fresh_envelope_for_managed_step( args, registry_path=registry_path, diff --git a/tests/test_turn_controller_projection.py b/tests/test_turn_controller_projection.py new file mode 100644 index 0000000000..1296b1af8a --- /dev/null +++ b/tests/test_turn_controller_projection.py @@ -0,0 +1,128 @@ +"""Producer/consumer qualification and real managed CLI routing cost boundary.""" +from __future__ import annotations + +import pytest + +from loopx.control_plane.turn_driver.managed_step import decide_managed_step +from tests.test_loopx_turn_managed_step import ( + AGENT_ID, GOAL_ID, TODO_ID, _decide, _failed_journal, +) + + +@pytest.mark.parametrize("user_action_required", [False, True]) +def test_managed_projection_reuses_route_without_trusting_a_mutable_wire( + monkeypatch, user_action_required, +) -> None: + from dataclasses import FrozenInstanceError, replace + + from loopx.control_plane.quota import turn_envelope + from loopx.control_plane.turn_driver import driver, loop_controller + from tests.test_turn_route_action_classification import _run_decision + + decision = _run_decision(goal_id=GOAL_ID, agent_id=AGENT_ID, + agent_identity={"agent_id": AGENT_ID}, + selected_todo={"todo_id": TODO_ID, "text": "Advance."}) + decision["interaction_contract"]["user_channel"]["action_required"] = user_action_required + expected_envelope = turn_envelope.build_turn_envelope(decision) + calls = [] + original = turn_envelope._turn_envelope_result + + def record(operation, **params): + calls.append(operation) + return original(operation, **params) + + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", record) + monkeypatch.setattr(driver, "_turn_envelope_result", record) + current = turn_envelope.build_turn_controller_input(decision) + assert current.envelope() == expected_envelope + assert calls == ["build_controller_input"] + assert not hasattr(current, "__dict__") + with pytest.raises(FrozenInstanceError): + current.route = "ready_for_host" + with pytest.raises(TypeError): + turn_envelope.TurnControllerInput(current.envelope_json, "ready_for_host") + with pytest.raises(TypeError): + replace(current, route="ready_for_host") + with pytest.raises(TypeError, match="original quota source"): + turn_envelope.TurnControllerInput() + + # A returned object may be edited for another purpose; it cannot edit the + # producer's snapshot or attach its route qualification to that copy. + changed = current.envelope() + changed["action"]["delivery_allowed"] = False + changed["action_signature"]["matches"] = False + changed["route"] = current.route + assert current.envelope() == expected_envelope + journal = _failed_journal() + expected = _decide(journal, envelope=expected_envelope) + assert expected["disposition"] == ("user_action_required" if user_action_required else "wait") + calls.clear() + with monkeypatch.context() as patch: + patch.setattr(loop_controller, "_typed_route", lambda *_: pytest.fail("redundant route RPC")) + actual = decide_managed_step(journal, current, goal_id=GOAL_ID, + agent_id=AGENT_ID, turn_key=journal["turn_key"]) + assert actual == expected + assert calls == [] + with pytest.raises(ValueError, match="shared envelope contract"): + _decide(journal, envelope=changed) + assert calls == ["driver_route"] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_managed_cli_reuses_fresh_route_and_preserves_failed_turn( + tmp_path, monkeypatch, provider, +) -> None: + import contextlib + import io + import json + + from loopx.cli import main as cli_main + from loopx.control_plane.quota import turn_envelope + from loopx.control_plane.turn_driver import driver, turn_journal_path + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from tests.test_loopx_turn_driver import _write_live_fixture + + isolate_sqlite_runtime(tmp_path, monkeypatch) + _project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) + lineage = {"goal_id": "loopx-turn-fixture", "agent_id": "codex-fixture", "todo_id": "todo_fixture0001"} + journal = _failed_journal(lineage=lineage) + path = turn_journal_path(runtime, goal_id=lineage["goal_id"], turn_key=journal["turn_key"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(journal)) + before = path.read_bytes() + calls = [] + original = turn_envelope._turn_envelope_result + + def record(operation, **params): + calls.append(operation) + return original(operation, **params) + + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", record) + monkeypatch.setattr(driver, "_turn_envelope_result", record) + argv = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "managed-step", "--goal-id", lineage["goal_id"], "--agent-id", lineage["agent_id"], + "--turn-key", journal["turn_key"], "--host", "dsh", "--execution-mode", "isolated-headless", + "--observed-attempt", "1", "--observed-max-attempts", "3"] + for _ in range(2): + calls.clear() + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main(argv) + result = json.loads(output.getvalue()) + assert code == 0 and result["disposition"] == "wait", result + assert result["retry_continuation"]["retry_after_seconds"] == 30 + assert result["retry_continuation"]["max_attempts"] == 3 + assert calls.count("build_controller_input") == 1 + assert "driver_route" not in calls + assert path.read_bytes() == before + # The observation discrepancy must still fail without offering a retry. + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*argv[:-4], "--observed-attempt", "2", "--observed-max-attempts", "3"]) + rejected = json.loads(output.getvalue()) + assert code == 1 and rejected["ok"] is False + assert "retry_continuation" not in rejected + assert path.read_bytes() == before + index = runtime / "goals" / lineage["goal_id"] / "runs/index.jsonl" + rows = [json.loads(line) for line in index.read_text().splitlines()] if index.exists() else [] + assert not any(row.get("classification") == "quota_slot_spent" for row in rows) diff --git a/tests/test_turn_route_action_classification.py b/tests/test_turn_route_action_classification.py index ed465e3e9f..8be48f3ab7 100644 --- a/tests/test_turn_route_action_classification.py +++ b/tests/test_turn_route_action_classification.py @@ -7,8 +7,6 @@ from __future__ import annotations -import json -from pathlib import Path from typing import Any import pytest @@ -17,14 +15,10 @@ from loopx.control_plane.quota.turn_envelope import build_turn_envelope from loopx.control_plane.turn_driver.turn_contract_generated import LoopXTurnRoute from loopx.control_plane.turn_driver.driver import ( - HOST_EXECUTION_ACTIONS, - REPAIR_ACTIONS, - REPLAN_ACTIONS, _typed_route, - unregistered_route, ) -REPO_ROOT = Path(__file__).resolve().parents[1] +REPLAN_ACTIONS = {"autonomous_replan", "autonomous_replan_required", "successor_replan_required"} EXPECTED_REGISTERED_REPAIR_ACTIONS = frozenset( { @@ -85,8 +79,7 @@ def _route_for(action: str) -> str: def test_registered_repair_actions_are_explicit_and_typed() -> None: - assert REPAIR_ACTIONS == EXPECTED_REGISTERED_REPAIR_ACTIONS - for action in REPAIR_ACTIONS: + for action in EXPECTED_REGISTERED_REPAIR_ACTIONS: assert isinstance(action, EffectiveAction) assert _route_for(action.value) == "repair_required", action @@ -142,7 +135,7 @@ def test_remaining_registered_actions_reach_the_host_route() -> None: special = ( {value for value in REPLAN_ACTIONS} | {EffectiveAction.GOVERNED_CAPABILITY_INTENT.value} - | {action.value for action in REPAIR_ACTIONS} + | {action.value for action in EXPECTED_REGISTERED_REPAIR_ACTIONS} ) for member in EffectiveAction: action = member.value @@ -151,44 +144,43 @@ def test_remaining_registered_actions_reach_the_host_route() -> None: assert _route_for(action) == "ready_for_host", action -def test_every_registered_action_is_classified_deliberately() -> None: - """The registered union is partitioned, not covered by a fallthrough. - - The root should-run/Envelope slot is a disjoint union of two owners. If an - owner gains a value, this fails until the value is classified, which is the - reason to state host execution rather than reach it by subtraction. - """ - - registry = json.loads( - (REPO_ROOT / "loopx" / "semantics" / "vocabulary_v0.json").read_text( - encoding="utf-8" - ) - )["vocabularies"] - registered = set(registry["effective_action"]["values"]) | set( - registry["agent_scope_frontier_action"]["values"] - ) - - classified = ( - {action.value for action in REPAIR_ACTIONS} - | set(HOST_EXECUTION_ACTIONS) - | (REPLAN_ACTIONS & registered) - | {EffectiveAction.GOVERNED_CAPABILITY_INTENT.value} - ) - - assert registered - classified == set(), "registered actions with no stated route" - assert classified - registered - REPLAN_ACTIONS == set(), "classified value no owner declares" +def test_unregistered_actions_keep_reaching_the_host_route() -> None: + """Admission is unchanged: an unknown value still runs, it is not refused.""" + for action in ("totally_unknown_action", "future_channel_due", "skip_repair"): + assert _route_for(action) == "ready_for_host", action -def test_host_execution_classifications_hold_at_runtime() -> None: - """Naming a value is only evidence if the driver actually routes it there.""" +@pytest.mark.parametrize("command", [None, "", False, 0, [], {}]) +def test_capability_route_rejects_empty_commands(command: Any) -> None: + envelope = build_turn_envelope(_run_decision()) + envelope["effective_action"] = "governed_capability_intent" + envelope["action"]["capability_intent"] = { + "schema_version": "pending_capability_intent_projection_v0", + "goal_id": "g", "agent_id": "a", "command": command, + } + assert _typed_route(envelope) is LoopXTurnRoute.CONTRACT_ERROR - for action in sorted(HOST_EXECUTION_ACTIONS): - assert _route_for(action) == "ready_for_host", action +@pytest.mark.parametrize("effective_action", ["normal_run", "control_plane_repair", "autonomous_replan_required", "governed_capability_intent", "future_repair"]) +def test_route_refusal_order_preserves_attempt_and_user_wait_gates(effective_action: str) -> None: + envelope = build_turn_envelope(_run_decision()) + envelope["effective_action"] = effective_action + envelope["user"]["action_required"] = True + envelope["action"].update(must_attempt=False, quiet_noop_allowed=True) + # A user notice does not override an admitted attempt's refusal in this + # owner. Loop controller applies its own user precedence after validation. + assert _typed_route(envelope) is LoopXTurnRoute.BLOCKED + envelope["should_run"] = False + assert _typed_route(envelope) is LoopXTurnRoute.USER_ACTION_REQUIRED + envelope["user"]["action_required"] = False + assert _typed_route(envelope) is LoopXTurnRoute.WAIT + envelope["action_signature"]["matches"] = False + assert _typed_route(envelope) is LoopXTurnRoute.CONTRACT_ERROR -def test_unregistered_actions_keep_reaching_the_host_route() -> None: - """Admission is unchanged: an unknown value still runs, it is not refused.""" - for action in ("totally_unknown_action", "future_channel_due", "skip_repair"): - assert _route_for(action) == "ready_for_host", action - assert unregistered_route(action) is LoopXTurnRoute.READY_FOR_HOST +@pytest.mark.parametrize("value", [1.0, 1e-7, True, [1, None], {"a": False}]) +def test_historical_route_text_keeps_python_scalar_codec(value: Any) -> None: + envelope = build_turn_envelope(_run_decision()) + envelope["action_signature"].update(source_hash=value, envelope_hash=str(value)) + envelope["effective_action"] = value + assert _typed_route(envelope) is LoopXTurnRoute.READY_FOR_HOST