diff --git a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md index 276b26155a..163fb62355 100644 --- a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md +++ b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md @@ -132,6 +132,54 @@ different questions; their proximity is not proof they should be one state. benefit experiments. Replace real consumers in the delivery slice; preserve historical decoding and original effect readback across code rollback. +**Source proposal (2026-10-28).** The existing Effect runtime now composes +journal receipt qualification and ordered continuation in the typed Turn owner. +`managed_step.py` sends one complete journal/decision request instead of rebuilding +an execution payload; the driver uses the same route classifier. The registered +ordered contract, historical journal decoder and independent recovery decision +remain. Public Python receipt/budget value objects keep their API, including exact +integer transport. Native File/SQLite retry, saved-result replay and lost-spend +response tests retain the original Turn and one debit. Rule mutations exercise +the actual TS evaluator. This is a single-session source proposal, not C1 cutover, +C4.2/C4.3 adoption, installed qualification or long-horizon benefit. Reconcile +original receipts before rollback; no journal or settlement schema changes. +The same inventory measures 43 maintained Python/TS pairs on the base and 45 on +this proposal: the new controller and host-failure pairs retain the Python API +and native failure producer while sharing TS decisions. The pair budget records +that maintenance cost (43 → 45); generated exclusions and the over-budget +negative check remain. Deleted Python decisions, rather than the pair count, +establish convergence. Added bridge traffic is disclosed without a speedup claim. + +Review repair rebases this proposal onto the current shared +`turnEnvelopeRoute` owner and retains `TurnControllerInput` at the Python API. +It does not restore the former route table or Python policy partitions. +Receipt qualification rejects empty legacy lineage values before creating a +validated receipt. Historical Mapping callers retain Python scalar spelling +through the existing route text codec, including `1.0` and `1e-07`; mismatched +signatures still fail. The same public API regressions reproduce six empty +container admissions and two numeric-route refusals at the original proposal +head, and pass after repair. This corrects compatibility; it does not qualify +numeric hashes as canonical producer output or close the remaining C4 work. + +A subsequent review reproduced a transport regression with a complete 2,420,000- +character Goal: both Mapping and producer-qualified controller inputs failed +before dispatch, although the existing envelope/plan accepted them. The pure +`turn.controller` operation now reuses the envelope's bounded local snapshot +fallback after a pre-send size rejection, with explicit client/server allowlist +entries. Full facts and the original 2 MiB wire / 64 MiB snapshot bounds remain; +signature rejection, private-file/hash/length checks and ambiguous-response +non-retry behavior are retained. Real transport tests cover both input forms, +small controls, escaped Unicode and the actual snapshot limit. This repairs the +source proposal's large-input path; independent review, installed adoption and +long-horizon cost/benefit remain separate qualification. + +Integration with the current ordinary-request caller retains its explicit +`SettlementIdentity` at the public Python boundary. The typed controller reuses +the existing identity decoder to require the exact request subject and effect; +it does not invent Todo lineage or weaken ordered material-settlement checks. +Real File/SQLite original-message execution and replay qualify this source +combination, while formal App adoption and complete outcome cost remain open. + ## C4.2 — Thin outer-worker semantic adoption **Candidate for bounded policy convergence.** diff --git a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md index 7fb4d48ae5..d5ed59427d 100644 --- a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md +++ b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md @@ -104,6 +104,38 @@ execution 形状的回执，同时调用 `turn_journal.ts` 检查一致性。`re 等待完整 C1 收益实验。在交付切片中替换真实消费者；代码回退时保留历史解码与原 effect readback。 +**源码提案（2026-10-28）。** 既有 Effect runtime 在 typed Turn owner 内组合 journal +回执资格与有序继续判定。`managed_step.py` 一次传入完整 journal/decision，不再重建 +execution 载荷；driver 复用同一路由分类器。原注册有序契约、历史 journal decoder 和 +独立恢复判定保持。公开 Python 回执/预算值对象保留 API，整数通过精确编码传递。 +真实 File/SQLite 重试、已存结果重放和扣额丢响应测试保留原 Turn 与单次扣额；规则 +mutation 直接验证实际 TS evaluator。本项仅为单会话源码提案，不代表 C1 切换、 +C4.2/C4.3 采用、已安装资格或长程收益。回滚前核对原回执；journal 和结算 schema 不变。 +同一 inventory 在 base 测得 43 对独立维护的 Python/TS 模块，本提案为 45 对：新增 +controller 与 host-failure 配对保留 Python API 和原生失败 producer，同时共享 TS +决策。配对预算据此记录维护成本（43 → 45）；生成排除规则和超预算负例保持。 +收敛依据是删除了 Python 决策，不是配对计数；额外 bridge 传输如实披露，不宣称提速。 + +评审修复将本提案重基到当前共享的 `turnEnvelopeRoute` owner，并保留 Python API 的 +`TurnControllerInput`，不恢复旧路由表或 Python 策略分区。回执资格在创建 validated +receipt 前拒绝历史空身份值；历史 Mapping 调用方通过既有 route text codec 保留 +Python 标量文本，包括 `1.0` 和 `1e-07`，签名不一致仍拒绝。同一组公开 API 回归在 +原提案 head 复现六个空容器误准入和两个数字路由误拒绝，修复后通过。这是兼容修复， +不表示 canonical producer 会生成数字 hash，也不关闭其余 C4 工作。 + +后续评审用包含 2,420,000 字符完整 Goal 的输入复现传输回归：既有 envelope/plan +可以接受，Mapping 与 producer-qualified 两种 controller 输入却都在派发前失败。 +纯 `turn.controller` 操作现于预发送大小拒绝后复用 envelope 的有界本地 snapshot +回退，并显式加入客户端/服务端白名单。完整事实及原 2 MiB wire / 64 MiB snapshot +限制保持；签名拒绝、私有文件/hash/长度校验和响应不确定时不重试的行为保持。 +真实传输测试覆盖两种输入、小输入控制、转义 Unicode 和实际 snapshot 上限。 +这修复源码提案的大输入路径；独立复审、安装采用和长程完整成本/收益仍分别验收。 + +与当前普通原始请求调用方整合后，公开 Python 边界保留其显式 `SettlementIdentity`。 +typed controller 复用既有 identity decoder，要求请求主体与 effect 精确匹配， +不伪造 Todo lineage，也不弱化有序 material settlement 检查。真实 File/SQLite +原始消息执行与重放验证了该源码组合；正式 App 采用及完整成果成本仍待验收。 + ## C4.2 — 外层 worker 的薄语义采用 **有界策略收敛候选。** `scripts/external_scheduler_worker.py` 的 `_host_attempt`、 diff --git a/examples/personal-workspace-browser/research-results.mjs b/examples/personal-workspace-browser/research-results.mjs index 77d2acba02..aed72dfa79 100644 --- a/examples/personal-workspace-browser/research-results.mjs +++ b/examples/personal-workspace-browser/research-results.mjs @@ -134,10 +134,15 @@ export const researchResultsScenario = { await refresh.click(); await catalogRow.click(); await results.getByText("123456789.1234567891 SYN", {exact: true}).waitFor(); - await refresh.focus(); + // Rendering the projection precedes the reader's scheduled focus handoff. + // Let that handoff finish before moving focus to the keyboard target. + await page.waitForFunction(element => element === document.activeElement, + await results.locator("[data-result-reader]").elementHandle()); + const beforeWrongGoalRead = reads; mode = "wrong-goal"; - await page.keyboard.press("Enter"); + await refresh.press("Enter"); await results.getByText(/does not match the requested detail_ref/).waitFor(); + assert.equal(reads, beforeWrongGoalRead + 1, "Keyboard refresh verifies exactly one projection"); assert.equal(await results.getByText("123456789.1234567891 SYN", {exact: true}).count(), 0); mode = "stale"; await refresh.click(); diff --git a/examples/semantic-vocabulary-drift-smoke.py b/examples/semantic-vocabulary-drift-smoke.py index 5f24fc5f38..63fa173105 100755 --- a/examples/semantic-vocabulary-drift-smoke.py +++ b/examples/semantic-vocabulary-drift-smoke.py @@ -202,6 +202,7 @@ LITERAL_SCAN_ROOTS = ["loopx"] # Input producers with an executable witness in loopx.semantics.production. INPUT_PRODUCER_ANCHOR = { + "turn_route": "loopx/control_plane/turn_driver/driver.py::_typed_route", "effective_action": "loopx/control_plane/quota/run_decision.ts::resolveQuotaRunDecision", "turn_result_kind": "loopx/control_plane/turn_driver/transaction.py::_result_kind", "loop_disposition": "loopx/control_plane/turn_driver/loop_controller.py::decide_loop_disposition", @@ -215,7 +216,7 @@ "settlement_binding_kind", } RETURN_PRODUCER_ANCHOR = { - "turn_route": {"loopx/control_plane/turn_driver/driver.py::_typed_route", "loopx/control_plane/turn_driver/loop_controller.py::_envelope_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan"}, + "turn_route": {"loopx/control_plane/turn_driver/driver.py::_typed_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan"}, "loop_disposition": {"loopx/control_plane/turn_driver/turn_contract_generated.py::project_turn_route"}, "effective_action": {"loopx/control_plane/quota/run_decision.ts::quotaEffectiveAction", "loopx/control_plane/quota/run_decision.ts::taskOrchestrationEffectiveAction"}, "turn_result_kind": {"loopx/control_plane/turn_driver/executor.py::_task_validation_receipt"}, @@ -223,7 +224,8 @@ # Reviewed output arguments/paths replace the old any-enum-use heuristic. These # anchors retain that coverage if registry metadata is accidentally removed. CALL_PRODUCER_ANCHOR = { - "loop_disposition": {"loopx/control_plane/turn_driver/loop_controller.py::_disposition": ["disposition"]}, + # The retired Python output builder is covered by the existing executable + # complete-controller witness; routes now also have an input witness. "agent_scope_frontier_action": {"loopx/control_plane/agents/agent_scope_frontier.py::build_agent_scope_frontier_payload": ["action"]}, "turn_result_kind": { "loopx/control_plane/turn_driver/executor.py::_host_failure": ["kind"], @@ -236,7 +238,7 @@ "turn_result_kind": {"loopx/control_plane/turn_driver/executor.py::_task_validation_receipt": ["recovery_kind"]}, } TWIN_ROOT_ANCHOR = "loopx/control_plane" -TWIN_BUDGET_ANCHOR = 43 +TWIN_BUDGET_ANCHOR = 45 BUDGET_ANCHOR = { "same_runtime_forks": 1, "same_runtime_fork_definitions": 3, diff --git a/loopx/cli_commands/turn_managed_step.py b/loopx/cli_commands/turn_managed_step.py index bc85c96c17..14f55c366a 100644 --- a/loopx/cli_commands/turn_managed_step.py +++ b/loopx/cli_commands/turn_managed_step.py @@ -1,9 +1,9 @@ """CLI surface for the managed-step consumer of same-Turn continuation. Split from ``turn.py`` to respect the CLI command-owner size budget. The -subcommand resolves the canonical Turn journal, rebuilds its validated receipt, -projects the current control-plane decision as a fresh envelope, asks the pure -controller for a disposition, and prints a typed answer. It never launches a +subcommand reads the canonical Turn journal and a fresh control-plane envelope, +then sends both to the typed controller for qualification and continuation. +It prints the existing managed-step answer. It never launches a host, writes state, or spends quota. """ @@ -12,6 +12,7 @@ import argparse from collections.abc import Callable from pathlib import Path +from typing import cast from ..control_plane.runtime.status_projection_cache import ( resolve_status_projection_cache_runtime_root, @@ -46,7 +47,7 @@ def _load_journal( journal = load_turn_journal(path) if journal is None: raise ValueError("LoopX Turn journal does not exist") - return journal + return cast(dict[str, object], journal) def handle_turn_managed_step( diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index b5c78ed674..1d28952cc3 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -57,6 +57,7 @@ "quota.settlement.read", "quota.spend.commit", "quota.turn_envelope.evaluate", + "turn.controller", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", "goal.checkpoint_read_context.inspect_replay", diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 7a435cacfd..dff62be072 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -266,6 +266,7 @@ export function createEffectRuntimeHandlers( "turn_journal.inspect", lazyHandler(() => import("./turn_driver/turn_journal.ts"), ({interpretTurnJournal}) => (params) => interpretTurnJournal(turnJournalInspectionRequest(params))), ], + ["turn.controller", lazyHandler(() => import("./turn_driver/loop_controller.ts"), ({controllerRequest}) => controllerRequest)], ["turn_journal.write", lazyHandler(() => import("./turn_driver/turn_journal_effects.ts"), ({commitTurnJournal}) => commitTurnJournal)], ["turn_journal.find_settlement", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({findTurnJournalBySettlement}) => findTurnJournalBySettlement)], ["turn_journal.observed_capabilities", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({readTurnJournalCapabilities}) => readTurnJournalCapabilities)], diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 9093f27c27..a1336023e4 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -43,6 +43,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([ "quota.settlement.read", "quota.spend.commit", "quota.turn_envelope.evaluate", + "turn.controller", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", "goal.checkpoint_read_context.inspect_replay", diff --git a/loopx/control_plane/turn_driver/host_failure.ts b/loopx/control_plane/turn_driver/host_failure.ts new file mode 100644 index 0000000000..ff0d6debd2 --- /dev/null +++ b/loopx/control_plane/turn_driver/host_failure.ts @@ -0,0 +1,47 @@ +/** Qualification shared by journal recovery and continuation; no host IO. */ +import { isDeepStrictEqual } from "node:util"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { jsonObject, type JsonObject } from "../runtime_decode.ts"; + +const hostFailureKinds: ReadonlySet = new Set([ + "auth_failed", "contract_rejected", "executor_timeout", "output_budget_exhausted", + "provider_capacity", "provider_overloaded", "quota_exhausted", "rate_limited", + "session_missing", "transport_lost", "unknown", +]); +const hostRetryPolicies: Readonly> = { + executor_timeout: [2, 5], provider_capacity: [3, 30], provider_overloaded: [3, 30], + rate_limited: [3, 60], transport_lost: [3, 10], +}; + +export function normalizeHostFailure(value: unknown, integerAttempt?: string | null): JsonObject { + const fail = (message: string): never => { throw new EffectRuntimeRequestError(message); }; + const record = jsonObject(value) ?? fail("host failure record must be one object"); + if (record.schema_version !== "loopx_turn_host_failure_v0") + fail("host failure record has an unsupported schema"); + const kind = typeof record.kind === "string" ? record.kind : ""; + if (!hostFailureKinds.has(kind)) fail("unsupported built-in host failure kind"); + const attempt = record.attempt; + const exact = integerAttempt === undefined ? null + : integerAttempt !== null && /^\d+$/.test(integerAttempt) ? BigInt(integerAttempt) : 0n; + if (exact !== null ? exact < 1n || Number(exact) !== attempt + : typeof attempt !== "number" || !Number.isInteger(attempt) || attempt < 1) + fail("host failure record has an invalid attempt"); + if (typeof record.retryable !== "boolean") fail("host failure record has an invalid retryable marker"); + const policy = hostRetryPolicies[kind]; + if (record.retryable !== (policy !== undefined)) + fail("host failure record retryability does not match its kind"); + const expected: JsonObject = {schema_version: "loopx_turn_host_failure_v0", kind, + attempt: exact !== null && exact > BigInt(Number.MAX_SAFE_INTEGER) ? String(exact) : attempt, + retryable: policy !== undefined}; + if (policy) { + expected.retry = {strategy: "same_configuration", max_attempts: policy[0], + backoff_seconds: Math.min(policy[1] * 2 ** Math.max(0, Number(attempt) - 1), 300)}; + if (!isDeepStrictEqual(record.retry, expected.retry)) fail("host failure record has an invalid retry policy"); + } else if (Object.hasOwn(record, "retry")) { + fail("non-retryable host failure must not declare retry policy"); + } + if (Object.keys(record).length !== Object.keys(expected).length + || Object.keys(record).some(key => !Object.hasOwn(expected, key))) + fail("host failure record contains unsupported fields"); + return expected; +} diff --git a/loopx/control_plane/turn_driver/loop_controller.py b/loopx/control_plane/turn_driver/loop_controller.py index f78262be85..33cd1220d1 100644 --- a/loopx/control_plane/turn_driver/loop_controller.py +++ b/loopx/control_plane/turn_driver/loop_controller.py @@ -1,127 +1,69 @@ -"""Pure Turn Loop Controller transition contract. +"""Public Python value objects and transport for the typed Turn controller. -This module decides the next disposition of a governed loop from one validated -Turn receipt plus a fresh quota/scheduler decision. It is a pure function: it -never invokes a model, sleeps, mutates a host scheduler, writes state, or -spends quota. `loopx turn run-once` remains the only delivery transaction; -scheduler process management, host wake APIs, and operator presentation belong -to later adapters (see the Turn Loop Controller plan in docs/development/contributor-tasks). - -The transition output includes host delivery, capability-adapter handoff, -waiting, stopping, user action, repair, replan, and terminal closure. - -Input validity is enforced at the typed-input boundary, not encoded as an -separate error disposition. ``decide_loop_disposition`` raises ``ValueError`` when a -receipt, envelope, or budget cannot be proven against the shared Turn -contracts, so the caller is responsible for feeding only validated, fresh -inputs. +Receipt qualification and ordered continuation live in loop_controller.ts. +The exported classes retain their existing Python API; no model, scheduler or +settlement effect is executed by this boundary. """ - from __future__ import annotations -# The generated data is the shared contract; Python alone evaluates decisions. -from .turn_contract_generated import ( - TURN_CONTROLLER_CONTRACT as _LOOP_CONTROLLER_CONTRACT, - project_turn_route as _route_to_disposition, -) -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 +from typing import Any -from ..effect_program import SettlementIdentity, SettlementStepKind -from .driver import LoopXTurnRoute, _typed_route, selected_turn_todo -from .host_failure import ( - host_failure_retry_available, - normalize_host_failure_record, -) -from .transaction import ( - LOOPX_TURN_EXECUTION_SCHEMA_VERSION, - LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION, - LoopXTurnResultKind, - require_loopx_turn_completion_outcome, -) +from ..effect_program import SettlementIdentity +from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result +from ..quota.turn_envelope import TurnControllerInput +from .driver import _historical_route_text +from .transaction import LoopXTurnResultKind +from .turn_contract_generated import LoopDisposition as LoopDisposition LOOP_CONTROLLER_DISPOSITION_SCHEMA_VERSION = "loop_turn_loop_disposition_v0" BOUNDED_TURN_BUDGET_SCHEMA_VERSION = "loop_bounded_turn_budget_v0" VALIDATED_TURN_RECEIPT_SCHEMA_VERSION = "loop_validated_turn_receipt_v0" -# Material result kinds that prove a durable delivery effect and may therefore -# drive a terminal or progress transition. They require a fully committed -# transaction (all seven phases), not a validation-only intermediate. -_MATERIAL_PROGRESS_KINDS = { - LoopXTurnResultKind.VALIDATED_COMPLETION, - LoopXTurnResultKind.VALIDATED_PROGRESS, -} -_MATERIAL_RECEIPT_ORDER = ( - SettlementStepKind.VALIDATION.value, - SettlementStepKind.DURABLE_WRITEBACK.value, - SettlementStepKind.QUOTA_SPEND.value, -) - - - -def _mapping(value: Any) -> dict[str, Any]: - return dict(value) if isinstance(value, Mapping) else {} - - -def _disposition( - disposition: LoopDisposition, - *, - reason: str, - lineage: Mapping[str, str] | None = None, - extra: Mapping[str, Any] | None = None, -) -> dict[str, Any]: - payload: dict[str, Any] = { - "schema_version": LOOP_CONTROLLER_DISPOSITION_SCHEMA_VERSION, - "disposition": disposition.value, - "reason": reason, - "spends_quota": False, - "launches_host": False, - "writes_state": False, - } - if lineage: - payload["lineage"] = dict(lineage) - if extra: - payload.update(dict(extra)) - return payload - - -def _decision_lineage(decision: Mapping[str, Any]) -> dict[str, str]: - selected_todo = selected_turn_todo(decision) - return { - "goal_id": str(decision.get("goal_id") or ""), - "agent_id": str(decision.get("agent_id") or ""), - "todo_id": str(selected_todo.get("todo_id") or ""), - } - - -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 - matching action signature with non-empty equal hashes. Compaction budget - warnings are diagnostic only. A projected user action outranks delivery, so it is resolved - before the typed delivery route. Raises ``ValueError`` when the envelope - fails the shared contract instead of accepting a forged or truncated - 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 " - "(schema or signature hashes)" - ) - user = _mapping(decision.get("user")) - if user.get("action_required") is True: - return LoopXTurnRoute.USER_ACTION_REQUIRED - return route - +def _controller_result(operation: str, **facts: Any) -> dict[str, Any]: + decision = facts.get("decision") + if isinstance(decision, TurnControllerInput): + decision = decision.envelope() + if isinstance(decision, Mapping): + facts["decision"] = dict(decision) + # Preserve scalar spelling here; the shared typed owner decides routes. + facts["historical_route_text"] = _historical_route_text(decision) + # JSON numbers erase Python's int/float distinction and round large ints. + # These are exact scalar codec facts, not precomputed admission decisions. + source = facts.get({"receipt": "execution", "disposition": "receipt"}.get(operation, "journal")) + if isinstance(source, Mapping): + failure = source.get("host_failure") + scalars = {"host_attempt_count": source.get("host_attempt_count"), + "failure_attempt": failure.get("attempt") if isinstance(failure, Mapping) else None, + "observed_attempt": facts.get("observed_attempt"), + "observed_max_attempts": facts.get("observed_max_attempts")} + facts["integer_facts"] = { + key: str(value) if type(value) is int else None for key, value in scalars.items() + } + request = {"operation": operation, **facts} + try: + payload = effect_runtime_result("turn.controller", request) + except EffectRuntimeRejected as error: + if error.diagnostic_code != "request_too_large": + raise + # The pure request was refused before sending. Preserve complete facts + # through the existing bounded snapshot transport, as the envelope does. + payload = effect_runtime_result("turn.controller", request, large_local_snapshot=True) + if not isinstance(payload, dict): + raise RuntimeError("TypeScript Turn controller returned an invalid response") + if payload.get("ok") is False and isinstance(payload.get("message"), str): + if payload.get("error_type") == "TypeError": + raise TypeError(payload["message"]) + if payload.get("error_type") == "ValueError": + raise ValueError(payload["message"]) + value = payload.get("value") + if payload.get("ok") is not True or not isinstance(value, dict): + raise RuntimeError("TypeScript Turn controller returned an invalid response") + for record in (value, value.get("host_failure"), value.get("retry_continuation")): + if isinstance(record, dict) and isinstance(record.get("attempt"), str): + record["attempt"] = int(record["attempt"]) + return value class ValidatedTurnReceipt: @@ -175,85 +117,15 @@ def __init__( def from_execution( cls, execution_payload: Mapping[str, Any], *, request_identity: SettlementIdentity | None = None, ) -> ValidatedTurnReceipt: - execution = _mapping(execution_payload) - if execution.get("schema_version") != LOOPX_TURN_EXECUTION_SCHEMA_VERSION: - raise ValueError( - "validated turn receipt requires schema_version=" - f"{LOOPX_TURN_EXECUTION_SCHEMA_VERSION}" - ) - receipt = _mapping(execution.get("receipt")) - if receipt.get("schema_version") != LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION: - raise ValueError( - "validated turn receipt requires schema_version=" - f"{LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION}" - ) - if receipt.get("ok") is not True: - raise ValueError("validated turn receipt requires ok=true") - raw_kind = receipt.get("result_kind") - try: - result_kind = LoopXTurnResultKind(str(raw_kind or "")) - except ValueError: - raise ValueError( - f"validated turn receipt has unsupported result_kind {raw_kind!r}" - ) from None - if execution.get("result_kind") != result_kind.value: - raise ValueError("execution result_kind does not match its receipt") - status = str(receipt.get("status") or "") - if result_kind in _MATERIAL_PROGRESS_KINDS and ( - status != "committed" or execution.get("status") != "committed" - ): - raise ValueError( - f"material result {result_kind.value} requires a fully committed " - "Turn execution and receipt" - ) - lineage = _mapping(receipt.get("lineage")) - if request_identity is not None: - expected = {"goal_id": request_identity.goal_id, "agent_id": request_identity.agent_id, - "todo_id": "", "request_binding_id": request_identity.binding_id} - if (request_identity.request_subject is None or lineage != expected - or receipt.get("settlement_effect_id") != request_identity.effect_id): - raise ValueError("validated receipt does not match its original request identity") - elif not all(lineage.get(k) for k in ("goal_id", "agent_id", "todo_id")): - raise ValueError( - "validated turn receipt is missing goal/agent/todo lineage" - ) - turn_key = str(receipt.get("turn_key") or "") or None - if not turn_key: - raise ValueError("validated turn receipt is missing turn_key") - settlement_effect_id = str(receipt.get("settlement_effect_id") or "") or None - todo_completion = _mapping(execution.get("todo_completion")) - host_failure = None - if "host_failure" in execution: - if result_kind is not LoopXTurnResultKind.HOST_FAILURE: - raise ValueError( - "only a host_failure receipt may carry host failure metadata" - ) - host_failure = normalize_host_failure_record(execution.get("host_failure")) - if result_kind in _MATERIAL_PROGRESS_KINDS: - settlement_effect_id = _qualified_material_effect_id( - execution, - expected_effect_id=settlement_effect_id, - ) - if result_kind is LoopXTurnResultKind.VALIDATED_COMPLETION: - todo_completion = require_loopx_turn_completion_outcome( - todo_completion, - expected_todo_id=str(lineage["todo_id"]), - ) - from ..effect_runtime import effect_runtime_result - task_failure = effect_runtime_result("turn.task_validation_failure", { - "result_kind": result_kind.value, "status": execution.get("status"), - "receipt": dict(receipt), "validation": execution.get("validation"), - **({"validation_stage": execution["validation_stage"]} if "validation_stage" in execution else {}), - })["failure"] if result_kind is LoopXTurnResultKind.VALIDATION_FAILED and "validation_stage" in execution else None + value = _controller_result("receipt", execution=( + dict(execution_payload) if isinstance(execution_payload, Mapping) else {} + ), **({"request_identity": request_identity.as_dict()} if request_identity is not None else {})) return cls( - task_failure=task_failure, - result_kind=result_kind, - status=status, - lineage=lineage, - turn_key=turn_key, - host_failure=host_failure, - settlement_effect_id=settlement_effect_id, - todo_completion=todo_completion, + result_kind=LoopXTurnResultKind(value["result_kind"]), + status=value["status"], lineage=value["lineage"], turn_key=value["turn_key"], + settlement_effect_id=value["settlement_effect_id"], + host_failure=value.get("host_failure"), task_failure=value.get("task_failure"), + todo_completion=value.get("todo_completion"), ) def to_mapping(self) -> dict[str, Any]: @@ -278,49 +150,6 @@ def to_mapping(self) -> dict[str, Any]: } -def _qualified_material_effect_id( - execution: Mapping[str, Any], - *, - expected_effect_id: str | None, -) -> str: - settlement = _mapping(execution.get("settlement_result")) - if settlement.get("ok") is not True or settlement.get("failure") is not None: - raise ValueError("material Turn execution requires a successful settlement") - raw_receipts = settlement.get("receipts") - if not isinstance(raw_receipts, list): - raise TypeError("material Turn execution is missing settlement receipts") - receipts = [_mapping(item) for item in raw_receipts] - if [str(item.get("step_kind") or "") for item in receipts] != list( - _MATERIAL_RECEIPT_ORDER - ): - raise ValueError( - "material Turn execution requires ordered validation/writeback/spend receipts" - ) - effect_ids = {str(item.get("effect_id") or "") for item in receipts} - if "" in effect_ids or len(effect_ids) != 1: - raise ValueError("material Turn settlement receipts must share one effect identity") - effect_id = next(iter(effect_ids)) - if not expected_effect_id or expected_effect_id != effect_id: - raise ValueError( - "material Turn settlement identity does not match the transaction receipt" - ) - if any(item.get("status") != "committed" for item in receipts): - raise ValueError("material Turn settlement receipts must be committed") - effects = _mapping(execution.get("effects")) - committed_replay = (execution.get("replayed") is True and execution.get("dry_run") is False - and execution.get("status") == "committed" and execution.get("quota_slot_spend_count") == 1 - and effects.get("host_invoked") is False - and effects.get("state_written") is False and effects.get("quota_spent") is False) - # A replay has no new effects; the ordered committed receipts above prove - # the original durable settlement under the same transaction identity. - if not committed_replay and (effects.get("state_written") is not True or effects.get("quota_spent") is not True): - raise ValueError("material Turn execution is missing durable effect evidence") - scheduler = _mapping(execution.get("scheduler")) - if scheduler.get("completed") is not True: - raise ValueError("material Turn execution is missing completed scheduler handoff") - return effect_id - - class BoundedTurnBudget: """A provider-neutral bounded Turn budget tied to one lineage. @@ -372,302 +201,6 @@ def remaining(self) -> int: return self.max_turns - self.completed_turns -def _assert_goal_agent_match( - *, - receipt_lineage: Mapping[str, str], - decision_lineage: Mapping[str, str], -) -> None: - for key in ("goal_id", "agent_id"): - if receipt_lineage.get(key) != decision_lineage.get(key): - raise ValueError( - "stale_receipt: receipt lineage does not match the fresh decision " - f"on {key} (receipt={receipt_lineage.get(key)!r}, " - f"decision={decision_lineage.get(key)!r})" - ) - - -def _assert_same_todo( - *, - receipt_lineage: Mapping[str, str], - decision_lineage: Mapping[str, str], -) -> None: - if receipt_lineage.get("todo_id") != decision_lineage.get("todo_id"): - raise ValueError( - "stale_receipt: receipt Todo does not match the fresh decision " - f"(receipt={receipt_lineage.get('todo_id')!r}, " - f"decision={decision_lineage.get('todo_id')!r})" - ) - - -def _assert_budget_lineage_match( - *, - budget_lineage: Mapping[str, str], - decision_lineage: Mapping[str, str], -) -> None: - for key in ("goal_id", "agent_id", "todo_id"): - if budget_lineage.get(key) != decision_lineage.get(key): - raise ValueError( - "stale_budget: budget lineage does not match the fresh decision " - f"on {key} (budget={budget_lineage.get(key)!r}, " - f"decision={decision_lineage.get(key)!r})" - ) - - -def _assert_predecessor_binding( - *, - receipt: ValidatedTurnReceipt, - predecessor_turn_key: str | None, -) -> None: - """Prove the fresh envelope causally succeeds the receipt. - - The outer adapter must bind the fresh decision to the receipt's - ``turn_key``. The causal assertion stays beside the envelope instead of - adding an unsigned field to ``loopx_turn_envelope_v0``. - """ - - predecessor = str(predecessor_turn_key or "") - if not predecessor: - raise ValueError( - "stale_receipt: continuation is missing predecessor_turn_key; " - "cannot prove causal succession from the receipt" - ) - if predecessor != receipt.turn_key: - raise ValueError( - "stale_receipt: fresh decision predecessor_turn_key " - f"({predecessor!r}) does not match the receipt turn_key " - f"({receipt.turn_key!r})" - ) - - -class _ControllerInputs: - """Lazy semantic partitions and the existing ordered admission checks. - - Facts are read only when a contract row reaches them: a user gate need not - have a progress budget, and recovery routing need not inspect retry policy. - Receipt/budget constructors still own their shape and evidence validation. - No arbitrary predicates, callbacks or evaluation language are accepted. - """ - - def __init__( - self, - receipt: ValidatedTurnReceipt | None, - decision: Mapping[str, Any] | TurnControllerInput, - predecessor: str | None, - budget: BoundedTurnBudget | None, - ) -> None: - self.receipt = receipt - 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 - self.decision_lineage: dict[str, str] - self.effective_lineage: dict[str, str] - self.retry_state: Literal[ - "available", "exhausted", "not_retryable", "not_applicable" - ] = "not_applicable" - - def fact(self, name: str) -> str | bool: - if name == "receipt_present": - return self.receipt is not None - if name == "validation_recovery": - return str((self.receipt.task_failure or {}).get("recovery_kind") or "absent") if self.receipt else "absent" - if name == "receipt_kind": - return self.receipt.result_kind.value if self.receipt else "absent" - if name == "route": - return self.route.value - if name == "terminal_action": - return ( - str(self.decision.get("effective_action") or "") - == EffectiveAction.TERMINAL_NO_FOLLOWUP.value - ) - if name == "completion": - if ( - not self.receipt - or self.receipt.result_kind - is not LoopXTurnResultKind.VALIDATED_COMPLETION - ): - return "not_applicable" - return str((self.receipt.todo_completion or {}).get("continuation") or "") - if name == "budget_state": - if ( - not self.receipt - or self.receipt.result_kind - is not LoopXTurnResultKind.VALIDATED_PROGRESS - ): - return "not_applicable" - if self.budget is None: - return "absent" - return "exhausted" if self.budget.remaining <= 0 else "available" - if name == "host_failure_present": - return bool(self.receipt and self.receipt.host_failure) - if name == "retry_state": - return self.retry_state - raise ValueError(f"unsupported controller partition: {name}") - - def check(self, name: str) -> None: - receipt = self.receipt - if name == "envelope": - self.route = _envelope_route(self.decision, projected_route=self.projected_route) - elif name == "decision_actor": - self.decision_lineage = _decision_lineage(self.decision) - if ( - not self.decision_lineage["goal_id"] - or not self.decision_lineage["agent_id"] - ): - raise ValueError("fresh quota decision is missing goal/agent lineage") - self.effective_lineage = dict(self.decision_lineage) - elif name == "receipt_binding": - assert receipt is not None - _assert_predecessor_binding( - receipt=receipt, predecessor_turn_key=self.predecessor - ) - _assert_goal_agent_match( - receipt_lineage=receipt.lineage, decision_lineage=self.decision_lineage - ) - elif name == "initial_terminal": - if self.decision.get("state") != "terminal_no_followup": - raise ValueError( - "terminal no-follow-up requires fresh Goal frontier state" - ) - elif name == "initial_todo": - if not self.decision_lineage["todo_id"] and _route_to_disposition( - self.route - ) not in { - LoopDisposition.WAIT, - LoopDisposition.USER_ACTION_REQUIRED, - }: - raise ValueError( - "executable quota decision is missing selected Todo lineage" - ) - elif name == "completion_terminal": - if ( - self.decision.get("effective_action") - != EffectiveAction.TERMINAL_NO_FOLLOWUP.value - or self.decision.get("state") != "terminal_no_followup" - ): - raise ValueError( - "no_followup completion requires fresh terminal Goal frontier evidence" - ) - elif name == "completion_todo": - assert receipt is not None - selected = self.decision_lineage["todo_id"] - if not selected: - raise ValueError( - "Todo completion continuation requires a fresh selected Todo" - ) - completion = receipt.todo_completion or {} - if self.fact("completion") == "successor": - successors = completion.get("successor_todo_ids") - if not isinstance(successors, list) or selected not in successors: - raise ValueError( - "stale_receipt: fresh decision is not a declared completion successor" - ) - elif ( - self.fact("completion") == "active_goal" - and selected == receipt.lineage["todo_id"] - ): - raise ValueError( - "stale_receipt: active Goal continuation reselected the completed Todo" - ) - elif name == "receipt_todo": - assert receipt is not None - if self.decision_lineage["todo_id"]: - _assert_same_todo( - receipt_lineage=receipt.lineage, - decision_lineage=self.decision_lineage, - ) - elif ( - receipt.result_kind is LoopXTurnResultKind.VALIDATED_PROGRESS - and self.route - not in { - LoopXTurnRoute.USER_ACTION_REQUIRED, - LoopXTurnRoute.WAIT, - } - ): - raise ValueError( - "receipt-backed executable decision is missing selected Todo lineage" - ) - else: - self.effective_lineage["todo_id"] = receipt.lineage["todo_id"] - elif name == "progress_budget": - if self.budget is None: - raise ValueError( - "validated progress cannot continue without a proven bounded turn budget" - ) - _assert_budget_lineage_match( - budget_lineage=self.budget.lineage, - decision_lineage=self.effective_lineage, - ) - elif name == "host_retry": - assert receipt is not None and receipt.host_failure - if host_failure_retry_available(receipt.host_failure): - self.retry_state = "available" - elif receipt.host_failure.get("retryable") is True: - self.retry_state = "exhausted" - else: - self.retry_state = "not_retryable" - else: - raise ValueError(f"unsupported controller check: {name}") - - def render(self, rule: Mapping[str, Any]) -> dict[str, Any]: - disposition = ( - _route_to_disposition(self.route) - if rule["disposition"] == "project_route" - else LoopDisposition(rule["disposition"]) - ) - reason = rule["reason"] - if isinstance(reason, Mapping): - reason = reason.get(disposition.value, reason.get("default")) - if not isinstance(reason, str): - raise ValueError( - f"controller rule {rule['id']} has no reason for {disposition.value}" - ) - failure = (self.receipt.host_failure or {}) if self.receipt else {} - reason = reason.format( - failure_kind=failure.get("kind"), receipt_kind=self.fact("receipt_kind") - ) - lineage = { - "decision": self.decision_lineage, - "effective": self.effective_lineage, - "receipt": self.receipt.lineage if self.receipt else {}, - }[rule["lineage"]] - if disposition is LoopDisposition.REPLAN: - return _replan_disposition(reason=reason, decision_lineage=lineage) - extra = None - if rule.get("extra") == "capability": - extra = { - "capability_action": _mapping( - _mapping(self.decision.get("action")).get("capability_intent") - ) - } - elif rule.get("extra") == "iteration_stop": - extra = { - "stop_scope": "iteration", - "goal_terminal": False, - "continuation_required": False, - } - elif rule.get("extra") == "retry": - retry = _mapping(failure.get("retry")) - extra = { - "retry_continuation": { - "same_turn": True, - "retry_failed_turn": True, - "strategy": retry["strategy"], - "retry_after_seconds": retry["backoff_seconds"], - "attempt": failure["attempt"], - "max_attempts": retry["max_attempts"], - "fresh_envelope_required": True, - "model_fallback_allowed": False, - } - } - elif rule.get("extra") is not None: - raise ValueError(f"unsupported controller continuation: {rule['extra']}") - return _disposition(disposition, reason=reason, lineage=lineage, extra=extra) - - def decide_loop_disposition( *, turn_receipt: ValidatedTurnReceipt | None, @@ -675,46 +208,13 @@ def decide_loop_disposition( predecessor_turn_key: str | None = None, bounded_turn_budget: BoundedTurnBudget | None = None, ) -> dict[str, Any]: - """Evaluate the bundled ordered controller contract without side effects. - - Receipt and budget admission remain :class:`ValidatedTurnReceipt` and - :class:`BoundedTurnBudget`. The separate predecessor key proves causal - succession without adding an unsigned field to the quota envelope. - - Each matching check must pass before evaluation proceeds; the first - matching return wins. The finite partitions and their precedence are in - ``turn_loop_controller_contract_v0.json``, including checks skipped by - earlier returns. Invalid or stale inputs raise, never become dispositions. - """ - - inputs = _ControllerInputs( - turn_receipt, quota_decision, predecessor_turn_key, bounded_turn_budget - ) - for rule in _LOOP_CONTROLLER_CONTRACT["rules"]: - if not all( - inputs.fact(name) in values for name, values in rule["when"].items() - ): - continue - if "check" in rule: - inputs.check(rule["check"]) - else: - return inputs.render(rule) - raise ValueError("controller contract has no disposition for the qualified inputs") - - -def _replan_disposition( - *, reason: str, decision_lineage: Mapping[str, str] -) -> dict[str, Any]: - return _disposition( - LoopDisposition.REPLAN, - reason=reason, - lineage=decision_lineage, - extra={ - "replan_continuation": { - "requires_bounded_delta": True, - "delta_kinds": ["todo_delta", "vision_delta"], - "stale_todo_rerun_allowed": False, - "fresh_envelope_required": True, - } - }, + """Transport complete facts; the TS owner preserves ordered early returns.""" + budget = bounded_turn_budget.to_mapping() if bounded_turn_budget is not None else None + if budget is not None: + budget = {**budget, "max_turns": str(budget["max_turns"]), + "completed_turns": str(budget["completed_turns"])} + return _controller_result( + "disposition", receipt=turn_receipt.to_mapping() if turn_receipt else None, + decision=quota_decision, predecessor_turn_key=predecessor_turn_key, + budget=budget, ) diff --git a/loopx/control_plane/turn_driver/loop_controller.ts b/loopx/control_plane/turn_driver/loop_controller.ts new file mode 100644 index 0000000000..0bbe90d0f1 --- /dev/null +++ b/loopx/control_plane/turn_driver/loop_controller.ts @@ -0,0 +1,377 @@ +/** Receipt qualification and ordered next-work decisions. Journal recovery is + * composed, not replaced: neither decision dispatches a host or settles a Turn. */ +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { settlementIdentityFromPlan } from "../effect_program.ts"; +import { jsonObject, type JsonObject } from "../runtime_decode.ts"; +import { turnEnvelopeRoute as typedTurnRoute } from "../quota/turn_envelope.ts"; +import { EffectiveAction } from "../quota/effective_action.generated.ts"; +import { stripPythonWhitespace } from "../coordination/todo_agents.ts"; +import { pythonFieldText } from "../todos/read_codec.ts"; +import { interpretTurnJournal } from "./turn_journal.ts"; +import { normalizeHostFailure } from "./host_failure.ts"; +import { taskValidationFailure } from "./task_validation_failure.ts"; +import { TURN_CONTROLLER_CONTRACT, TURN_RESULT_KINDS, + type LoopXTurnRouteValue, type LoopDispositionValue } from "./turn_contract_generated.ts"; + +const object = (value: unknown): JsonObject => jsonObject(value) ?? {}; +const text = (value: unknown): string => pythonFieldText({value}, "value"); +const repr = (value: unknown): string => text([value]).slice(1, -1); +const has = (value: JsonObject, key: string): boolean => Object.hasOwn(value, key); +class ControllerTypeError extends EffectRuntimeRequestError {} +function requireThat(condition: unknown, message: string): asserts condition { + if (!condition) throw new EffectRuntimeRequestError(message); +} +type Lineage = Record<"goal_id" | "agent_id" | "todo_id", string>; +function lineage(value: unknown): Lineage { + const raw = object(value); + return {goal_id: text(raw.goal_id), agent_id: text(raw.agent_id), todo_id: text(raw.todo_id)}; +} + +// Retain the public TS name while sharing the current driver/envelope owner. +export { typedTurnRoute }; + +function decisionLineage(decision: JsonObject): Lineage { + const orchestration = object(decision.task_orchestration_contract); + const primary = stripPythonWhitespace(text(orchestration.primary_todo_id)); + const todo = orchestration.schema_version === "task_orchestration_contract_v2" + && orchestration.mode === "adaptive" && primary + ? primary : text(object(object(decision.action).selected_todo).todo_id); + return {...lineage(decision), todo_id: todo}; +} + +function materialEffect(execution: JsonObject, expectedEffect: string | null): string { + const settlement = object(execution.settlement_result); + requireThat(settlement.ok === true && settlement.failure == null, + "material Turn execution requires a successful settlement"); + if (!Array.isArray(settlement.receipts)) + throw new ControllerTypeError("material Turn execution is missing settlement receipts"); + const receipts = settlement.receipts.map(object); + requireThat(JSON.stringify(receipts.map(item => text(item.step_kind))) + === JSON.stringify(["validation", "durable_writeback", "quota_spend"]), + "material Turn execution requires ordered validation/writeback/spend receipts"); + const ids = new Set(receipts.map(item => text(item.effect_id))); + requireThat(!ids.has("") && ids.size === 1, + "material Turn settlement receipts must share one effect identity"); + const effect = [...ids][0]; + requireThat(expectedEffect && expectedEffect === effect, + "material Turn settlement identity does not match the transaction receipt"); + requireThat(receipts.every(item => item.status === "committed"), + "material Turn settlement receipts must be committed"); + const effects = object(execution.effects); + const replay = execution.replayed === true && execution.dry_run === false + && execution.status === "committed" + // Retain the historical Python numeric equality for this v0 replay marker. + && (execution.quota_slot_spend_count === 1 || execution.quota_slot_spend_count === true) + && effects.host_invoked === false && effects.state_written === false && effects.quota_spent === false; + requireThat(replay || effects.state_written === true && effects.quota_spent === true, + "material Turn execution is missing durable effect evidence"); + requireThat(object(execution.scheduler).completed === true, + "material Turn execution is missing completed scheduler handoff"); + return effect; +} + +function completionOutcome(value: unknown, expectedTodo: string): JsonObject { + if (!jsonObject(value)) throw new ControllerTypeError("validated completion is missing its durable Todo outcome"); + const completion = object(value); + const todo = text(completion.todo_id); + requireThat(expectedTodo && todo === expectedTodo, "validated completion is missing its durable Todo outcome"); + const continuation = text(completion.continuation); + requireThat(["successor", "active_goal", "no_followup"].includes(continuation), + "validated completion has an invalid continuation outcome"); + const outcome: JsonObject = {todo_id: todo, continuation}; + if (continuation === "successor") { + const successors = completion.successor_todo_ids; + requireThat(Array.isArray(successors) && successors.length + && successors.every(item => typeof item === "string" && item), + "successor completion requires successor Todo ids"); + outcome.successor_todo_ids = [...successors]; + } + return outcome; +} + +export function qualifyExecution(value: unknown, integers: JsonObject = {}, requestIdentity?: unknown): JsonObject { + const execution = object(value); + requireThat(execution.schema_version === "loopx_turn_execution_v0", + "validated turn receipt requires schema_version=loopx_turn_execution_v0"); + const receipt = object(execution.receipt); + requireThat(receipt.schema_version === "loopx_turn_receipt_validation_v0", + "validated turn receipt requires schema_version=loopx_turn_receipt_validation_v0"); + requireThat(receipt.ok === true, "validated turn receipt requires ok=true"); + const kind = text(receipt.result_kind); + requireThat((TURN_RESULT_KINDS as readonly string[]).includes(kind), + `validated turn receipt has unsupported result_kind ${repr(receipt.result_kind)}`); + requireThat(execution.result_kind === kind, "execution result_kind does not match its receipt"); + const material = ["validated_completion", "validated_progress"].includes(kind); + requireThat(!material || receipt.status === "committed" && execution.status === "committed", + `material result ${kind} requires a fully committed Turn execution and receipt`); + const identity = object(receipt.lineage); + if (requestIdentity != null) { + const decoded = settlementIdentityFromPlan({settlement_plan: {identity: requestIdentity}}); + const expected = decoded.value; + requireThat(expected?.binding_kind === "request" && Object.keys(identity).length === 4 + && identity.goal_id === expected.goal_id && identity.agent_id === expected.agent_id + && identity.todo_id === "" && identity.request_binding_id === expected.binding_id + && receipt.settlement_effect_id === expected.effect_id, + "validated receipt does not match its original request identity"); + } else { + requireThat(text(identity.goal_id) && text(identity.agent_id) && text(identity.todo_id), + "validated turn receipt is missing goal/agent/todo lineage"); + } + const turnKey = text(receipt.turn_key); + requireThat(turnKey, "validated turn receipt is missing turn_key"); + let effect = text(receipt.settlement_effect_id) || null; + let failure: JsonObject | null = null; + if (has(execution, "host_failure")) { + requireThat(kind === "host_failure", "only a host_failure receipt may carry host failure metadata"); + failure = normalizeHostFailure(execution.host_failure, integers.failure_attempt as string | null | undefined); + } + if (material) effect = materialEffect(execution, effect); + const completion = kind === "validated_completion" + ? completionOutcome(object(execution.todo_completion), text(identity.todo_id)) : object(execution.todo_completion); + const taskFailure = kind === "validation_failed" && has(execution, "validation_stage") + ? taskValidationFailure({result_kind: kind, status: execution.status, receipt, + validation: execution.validation, validation_stage: execution.validation_stage}).failure : null; + return {schema_version: "loop_validated_turn_receipt_v0", result_kind: kind, status: text(receipt.status), + lineage: {...lineage(identity), ...(text(identity.request_binding_id) + ? {request_binding_id: identity.request_binding_id} : {})}, turn_key: turnKey, settlement_effect_id: effect, + ...(failure ? {host_failure: failure} : {}), ...(taskFailure ? {task_failure: taskFailure} : {}), + ...(Object.keys(completion).length ? {todo_completion: completion} : {})}; +} + +interface Rule {id: string; when: Record; + check?: string; disposition?: string; reason?: string | Record; lineage?: string; extra?: string} +type Budget = {lineage: Lineage; max_turns: bigint; completed_turns: bigint}; +function budget(value: unknown): Budget | null { + if (value == null) return null; + const raw = object(value); + // Python's public value object admits arbitrary integers. Decimal transport + // preserves their exact comparison instead of rounding at the JSON bridge. + const integer = (v: unknown): bigint => { + requireThat(typeof v === "string" && /^-?\d+$/.test(v), "invalid bounded turn budget transport"); + return BigInt(v); + }; + const maximum = integer(raw.max_turns); + const completed = integer(raw.completed_turns); + requireThat(maximum > 0n, "bounded turn budget max_turns must be > 0"); + requireThat(completed >= 0n, "bounded turn budget completed_turns must be >= 0"); + requireThat(completed <= maximum, "bounded turn budget completed_turns must be <= max_turns"); + const identity = lineage(raw.lineage); + requireThat(identity.goal_id && identity.agent_id && identity.todo_id, + "bounded turn budget is missing goal/agent/todo lineage"); + return {lineage: identity, max_turns: maximum, completed_turns: completed}; +} + +/** Lazy rows retain validation precedence. In particular capability/user/terminal + * returns never require a later progress budget or host retry policy. */ +export function decideLoopDisposition(params: JsonObject): JsonObject { + const receipt = params.receipt == null ? null : object(params.receipt); + const decision = object(params.decision); + let route: LoopXTurnRouteValue = "contract_error"; + let chosen: Lineage = lineage({}); + let effective = {...chosen}; + let retryState = "not_applicable"; + let bounded: Budget | null = null; + const kind = receipt ? text(receipt.result_kind) : "absent"; + const receiptLineage = lineage(receipt?.lineage); + let failure = object(receipt?.host_failure); + const completion = object(receipt?.todo_completion); + const projectRoute = (): LoopDispositionValue => { + const projected = TURN_CONTROLLER_CONTRACT.route_projection[route]; + requireThat(projected !== null, "quota decision failed the shared envelope contract (schema or signature hashes)"); + return projected; + }; + const fact = (name: string): string | boolean => { + switch (name) { + case "receipt_present": return receipt !== null; + case "receipt_kind": return kind; + case "route": return route; + case "validation_recovery": return text(object(receipt?.task_failure).recovery_kind) || "absent"; + case "terminal_action": return decision.effective_action === EffectiveAction.TERMINAL_NO_FOLLOWUP; + case "completion": return kind === "validated_completion" ? text(completion.continuation) : "not_applicable"; + case "host_failure_present": return Object.keys(failure).length > 0; + case "retry_state": return retryState; + case "budget_state": return kind !== "validated_progress" ? "not_applicable" + : bounded === null ? "absent" : bounded.completed_turns >= bounded.max_turns ? "exhausted" : "available"; + default: throw new EffectRuntimeRequestError(`unsupported controller partition: ${name}`); + } + }; + const check = (name: string): void => { + switch (name) { + case "envelope": + route = typedTurnRoute(decision, jsonObject(params.historical_route_text) ?? undefined); + requireThat(route !== "contract_error", "quota decision failed the shared envelope contract (schema or signature hashes)"); + if (object(decision.user).action_required === true) route = "user_action_required"; + break; + case "decision_actor": + chosen = decisionLineage(decision); + requireThat(chosen.goal_id && chosen.agent_id, "fresh quota decision is missing goal/agent lineage"); + effective = {...chosen}; break; + case "receipt_binding": + requireThat(text(params.predecessor_turn_key), "stale_receipt: continuation is missing predecessor_turn_key; cannot prove causal succession from the receipt"); + requireThat(text(params.predecessor_turn_key) === receipt?.turn_key, + `stale_receipt: fresh decision predecessor_turn_key ('${text(params.predecessor_turn_key)}') does not match the receipt turn_key ('${text(receipt?.turn_key)}')`); + for (const key of ["goal_id", "agent_id"] as const) + requireThat(receiptLineage[key] === chosen[key], + `stale_receipt: receipt lineage does not match the fresh decision on ${key} (receipt='${receiptLineage[key]}', decision='${chosen[key]}')`); + break; + case "initial_terminal": + requireThat(decision.state === EffectiveAction.TERMINAL_NO_FOLLOWUP, "terminal no-follow-up requires fresh Goal frontier state"); break; + case "initial_todo": + requireThat(chosen.todo_id || ["wait", "user_action_required"].includes(projectRoute()), + "executable quota decision is missing selected Todo lineage"); break; + case "completion_terminal": + requireThat(decision.effective_action === EffectiveAction.TERMINAL_NO_FOLLOWUP && decision.state === EffectiveAction.TERMINAL_NO_FOLLOWUP, + "no_followup completion requires fresh terminal Goal frontier evidence"); break; + case "completion_todo": + requireThat(chosen.todo_id, "Todo completion continuation requires a fresh selected Todo"); + if (fact("completion") === "successor") + requireThat(Array.isArray(completion.successor_todo_ids) && completion.successor_todo_ids.includes(chosen.todo_id), + "stale_receipt: fresh decision is not a declared completion successor"); + else if (fact("completion") === "active_goal") requireThat(chosen.todo_id !== receiptLineage.todo_id, + "stale_receipt: active Goal continuation reselected the completed Todo"); + break; + case "receipt_todo": + if (chosen.todo_id) requireThat(chosen.todo_id === receiptLineage.todo_id, + `stale_receipt: receipt Todo does not match the fresh decision (receipt='${receiptLineage.todo_id}', decision='${chosen.todo_id}')`); + else { + requireThat(kind !== "validated_progress" || ["user_action_required", "wait"].includes(route), + "receipt-backed executable decision is missing selected Todo lineage"); + effective.todo_id = receiptLineage.todo_id; + } + break; + case "progress_budget": + requireThat(params.budget != null, "validated progress cannot continue without a proven bounded turn budget"); + bounded = budget(params.budget); + for (const key of ["goal_id", "agent_id", "todo_id"] as const) + requireThat(bounded?.lineage[key] === effective[key], + `stale_budget: budget lineage does not match the fresh decision on ${key} (budget='${bounded?.lineage[key]}', decision='${effective[key]}')`); + break; + case "host_retry": { + const normalized = normalizeHostFailure(failure, + object(params.integer_facts).failure_attempt as string | null | undefined); + failure = normalized; + retryState = normalized.retryable === true + ? Number(normalized.attempt) < Number(object(normalized.retry).max_attempts) ? "available" : "exhausted" + : "not_retryable"; + break; + } + default: throw new EffectRuntimeRequestError(`unsupported controller check: ${name}`); + } + }; + for (const rule of TURN_CONTROLLER_CONTRACT.rules as readonly Rule[]) { + if (!Object.entries(rule.when).every(([name, values]) => values.includes(fact(name)))) continue; + if (rule.check) { check(rule.check); continue; } + const disposition = rule.disposition === "project_route" ? projectRoute() : rule.disposition; + const reasons = rule.reason; + const reason = typeof reasons === "string" ? reasons : reasons?.[String(disposition)] ?? reasons?.default; + requireThat(typeof reason === "string", `controller rule ${rule.id} has no reason for ${disposition}`); + const result: JsonObject = {schema_version: "loop_turn_loop_disposition_v0", disposition, + reason: reason.replaceAll("{failure_kind}", text(failure.kind) || "None").replaceAll("{receipt_kind}", kind), + spends_quota: false, launches_host: false, writes_state: false, + lineage: rule.lineage === "receipt" ? receiptLineage : rule.lineage === "decision" ? chosen : effective}; + if (disposition === "replan") result.replan_continuation = { + requires_bounded_delta: true, delta_kinds: ["todo_delta", "vision_delta"], + stale_todo_rerun_allowed: false, fresh_envelope_required: true}; + else if (rule.extra === "capability") result.capability_action = object(object(decision.action).capability_intent); + else if (rule.extra === "iteration_stop") Object.assign(result, + {stop_scope: "iteration", goal_terminal: false, continuation_required: false}); + else if (rule.extra === "retry") { + const retry = object(failure.retry); + result.retry_continuation = {same_turn: true, retry_failed_turn: true, strategy: retry.strategy, + retry_after_seconds: retry.backoff_seconds, attempt: failure.attempt, max_attempts: retry.max_attempts, + fresh_envelope_required: true, model_fallback_allowed: false}; + } + return result; + } + throw new EffectRuntimeRequestError("controller contract has no disposition for the qualified inputs"); +} + +function positive(value: unknown, field: string, exact?: unknown): number | string { + if (exact === undefined) { + requireThat(typeof value === "number" && Number.isInteger(value) && value >= 1, `${field} must be a positive integer`); + return value; + } + requireThat(typeof exact === "string" && /^\d+$/.test(exact) && BigInt(exact) >= 1n + && Number(exact) === value, `${field} must be a positive integer`); + return BigInt(exact) > BigInt(Number.MAX_SAFE_INTEGER) ? exact : Number(exact); +} + +export function decideManagedStep(params: JsonObject): JsonObject { + const journal = object(params.journal); + requireThat(journal.status === "failed", "managed step requires a failed Turn journal"); + requireThat(text(journal.result_kind) === "host_failure", "managed step requires a typed host failure Turn"); + const integers = object(params.integer_facts); + const attempt = positive(journal.host_attempt_count, "Turn journal host_attempt_count", integers.host_attempt_count); + const failure = normalizeHostFailure(journal.host_failure, integers.failure_attempt as string | null | undefined); + const retry = object(failure.retry); + const maxAttempts = failure.retryable === true ? positive(retry.max_attempts, "Turn journal retry max_attempts") : null; + for (const [name, authority, message] of [ + ["observed_attempt", attempt, "observed_attempt disagrees with the Turn journal attempt authority"], + ["observed_max_attempts", maxAttempts, "observed_max_attempts disagrees with the Turn journal retry policy"], + ] as const) { + if (params[name] != null) { + const observation = positive(params[name], name, integers[name]); + requireThat(authority === null || observation === authority, message); + } + } + requireThat(text(journal.turn_key) === params.turn_key, "Turn journal turn_key does not match the requested Turn"); + requireThat(failure.retryable === true, `host failure ${failure.kind} is not retryable; repair instead`); + // normalizeHostFailure already proved the exact integer; the decimal form is + // retained above the JSON safe range for equality and Python readback. + requireThat(attempt === failure.attempt, "Turn journal host_failure attempt disagrees with host_attempt_count"); + if (!jsonObject(journal.plan)) throw new ControllerTypeError("Turn journal has no stored plan to reconcile"); + const transaction = object(object(journal.plan).transaction); + requireThat(text(journal.turn_key) && text(transaction.turn_key) === text(journal.turn_key), + "Turn journal turn_key does not match its stored transaction"); + if (!jsonObject(journal.receipt)) throw new ControllerTypeError("Turn journal has no validated receipt to reconcile"); + const receipt = object(journal.receipt); + requireThat(receipt.schema_version === "loopx_turn_receipt_validation_v0", "Turn journal receipt has an unsupported schema"); + requireThat(receipt.ok === true, "Turn journal receipt is not a validated receipt"); + // Qualify persisted receipt fields directly. No execution-shaped object or + // caller-invented effects stand in for the original journal. + requireThat((TURN_RESULT_KINDS as readonly unknown[]).includes(receipt.result_kind), + `validated turn receipt has unsupported result_kind ${repr(receipt.result_kind)}`); + requireThat(!["validated_progress", "validated_completion"].includes(text(receipt.result_kind)), + `material result ${text(receipt.result_kind)} requires a fully committed Turn execution and receipt`); + requireThat(receipt.result_kind === "host_failure", "only a host_failure receipt may carry host failure metadata"); + const identity = lineage(receipt.lineage); + requireThat(identity.goal_id && identity.agent_id && identity.todo_id, + "validated turn receipt is missing goal/agent/todo lineage"); + requireThat(text(receipt.turn_key), "validated turn receipt is missing turn_key"); + requireThat(identity.goal_id === params.goal_id && identity.agent_id === params.agent_id, + "Turn journal lineage does not match the requested goal/agent"); + const inspection = interpretTurnJournal({schema_version: "loopx_turn_journal_interpretation_request_v0", + journal, goal_id: text(params.goal_id), agent_id: text(params.agent_id), turn_key: text(params.turn_key)}); + requireThat(inspection.journal_consistent, `Turn journal replay is blocked: ${inspection.violations.join(", ")}`); + const decision = object(params.decision); + for (const key of ["goal_id", "agent_id"]) + requireThat(!text(decision[key]) || text(decision[key]) === params[key], `fresh decision ${key} does not match the managed step`); + const disposition = decideLoopDisposition({receipt: {...receipt, lineage: identity, host_failure: journal.host_failure}, decision, + integer_facts: integers, historical_route_text: params.historical_route_text, + predecessor_turn_key: params.predecessor_turn_key == null ? params.turn_key : params.predecessor_turn_key}); + const result: JsonObject = {schema_version: "loopx_turn_managed_step_v0", disposition: disposition.disposition, + reason: disposition.reason, turn_key: params.turn_key, attempt}; + for (const key of ["lineage", "stop_scope", "goal_terminal", "continuation_required", "successor_todo_ids", + "capability_action", "capability_action_required", "user_action", "retry_continuation"]) + if (has(disposition, key)) result[key] = disposition[key]; + if (jsonObject(disposition.retry_continuation)) result.max_attempts = object(disposition.retry_continuation).max_attempts; + return result; +} + +/** One request carries all facts for the chosen public boundary. Preserve Python + * input-error classes/messages; transport/internal failures remain runtime errors. */ +export function controllerRequest(params: JsonObject): JsonObject { + try { + switch (params.operation) { + case "receipt": return {ok: true, value: qualifyExecution(params.execution, object(params.integer_facts), params.request_identity)}; + case "route": return {ok: true, value: {route: typedTurnRoute(object(params.decision), + jsonObject(params.historical_route_text) ?? undefined)}}; + case "disposition": return {ok: true, value: decideLoopDisposition(params)}; + case "managed": return {ok: true, value: decideManagedStep(params)}; + default: throw new EffectRuntimeRequestError("unsupported Turn controller operation"); + } + } catch (error) { + if (!(error instanceof EffectRuntimeRequestError)) throw error; + return {ok: false, error_type: error instanceof ControllerTypeError ? "TypeError" : "ValueError", message: error.message}; + } +} diff --git a/loopx/control_plane/turn_driver/managed_step.py b/loopx/control_plane/turn_driver/managed_step.py index 7a2bd1d5bb..88a32dbd95 100644 --- a/loopx/control_plane/turn_driver/managed_step.py +++ b/loopx/control_plane/turn_driver/managed_step.py @@ -1,332 +1,30 @@ -"""The managed-step consumer of one bounded same-Turn continuation. +"""Read-only managed continuation over one complete journal and fresh decision. -``loopx turn run-once`` commits a governed Turn and stops. When the host -failed for a retryable reason, the pure Turn Loop Controller already answers -``wait`` with a typed ``retry_continuation`` block, but nothing in production -consumed it: the outer scheduler was left to infer retryability from prose. - -This module is that consumer. It rebuilds one validated receipt from the -canonical Turn journal, asks the controller for a disposition against a fresh -``loopx_turn_envelope_v0`` decision, and returns the typed answer. It is -read-only by construction: - -- it never invokes a host, writes state, spends quota, or sleeps; -- the Turn journal stays the sole authority for attempt and max_attempts, so a - caller-supplied observation is only ever reconciled, never believed; -- a disagreement between the observation and the journal is a ``ValueError`` at - the typed-input boundary rather than a silently adopted number. - -The caller decides whether to carry the existing ``--retry-failed-turn`` / -``--resume-turn-key`` flags on the next ``run-once``; this module grants no -execution authority of its own. +The typed owner composes current journal qualification with the ordered +controller. Host scheduling, retries, dispatch and settlement keep their owners. """ - from __future__ import annotations from collections.abc import Mapping -from typing import Any, cast +from typing import Any 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, - ValidatedTurnReceipt, - decide_loop_disposition, -) -from .turn_journal_runtime import interpret_turn_journal_projection -from .transaction import ( - LOOPX_TURN_EXECUTION_SCHEMA_VERSION, - LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION, - LoopXTurnResultKind, -) +from .loop_controller import _controller_result LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION = "loopx_turn_managed_step_v0" -def _mapping(value: Any) -> dict[str, Any]: - return dict(value) if isinstance(value, Mapping) else {} - - -def _require_positive_int(value: Any, *, field: str) -> int: - if isinstance(value, bool) or not isinstance(value, int) or value < 1: - raise ValueError(f"{field} must be a positive integer") - return value - - -def _require_matching_attempts(journal_attempt: Any, failure_attempt: Any) -> None: - """Refuse a journal whose two persisted attempt fields disagree. - - The executor increments ``host_attempt_count`` and ``record_host_failure`` - writes the same value into ``host_failure.attempt``. The controller reads - the nested field for the retry ceiling while the projection reports the top - level, so a divergence would present one attempt count and authorize on - another. - """ - - authority = _require_positive_int( - journal_attempt, - field="Turn journal host_attempt_count", - ) - nested = _require_positive_int( - failure_attempt, - field="Turn journal host_failure attempt", - ) - if nested != authority: - raise ValueError( - "Turn journal host_failure attempt disagrees with host_attempt_count" - ) - - -def _validated_turn_receipt(journal: Mapping[str, Any]) -> ValidatedTurnReceipt: - """Rebuild the controller receipt from the canonical journal alone. - - The journal stores the plan and the validated receipt rather than an - execution payload, so the projection here is reconstructed from committed - fields only. ``ValidatedTurnReceipt.from_execution`` then re-enforces the - material-effect and lineage contracts, which keeps this reader honest - without duplicating them. - """ - - plan = journal.get("plan") - if not isinstance(plan, Mapping): - raise TypeError("Turn journal has no stored plan to reconcile") - stored_transaction = _mapping(plan.get("transaction")) - turn_key = str(journal.get("turn_key") or "") - if not turn_key or str(stored_transaction.get("turn_key") or "") != turn_key: - raise ValueError("Turn journal turn_key does not match its stored transaction") - - receipt = journal.get("receipt") - if not isinstance(receipt, Mapping): - raise TypeError("Turn journal has no validated receipt to reconcile") - result_kind = str(receipt.get("result_kind") or "") - if receipt.get("schema_version") != LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION: - raise ValueError("Turn journal receipt has an unsupported schema") - if receipt.get("ok") is not True: - raise ValueError("Turn journal receipt is not a validated receipt") - - execution: dict[str, Any] = { - "ok": journal.get("status") == "committed", - "schema_version": LOOPX_TURN_EXECUTION_SCHEMA_VERSION, - "mode": "managed_step", - "status": journal.get("status"), - "result_kind": result_kind, - "receipt": dict(receipt), - "scheduler": _mapping(journal.get("scheduler")), - "validation": journal.get("task_validation"), - **({"validation_stage": journal["validation_stage"]} if journal.get("validation_stage") else {}), - **project_host_failure(journal), - } - settlement_result = journal.get("settlement_result") - if isinstance(settlement_result, Mapping): - execution["settlement_result"] = dict(settlement_result) - writeback = _mapping(journal.get("writeback")) - completion = writeback.get("completion") - if isinstance(completion, Mapping): - execution["todo_completion"] = dict(completion) - return ValidatedTurnReceipt.from_execution(execution) - - -def managed_step_receipt_from_journal( - journal: Mapping[str, Any], - *, - goal_id: str, - agent_id: str, - turn_key: str, -) -> ValidatedTurnReceipt: - """Qualify one journaled Turn for the managed-step transition. - - Fails closed unless the journal is a finished failed Turn for exactly this - goal/agent/turn identity whose typed host failure is retryable. The current - snapshot must pass the canonical TS journal consistency check before the - controller is called; historical recovery audits never establish eligibility. - """ - - if journal.get("status") != "failed": - raise ValueError("managed step requires a failed Turn journal") - if str(journal.get("turn_key") or "") != turn_key: - raise ValueError("Turn journal turn_key does not match the requested Turn") - result_kind = str(journal.get("result_kind") or "") - if result_kind != LoopXTurnResultKind.HOST_FAILURE.value: - raise ValueError("managed step requires a typed host failure Turn") - - failure = normalize_host_failure_record(journal.get("host_failure")) - if failure.get("retryable") is not True: - raise ValueError( - f"host failure {failure.get('kind')} is not retryable; repair instead" - ) - # The journal records the attempt twice: once as the top-level - # `host_attempt_count` the executor increments, and once inside the typed - # `host_failure` record the controller reads to decide whether the retry - # ceiling is reached. `record_host_failure` writes them from one value, so a - # disagreement means the journal was edited or corrupted. Preferring either - # side would let the presentation report `3/3` while the controller still - # authorized a retry, so refuse the Turn instead. - _require_matching_attempts( - journal.get("host_attempt_count"), - failure.get("attempt"), - ) - - receipt = _validated_turn_receipt(journal) - lineage = receipt.lineage - if lineage["goal_id"] != goal_id or lineage["agent_id"] != agent_id: - raise ValueError("Turn journal lineage does not match the requested goal/agent") - # Historical recovery_audit is explanatory, not proof about this snapshot. - # Reuse the same TS consistency rules as inspect-journal/run-once. This is - # not a retry request: fresh host-session validation remains with run-once, - # while the controller below owns the bounded retry disposition. - inspection = interpret_turn_journal_projection( - journal, goal_id=goal_id, agent_id=agent_id, turn_key=turn_key, - ) - if inspection["journal_consistent"] is not True: - raise ValueError( - "Turn journal replay is blocked: " + ", ".join(cast(list[str], inspection["violations"])) - ) - return receipt - - -def reconcile_observed_attempt( - journal: Mapping[str, Any], - *, - observed_attempt: int | None, - observed_max_attempts: int | None = None, -) -> None: - """Reconcile a caller-supplied observation against the journal authority. - - The journal owns the attempt count. An observation that disagrees with it is - a contract violation, not a correction, so this raises instead of preferring - either side. - """ - - authority = _require_positive_int( - journal.get("host_attempt_count"), - field="Turn journal host_attempt_count", - ) - failure = normalize_host_failure_record(journal.get("host_failure")) - retry = _mapping(failure.get("retry")) - # A non-retryable failure carries no retry policy, so there is no ceiling to - # reconcile against; the caller is refused for the failure kind itself. - max_attempts = ( - _require_positive_int( - retry.get("max_attempts"), - field="Turn journal retry max_attempts", - ) - if failure.get("retryable") is True - else None - ) - if observed_attempt is not None: - _require_positive_int(observed_attempt, field="observed_attempt") - if observed_attempt != authority: - raise ValueError( - "observed_attempt disagrees with the Turn journal attempt authority" - ) - if observed_max_attempts is not None: - _require_positive_int(observed_max_attempts, field="observed_max_attempts") - if max_attempts is not None and observed_max_attempts != max_attempts: - raise ValueError( - "observed_max_attempts disagrees with the Turn journal retry policy" - ) - - -def _managed_payload( - disposition: Mapping[str, Any], - *, - turn_key: str, - journal: Mapping[str, Any], -) -> dict[str, Any]: - """Project the controller decision plus its journal-proven provenance.""" - - payload: dict[str, Any] = { - "schema_version": LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, - "disposition": disposition.get("disposition"), - "reason": disposition.get("reason"), - "turn_key": turn_key, - "attempt": _require_positive_int( - journal.get("host_attempt_count"), - field="Turn journal host_attempt_count", - ), - } - for key in ( - "lineage", - "stop_scope", - "goal_terminal", - "continuation_required", - "successor_todo_ids", - "capability_action", - "capability_action_required", - "user_action", - ): - if key in disposition: - payload[key] = disposition[key] - retry_continuation = disposition.get("retry_continuation") - if isinstance(retry_continuation, Mapping): - payload["retry_continuation"] = dict(retry_continuation) - # The journal is the authority for the budget; echoing it here keeps a - # caller from having to trust the controller's copy. - payload["max_attempts"] = _require_positive_int( - retry_continuation.get("max_attempts"), - field="retry_continuation max_attempts", - ) - return payload - - def decide_managed_step( - journal: Mapping[str, Any], - fresh_decision: Mapping[str, Any] | TurnControllerInput, - *, - goal_id: str, - agent_id: str, - turn_key: str, + journal: Mapping[str, Any], fresh_decision: Mapping[str, Any] | TurnControllerInput, *, + goal_id: str, agent_id: str, turn_key: str, predecessor_turn_key: str | None = None, - observed_attempt: int | None = None, - observed_max_attempts: int | None = None, + observed_attempt: int | None = None, observed_max_attempts: int | None = None, ) -> dict[str, Any]: - """Decide the managed continuation for one failed Turn. - - ``fresh_decision`` is a fresh ``loopx_turn_envelope_v0`` from the control - plane, exactly as ``run-once`` builds one. ``predecessor_turn_key`` carries - the causal link to the failed Turn and is deliberately *not* written into - the signed envelope; it defaults to the Turn under decision, which is the - only successor relationship a managed step can prove on its own. - - The answer is typed and grants no execution authority: ``wait`` means "wake - the same Turn after the bounded backoff", never "spend or write now". - """ - - journal = _mapping(journal) - # Refuse a non-failed or non-host-failure Turn before any reconciliation, so - # the caller's first error is about the input it actually supplied. - if journal.get("status") != "failed": - raise ValueError("managed step requires a failed Turn journal") - if str(journal.get("result_kind") or "") != LoopXTurnResultKind.HOST_FAILURE.value: - raise ValueError("managed step requires a typed host failure Turn") - # Reconcile before deciding so a forged observation cannot influence which - # branch is taken. - reconcile_observed_attempt( - journal, - observed_attempt=observed_attempt, + """Qualify and decide in one runtime crossing, without reconstructing execution.""" + return _controller_result( + "managed", journal=dict(journal) if isinstance(journal, Mapping) else {}, + decision=fresh_decision, + goal_id=goal_id, agent_id=agent_id, turn_key=turn_key, + predecessor_turn_key=predecessor_turn_key, observed_attempt=observed_attempt, observed_max_attempts=observed_max_attempts, ) - receipt = managed_step_receipt_from_journal( - journal, - goal_id=goal_id, - agent_id=agent_id, - turn_key=turn_key, - ) - 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=fresh_decision if isinstance(fresh_decision, TurnControllerInput) else envelope, - predecessor_turn_key=( - turn_key if predecessor_turn_key is None else predecessor_turn_key - ), - ) - if str(disposition.get("schema_version") or "") not in { - "", - LOOP_CONTROLLER_DISPOSITION_SCHEMA_VERSION, - }: - raise RuntimeError("Turn Loop Controller returned an unsupported schema") - return _managed_payload(disposition, turn_key=turn_key, journal=journal) diff --git a/loopx/control_plane/turn_driver/turn_contract_generated.py b/loopx/control_plane/turn_driver/turn_contract_generated.py index 66afc6c1eb..1a12d54f5e 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: 1d0bc416332a409159549e88ce700ceef4a4c64026154ae091469ba5b6c4e293""" +SHA256: be2734e9f192e8d78598c06e338dafc72ed71bb0251a2ec3784cb8421c4fd78d""" from __future__ import annotations from enum import Enum @@ -40,20 +40,21 @@ class LoopDisposition(str, Enum): TURN_CONTROLLER_CONTRACT = {'schema_version': 'loop_turn_controller_decision_contract_v0', 'evaluation': 'ordered_checks_then_first_matching_return', - 'qualification': {'receipt': 'ValidatedTurnReceipt.from_execution; material settlement, durable ' + 'qualification': {'receipt': 'TypeScript qualifyExecution; material settlement, durable ' 'completion and host failure qualification remain admission ' - 'preconditions', - 'budget': 'BoundedTurnBudget; strict integers, finite ranges and nonempty ' - 'lineage remain admission preconditions', + 'preconditions; Python retains the public value-object API', + 'budget': 'BoundedTurnBudget retains its strict integer public API; exact ' + 'decimal transport avoids JSON integer rounding; the TS owner checks ' + 'the reached continuation budget', '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. ' '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.'}, + 'in TypeScript; the TS ordered controller combines ' + 'qualified receipt, quota, causal predecessor and ' + 'budget. Python transports complete requests; journal ' + 'recovery remains a separate composed decision.'}, 'partitions': {'receipt_kind': {'owner': 'LoopXTurnResultKind', 'additional': ['absent']}, 'route': {'owner': 'LoopXTurnRoute'}, 'receipt_present': [False, True], diff --git a/loopx/control_plane/turn_driver/turn_contract_generated.ts b/loopx/control_plane/turn_driver/turn_contract_generated.ts index abf0fa2fb4..54499d4b48 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: 1d0bc416332a409159549e88ce700ceef4a4c64026154ae091469ba5b6c4e293 +// SHA256: be2734e9f192e8d78598c06e338dafc72ed71bb0251a2ec3784cb8421c4fd78d export const TURN_RESULT_KINDS = [ "validated_progress", @@ -47,10 +47,10 @@ export const TURN_CONTROLLER_CONTRACT = { "schema_version": "loop_turn_controller_decision_contract_v0", "evaluation": "ordered_checks_then_first_matching_return", "qualification": { - "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", + "receipt": "TypeScript qualifyExecution; material settlement, durable completion and host failure qualification remain admission preconditions; Python retains the public value-object API", + "budget": "BoundedTurnBudget retains its strict integer public API; exact decimal transport avoids JSON integer rounding; the TS owner checks the reached continuation budget", "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. 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." + "result_route_relation": "No unary result-kind to route projection exists. turn_envelope.turnEnvelopeRoute qualifies quota input in TypeScript; the TS ordered controller combines qualified receipt, quota, causal predecessor and budget. Python transports complete requests; journal recovery remains a separate composed decision." }, "partitions": { "receipt_kind": { diff --git a/loopx/control_plane/turn_driver/turn_journal.ts b/loopx/control_plane/turn_driver/turn_journal.ts index 442c5f295a..91e93362a4 100644 --- a/loopx/control_plane/turn_driver/turn_journal.ts +++ b/loopx/control_plane/turn_driver/turn_journal.ts @@ -1,3 +1,4 @@ +import { normalizeHostFailure } from "./host_failure.ts"; import transactionContract from "../turn_transaction_contract.json" with { type: "json", }; @@ -160,29 +161,6 @@ export function journalPhaseViolation( return null; } -const hostFailureKinds: ReadonlySet = new Set([ - "auth_failed", - "contract_rejected", - "executor_timeout", - "output_budget_exhausted", - "provider_capacity", - "provider_overloaded", - "quota_exhausted", - "rate_limited", - "session_missing", - "transport_lost", - "unknown", -]); -const hostRetryPolicies: Readonly> = Object.freeze({ - executor_timeout: { maxAttempts: 2, baseBackoffSeconds: 5 }, - provider_capacity: { maxAttempts: 3, baseBackoffSeconds: 30 }, - provider_overloaded: { maxAttempts: 3, baseBackoffSeconds: 30 }, - rate_limited: { maxAttempts: 3, baseBackoffSeconds: 60 }, - transport_lost: { maxAttempts: 3, baseBackoffSeconds: 10 }, -}); function asObject(value: unknown): JsonObject { return typeof value === "object" && value !== null && !Array.isArray(value) @@ -340,58 +318,20 @@ function preparedEffectCheck( }; } -function hasExactKeys(value: JsonObject, expected: readonly string[]): boolean { - const actual = Object.keys(value); - return actual.length === expected.length && - expected.every((key) => Object.hasOwn(value, key)); -} - function hostRetryPolicyCheck(journal: JsonObject): TurnRecoveryCheck | null { if (!Object.hasOwn(journal, "host_failure")) return null; - const failure = asObject(journal.host_failure); + let failure: JsonObject; + try { failure = normalizeHostFailure(journal.host_failure); } + catch (error) { + if (!(error instanceof EffectRuntimeRequestError)) throw error; + return {kind: "host_retry_policy", outcome: "failed", reason: "host_retry_contract_invalid"}; + } const retry = asObject(failure.retry); const attempt = failure.attempt; - const hostAttemptCount = journal.host_attempt_count; + const kind = String(failure.kind); const retryable = failure.retryable; - const kind = typeof failure.kind === "string" ? failure.kind : ""; - const policy = hostRetryPolicies[kind]; - const expectedRetryable = policy !== undefined; - const expectedBackoff = policy && Number.isInteger(attempt) - ? Math.min(policy.baseBackoffSeconds * (2 ** Math.max(0, Number(attempt) - 1)), 300) - : null; - const failureFieldsValid = hasExactKeys( - failure, - expectedRetryable - ? ["schema_version", "kind", "attempt", "retryable", "retry"] - : ["schema_version", "kind", "attempt", "retryable"], - ); - const contractValid = - failureFieldsValid && - failure.schema_version === "loopx_turn_host_failure_v0" && - hostFailureKinds.has(kind) && - Number.isInteger(attempt) && - Number(attempt) >= 1 && - hostAttemptCount === attempt && - typeof retryable === "boolean" && - retryable === expectedRetryable && - ( - retryable === false - ? !Object.hasOwn(failure, "retry") - : hasExactKeys( - retry, - ["strategy", "max_attempts", "backoff_seconds"], - ) && - retry.strategy === "same_configuration" && - retry.max_attempts === policy?.maxAttempts && - retry.backoff_seconds === expectedBackoff - ); - if (!contractValid) { - return { - kind: "host_retry_policy", - outcome: "failed", - reason: "host_retry_contract_invalid", - }; - } + if (journal.host_attempt_count !== attempt) + return {kind: "host_retry_policy", outcome: "failed", reason: "host_retry_contract_invalid"}; if (retryable === true && Number(attempt) >= Number(retry.max_attempts)) { return { kind: "host_retry_policy", diff --git a/loopx/control_plane/turn_loop_controller_contract_v0.json b/loopx/control_plane/turn_loop_controller_contract_v0.json index 6417a482da..0cf1876864 100644 --- a/loopx/control_plane/turn_loop_controller_contract_v0.json +++ b/loopx/control_plane/turn_loop_controller_contract_v0.json @@ -2,10 +2,10 @@ "schema_version": "loop_turn_controller_decision_contract_v0", "evaluation": "ordered_checks_then_first_matching_return", "qualification": { - "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", + "receipt": "TypeScript qualifyExecution; material settlement, durable completion and host failure qualification remain admission preconditions; Python retains the public value-object API", + "budget": "BoundedTurnBudget retains its strict integer public API; exact decimal transport avoids JSON integer rounding; the TS owner checks the reached continuation budget", "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. 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." + "result_route_relation": "No unary result-kind to route projection exists. turn_envelope.turnEnvelopeRoute qualifies quota input in TypeScript; the TS ordered controller combines qualified receipt, quota, causal predecessor and budget. Python transports complete requests; journal recovery remains a separate composed decision." }, "partitions": { "receipt_kind": { diff --git a/loopx/semantics/production.py b/loopx/semantics/production.py index a7574660a5..89a227d129 100644 --- a/loopx/semantics/production.py +++ b/loopx/semantics/production.py @@ -367,7 +367,13 @@ def probe_quota_run_decision_production(vocabulary: dict[str, Any]) -> list[Prod SETTLEMENT_IDENTITY_LINE = 408 QUOTA_RUN_DECISION_SITE = 'loopx/control_plane/quota/run_decision.ts::resolveQuotaRunDecision' +def _probe_turn_routes(vocabulary: dict[str, Any]) -> list[Production]: + from .turn_contract_witness import probe_route_production + return probe_route_production() + + INPUT_WITNESSES: dict[str, Callable[[dict[str, Any]], list[Production]]] = { + 'loopx/control_plane/turn_driver/driver.py::_typed_route': _probe_turn_routes, QUOTA_RUN_DECISION_SITE: probe_quota_run_decision_production, 'loopx/control_plane/turn_driver/transaction.py::_result_kind': probe_turn_result_input_domain, 'loopx/control_plane/turn_driver/loop_controller.py::decide_loop_disposition': _probe_controller_domain, diff --git a/loopx/semantics/turn_contract_witness.py b/loopx/semantics/turn_contract_witness.py index 62f0fc4f03..84547c92d5 100644 --- a/loopx/semantics/turn_contract_witness.py +++ b/loopx/semantics/turn_contract_witness.py @@ -190,11 +190,8 @@ def probe_controller_production(): ) verified_generated_paths() - validate_contract(c._LOOP_CONTROLLER_CONTRACT) - if ( - c._LOOP_CONTROLLER_CONTRACT is not shared.TURN_CONTROLLER_CONTRACT - or c._LOOP_CONTROLLER_CONTRACT != read_contract() - ): + validate_contract(shared.TURN_CONTROLLER_CONTRACT) + if shared.TURN_CONTROLLER_CONTRACT != read_contract(): raise ValueError("controller is disconnected from its generated contract") site = "loopx/control_plane/turn_driver/loop_controller.py::decide_loop_disposition" rows = [] @@ -294,3 +291,29 @@ def probe_projection_production(): ) ) return rows + + +def probe_route_production(): + """Execute the actual Python-to-TS route boundary over independent inputs.""" + from loopx.control_plane.turn_driver import driver + + rows = [] + cases = (("ready", "ready_for_host"), ("capability", "capability_action_required"), + ("repair", "repair_required"), ("replan", "replan_required"), + ("user", "user_action_required"), ("blocked", "blocked"), + ("quiet", "wait"), ("invalid", "contract_error")) + for case, expected in cases: + decision = controller_input("absent", case)["quota_decision"] + if case == "user": + decision["should_run"] = False + elif case == "quiet": + decision["should_run"] = False + decision["action"]["quiet_noop_allowed"] = True + elif case == "invalid": + decision["action_signature"] = {} + actual = driver._typed_route(decision).value + if actual != expected: + raise ValueError(f"Turn route witness {case} violates the independent input contract") + rows.append(Production("loopx/control_plane/turn_driver/driver.py::_typed_route", + driver._typed_route.__code__.co_firstlineno, "input_witness", frozenset({actual}), False)) + return rows diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index 468fb17415..64e18f3d94 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -274,13 +274,11 @@ "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::build_loopx_turn_plan", - "loopx/control_plane/turn_driver/loop_controller.py::_envelope_route" + "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan" ], "return_producers": [ "loopx/control_plane/quota/turn_envelope.ts::turnEnvelopeRoute", "loopx/control_plane/turn_driver/driver.py::_typed_route", - "loopx/control_plane/turn_driver/loop_controller.py::_envelope_route", "loopx/control_plane/turn_driver/driver.py::build_loopx_turn_plan" ], "return_paths": { @@ -288,7 +286,8 @@ "route", "kind" ] - } + }, + "input_producer": "loopx/control_plane/turn_driver/driver.py::_typed_route" }, "loop_disposition": { "meaning": "Pure controller verdict for the outer loop after combining the last Turn receipt with the fresh route.", @@ -319,18 +318,13 @@ "terminal": "A terminal action with no prior receipt, or a validated completion declaring the no_followup continuation." }, "producers": [ - "loopx/control_plane/turn_driver/loop_controller.py::_replan_disposition", "loopx/control_plane/turn_driver/loop_controller.py::decide_loop_disposition", "loopx/control_plane/turn_driver/turn_contract_generated.py::project_turn_route" ], "return_producers": [ "loopx/control_plane/turn_driver/turn_contract_generated.py::project_turn_route" ], - "call_producers": { - "loopx/control_plane/turn_driver/loop_controller.py::_disposition": [ - "disposition" - ] - }, + "call_producers": {}, "input_producer": "loopx/control_plane/turn_driver/loop_controller.py::decide_loop_disposition" }, "agent_scope_frontier_action": { @@ -1167,7 +1161,7 @@ "dual_runtime_twins": { "meaning": "Same-basename .py/.ts module pairs under the control plane. The TypeScript migration RFC is replacement-first, so this count may only fall.", "root": "loopx/control_plane", - "module_budget": 43 + "module_budget": 45 }, "inventory_ratchets": { "meaning": "Counts read from the generated inventory. A same-runtime fork is one constant name with one value defined in two or more modules of the same runtime; a conflicting value is one name with different values. Both the number of affected names and the number of definitions are budgets, so a third spelling of an already-conflicting name is still a regression.", diff --git a/tests/architecture/test_turn_contract_generation.py b/tests/architecture/test_turn_contract_generation.py index c2275161b8..cfeffd9788 100644 --- a/tests/architecture/test_turn_contract_generation.py +++ b/tests/architecture/test_turn_contract_generation.py @@ -195,7 +195,7 @@ def test_controller_and_projection_witnesses_cover_live_dispositions(): @pytest.mark.parametrize( "defect", - ["missing_rule", "unregistered_output", "disconnected_table", "constant_emitter"], + ["missing_rule", "unregistered_output", "constant_emitter"], ) def test_controller_production_rejects_contract_or_wiring_regressions( monkeypatch, defect @@ -203,7 +203,8 @@ def test_controller_production_rejects_contract_or_wiring_regressions( from loopx.control_plane.turn_driver import loop_controller as c from loopx.semantics.turn_contract_witness import probe_controller_production - table = deepcopy(c._LOOP_CONTROLLER_CONTRACT) + from loopx.control_plane.turn_driver import turn_contract_generated as shared + table = deepcopy(shared.TURN_CONTROLLER_CONTRACT) if defect == "missing_rule": table["rules"] = [ r for r in table["rules"] if r["id"] != "receipt_iteration_failed" @@ -215,7 +216,7 @@ def test_controller_production_rejects_contract_or_wiring_regressions( c, "decide_loop_disposition", lambda **kw: {"disposition": "run_now"} ) else: - monkeypatch.setattr(c, "_LOOP_CONTROLLER_CONTRACT", table) + monkeypatch.setattr(shared, "TURN_CONTROLLER_CONTRACT", table) with pytest.raises(ValueError): probe_controller_production() @@ -231,31 +232,9 @@ def test_generated_provenance_is_not_a_filename_exception(tmp_path, monkeypatch) generator.verified_generated_paths() -@pytest.mark.parametrize("defect", ["remove_stop", "terminal_becomes_repair"]) -def test_live_witness_fails_even_if_changed_table_is_freshly_generated( - monkeypatch, defect -): - from loopx.control_plane.turn_driver import turn_contract_generated as shared - from loopx.semantics.turn_contract_witness import probe_controller_production - - rules = deepcopy(shared.TURN_CONTROLLER_CONTRACT["rules"]) - if defect == "remove_stop": - rules = [row for row in rules if row["id"] != "receipt_iteration_failed"] - else: - next(row for row in rules if row["id"] == "initial_terminal")["disposition"] = ( - "repair" - ) - monkeypatch.setitem(shared.TURN_CONTROLLER_CONTRACT, "rules", rules) - # Isolate the liveness obligation from the separately tested byte-provenance - # gate: model a freshly regenerated but semantically wrong contract. The - # real controller still interprets the changed rules and the oracle stays - # the independent input/output cases, not the changed data. - monkeypatch.setattr(generator, "verified_generated_paths", lambda: frozenset()) - monkeypatch.setattr( - generator, "read_contract", lambda: shared.TURN_CONTROLLER_CONTRACT - ) - with pytest.raises(ValueError): - probe_controller_production() +# Live rule mutations run against the actual TS owner in +# tests/control_plane_ts/turn_loop_controller.test.ts. Changing a Python cache +# cannot reprogram that owner after this migration. def test_new_independent_twin_cannot_hide_behind_generated_pair(monkeypatch): diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 681d147346..33ba74cd66 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import json import os import shutil @@ -1748,6 +1749,59 @@ def test_large_local_response_digest_failure_is_ambiguous(tmp_path: Path) -> Non assert error.value.diagnostic_code == "runtime_response_ambiguous" +def test_controller_snapshot_retains_private_file_and_method_guards( + tmp_path: Path, monkeypatch, +) -> None: + import tempfile + + from tests.test_loop_turn_loop_controller import _envelope + + monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: tmp_path / "runtime") + effect_runtime.effect_runtime_result("runtime.ping", {}) + fingerprint = effect_runtime._runtime_fingerprint() + info = effect_runtime._read_info(effect_runtime._runtime_info_path(fingerprint), fingerprint=fingerprint) + assert info is not None + with tempfile.TemporaryDirectory(prefix="loopx-effect-", dir=tmp_path) as directory: + root = Path(directory) + params = root / "params.json" + sink = root / "response.json" + decision = _envelope(should_run=True) + decision["work_context"] = {"text": "G" * 2_420_000} + params.write_text(json.dumps({"operation": "disposition", "decision": decision, "receipt": None})) + sink.touch() + params.chmod(0o600) + sink.chmod(0o600) + ref = {"schema_version": "loopx_effect_runtime_snapshot_v0", "path": str(params), + "byte_count": params.stat().st_size, "sha256": hashlib.sha256(params.read_bytes()).hexdigest()} + + def request(reference, method="turn.controller"): + return _raw_runtime_response(info, (json.dumps({ + "schema_version": effect_runtime.EFFECT_RUNTIME_REQUEST_SCHEMA_VERSION, + "token": info["token"], "request_id": "controller-snapshot", "method": method, + "params_ref": reference, "response_sink": str(sink), + }) + "\n").encode()) + + result = request(ref) + assert result["ok"] is True and result["result"]["value"]["disposition"] == "run_now" + for bad_ref in ({**ref, "sha256": "0" * 64}, {**ref, "byte_count": ref["byte_count"] - 1}, + {**ref, "byte_count": effect_runtime.MAX_LOCAL_SNAPSHOT_BYTES + 1}, + {**ref, "schema_version": "unsupported"}): + rejected = request(bad_ref) + assert rejected["ok"] is False and rejected["error"]["kind"] == "request_rejected" + rejected = request(ref, method="runtime.ping") + assert rejected["ok"] is False and "snapshot transport is unavailable" in rejected["error"]["message"] + if os.name != "nt": + params.chmod(0o644) + assert request(ref)["ok"] is False + params.chmod(0o600) + link = root / "link.json" + link.symlink_to(params) + assert request({**ref, "path": str(link)})["ok"] is False + assert request(ref)["ok"] is True + assert not root.exists() + effect_runtime.effect_runtime_result("runtime.shutdown", {}, retry_safe=False) + + @pytest.mark.parametrize("unit", [b"x", "界".encode()]) def test_oversized_raw_socket_request_returns_typed_error( tmp_path: Path, diff --git a/tests/control_plane_ts/turn_loop_controller.test.ts b/tests/control_plane_ts/turn_loop_controller.test.ts new file mode 100644 index 0000000000..90ecfb6d1d --- /dev/null +++ b/tests/control_plane_ts/turn_loop_controller.test.ts @@ -0,0 +1,126 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { controllerRequest } from "../../loopx/control_plane/turn_driver/loop_controller.ts"; +import { TURN_CONTROLLER_CONTRACT } from "../../loopx/control_plane/turn_driver/turn_contract_generated.ts"; +import type { JsonObject } from "../../loopx/control_plane/runtime_decode.ts"; + +const identity = {goal_id: "fixture-goal", agent_id: "fixture-agent", todo_id: "fixture-todo"}; +function input(kind = "validated_progress", action = "normal_run"): JsonObject { + return {operation: "disposition", predecessor_turn_key: "fixture-turn", + receipt: {result_kind: kind, lineage: identity, turn_key: "fixture-turn", + host_failure: kind === "host_failure" ? {schema_version: "loopx_turn_host_failure_v0", + kind: "provider_capacity", attempt: 1, retryable: true, + retry: {strategy: "same_configuration", max_attempts: 3, backoff_seconds: 30}} : undefined, + task_failure: kind === "validation_failed" ? {recovery_kind: "replan_required"} : undefined}, + decision: {...identity, schema_version: "loopx_turn_envelope_v0", should_run: true, effective_action: action, + action_signature: {matches: true, source_hash: "same", envelope_hash: "same"}, + action: {must_attempt: true, delivery_allowed: true, selected_todo: {todo_id: identity.todo_id}}}, + budget: {lineage: identity, max_turns: "1", completed_turns: "1"}}; +} + +// Mutate the real imported contract only within a synchronous test, then restore +// it. The production request does not accept a caller-provided program. +type MutableRule = {id: string; when: Record; [key: string]: unknown}; +const contract = TURN_CONTROLLER_CONTRACT as unknown as {rules: MutableRule[]}; +test("conjunction key order cannot change the actual owner's decisions", () => { + const original = contract.rules; + const cases = [input(), input("host_failure"), input("host_failure", "agent_workspace_repair"), + input("validation_failed"), input("iteration_failed"), {...input(), receipt: null}]; + const expected = cases.map(controllerRequest); + try { + contract.rules = original.map(rule => ({...rule, when: Object.fromEntries(Object.entries(rule.when).reverse())})); + assert.deepEqual(cases.map(controllerRequest), expected); + } finally { contract.rules = original; } +}); + +for (const [specific, general, request] of [ + ["progress_exhausted", "progress_route", input()], + ["host_replan", "failed_receipt", input("host_failure", "autonomous_replan")], + ["host_repair", "failed_receipt", input("host_failure", "agent_workspace_repair")], + ["host_retry", "failed_receipt", input("host_failure")], + ["host_exhausted", "failed_receipt", {...input("host_failure"), receipt: { + result_kind: "host_failure", lineage: identity, turn_key: "fixture-turn", + host_failure: {schema_version: "loopx_turn_host_failure_v0", kind: "provider_capacity", attempt: 3, + retryable: true, retry: {strategy: "same_configuration", max_attempts: 3, backoff_seconds: 120}}}}], + ["task_validation_replan", "failed_receipt", input("validation_failed")], +] as const) { + test(`${specific} must precede ${general} in the production owner`, () => { + const original = contract.rules; + const expected = controllerRequest(request); + assert.equal(expected.ok, true); + const moved = original.find(rule => rule.id === specific)!; + assert.ok(original.indexOf(moved) < original.findIndex(rule => rule.id === general)); + try { + const reordered = original.filter(rule => rule !== moved); + reordered.splice(reordered.findIndex(rule => rule.id === general) + 1, 0, moved); + contract.rules = reordered; + assert.notDeepEqual(controllerRequest(request), expected); + } finally { contract.rules = original; } + }); +} + +test("earlier user/capability returns do not demand later budget or retry facts", () => { + const request = input(); + request.budget = {not_a_budget: true}; + const decision = request.decision as JsonObject; + decision.user = {action_required: true}; + assert.equal((controllerRequest(request).value as JsonObject).disposition, "user_action_required"); + delete decision.user; + decision.effective_action = "governed_capability_intent"; + (decision.action as JsonObject).capability_intent = {schema_version: "pending_capability_intent_projection_v0", + goal_id: identity.goal_id, agent_id: identity.agent_id, command: "fixture"}; + assert.equal((controllerRequest(request).value as JsonObject).disposition, "capability_action_required"); + const failure = input("host_failure", "agent_workspace_repair"); + ((failure.receipt as JsonObject).host_failure as JsonObject).retry = {unreadable_later_policy: true}; + assert.equal((controllerRequest(failure).value as JsonObject).disposition, "repair"); +}); + +test("budget comparison preserves integers beyond JSON's exact numeric range", () => { + const request = input(); + request.budget = {lineage: identity, max_turns: "9007199254740993", completed_turns: "9007199254740992"}; + assert.equal((controllerRequest(request).value as JsonObject).disposition, "run_now"); +}); + +test("one input with multiple errors keeps the earlier causal refusal", () => { + const request = input(); + request.predecessor_turn_key = "foreign-turn"; + request.budget = null; + const response = controllerRequest(request); + assert.equal(response.ok, false); + assert.match(String(response.message), /predecessor_turn_key/); + assert.equal(response.error_type, "ValueError"); +}); + +for (const defect of ["remove_stop", "terminal_becomes_repair", "unregistered_output", "disconnected_table"]) { + test(`independent expectations detect the live TS contract defect: ${defect}`, () => { + const original = contract.rules; + const request = defect === "remove_stop" ? input("iteration_failed") : {...input(), receipt: null}; + if (defect === "terminal_becomes_repair") { + (request.decision as JsonObject).effective_action = "terminal_no_followup"; + (request.decision as JsonObject).state = "terminal_no_followup"; + } + const expected = controllerRequest(request); + assert.equal(expected.ok, true); + try { + contract.rules = structuredClone(original); + if (defect === "remove_stop") contract.rules = contract.rules.filter(rule => rule.id !== "receipt_iteration_failed"); + else if (defect === "disconnected_table") contract.rules = []; + else contract.rules.find(rule => rule.id === (defect === "terminal_becomes_repair" ? "initial_terminal" : "initial_route"))!.disposition = defect === "unregistered_output" ? "unknown" : "repair"; + assert.notDeepEqual(controllerRequest(request), expected); + } finally { contract.rules = original; } + }); +} + +test("empty legacy JSON values do not grant capability or signature admission", () => { + for (const empty of [[], {}]) { + const request = input(); + request.receipt = null; + const decision = request.decision as JsonObject; + decision.effective_action = "governed_capability_intent"; + (decision.action as JsonObject).capability_intent = {schema_version: "pending_capability_intent_projection_v0", + goal_id: identity.goal_id, agent_id: identity.agent_id, command: empty}; + assert.equal(controllerRequest(request).ok, false); + decision.action_signature = {matches: true, source_hash: empty, envelope_hash: empty}; + assert.equal(controllerRequest(request).ok, false); + } +}); diff --git a/tests/test_loop_turn_controller_contract.py b/tests/test_loop_turn_controller_contract.py index 11f1577392..c80768b31e 100644 --- a/tests/test_loop_turn_controller_contract.py +++ b/tests/test_loop_turn_controller_contract.py @@ -13,6 +13,10 @@ from loopx.control_plane.turn_driver import loop_controller as controller from loopx.control_plane.turn_driver.driver import LoopXTurnRoute +from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan +from loopx.control_plane.turn_driver.turn_contract_generated import ( + TURN_CONTROLLER_CONTRACT as CONTRACT, project_turn_route, +) from loopx.control_plane.turn_driver.host_failure import build_host_failure_record from loopx.control_plane.turn_driver.transaction import LoopXTurnResultKind from test_loop_turn_loop_controller import ( @@ -25,6 +29,42 @@ ) +@pytest.mark.parametrize("field", ["goal_id", "agent_id", "todo_id"]) +@pytest.mark.parametrize("empty", [[], {}, None, False, 0, ""]) +def test_execution_receipt_rejects_empty_lineage_before_qualification(field, empty): + execution = _execution() + execution["receipt"]["lineage"][field] = empty + with pytest.raises(ValueError, match="missing goal/agent/todo lineage"): + controller.ValidatedTurnReceipt.from_execution(execution) + + +@pytest.mark.parametrize("field", ["goal_id", "agent_id", "todo_id"]) +def test_execution_receipt_retains_nonempty_lineage(field): + execution = _execution() + expected = deepcopy(execution["receipt"]["lineage"]) + expected[field] = "identity-中文" + execution["receipt"]["lineage"] = expected + assert controller.ValidatedTurnReceipt.from_execution(execution).lineage == expected + + +@pytest.mark.parametrize("source,signature", [(1.0, "1.0"), (1e-7, "1e-07"), + (True, "True"), ("sha256:fixture", "sha256:fixture")]) +def test_historical_scalar_route_reaches_public_plan_and_controller(source, signature): + envelope = _envelope(should_run=True) + envelope["action_signature"].update(source_hash=source, envelope_hash=signature) + plan = build_loopx_turn_plan(envelope, host="codex-cli", execution_mode="interactive-visible") + assert plan["route"]["kind"] == "ready_for_host" + assert plan["route"]["would_invoke_host"] is True + decision = controller.decide_loop_disposition(turn_receipt=None, quota_decision=envelope) + assert decision["disposition"] == "run_now" + envelope["action_signature"]["envelope_hash"] = "different-signature" + rejected = build_loopx_turn_plan(envelope, host="codex-cli", execution_mode="interactive-visible") + assert rejected["route"]["kind"] == "contract_error" + assert rejected["route"]["would_invoke_host"] is False + with pytest.raises(ValueError, match="signature hashes"): + controller.decide_loop_disposition(turn_receipt=None, quota_decision=envelope) + + # ready / capability / repair / replan / user / quiet wait / blocked _ROUTES = [ "ready_for_host", @@ -262,20 +302,20 @@ def test_independent_full_decision_cells(row, route, expected): def test_contract_projection_covers_all_routes_with_explicit_rejection(): expected = dict(zip(_ROUTES, _PROJECTED, strict=True)) expected["contract_error"] = None - assert controller._LOOP_CONTROLLER_CONTRACT["route_projection"] == expected + assert CONTRACT["route_projection"] == expected assert set(expected) == {route.value for route in LoopXTurnRoute} for route in LoopXTurnRoute: if route is LoopXTurnRoute.CONTRACT_ERROR: with pytest.raises(ValueError, match="envelope contract"): - controller._route_to_disposition(route) + project_turn_route(route) else: assert ( - controller._route_to_disposition(route).value == expected[route.value] + project_turn_route(route).value == expected[route.value] ) def test_contract_uses_closed_finite_partitions_and_unique_rules(): - contract = controller._LOOP_CONTROLLER_CONTRACT + contract = CONTRACT domains = dict(contract["partitions"]) domains["route"] = [route.value for route in LoopXTurnRoute] domains["receipt_kind"] = ["absent", *[kind.value for kind in LoopXTurnResultKind]] @@ -330,17 +370,7 @@ def test_bundled_contract_is_the_one_consumed_by_controller(): path = ( Path(controller.__file__).parents[1] / "turn_loop_controller_contract_v0.json" ) - assert json.loads(path.read_text()) == controller._LOOP_CONTROLLER_CONTRACT - - -def test_conjunction_does_not_depend_on_json_object_key_order(monkeypatch): - contract = deepcopy(controller._LOOP_CONTROLLER_CONTRACT) - for rule in contract["rules"]: - rule["when"] = dict(reversed(list(rule["when"].items()))) - monkeypatch.setattr(controller, "_LOOP_CONTROLLER_CONTRACT", contract) - for row, outcomes in _ROWS: - for route, expected in zip(_ROUTES, outcomes, strict=True): - test_independent_full_decision_cells(row, route, expected) + assert json.loads(path.read_text()) == CONTRACT # The controller evaluates `rules` in order and stops at the first match, so a @@ -383,60 +413,9 @@ def test_conjunction_does_not_depend_on_json_object_key_order(monkeypatch): def test_contract_rule_sequence_is_pinned(): """A reordered contract is a different program; regeneration must not hide it.""" - rules = controller._LOOP_CONTROLLER_CONTRACT["rules"] + rules = CONTRACT["rules"] assert tuple(rule["id"] for rule in rules) == _RULE_SEQUENCE -def _decisions_for(contract, monkeypatch): - """Every cell's operator-visible decision: the disposition and its reason. - - Two rules can share a disposition and still explain it differently, so the - reason is part of the observable decision, not commentary. - """ - - monkeypatch.setattr(controller, "_LOOP_CONTROLLER_CONTRACT", contract) - observed = {} - for row, outcomes in _ROWS: - for route in _ROUTES: - arguments = _cell(row, route) - try: - payload = controller.decide_loop_disposition(**arguments) - except ValueError: - observed[(row, route)] = ("reject", "reject") - else: - observed[(row, route)] = ( - payload["disposition"], - str(payload["reason"]), - ) - return observed - - -@pytest.mark.parametrize( - "specific, general", - [ - ("progress_exhausted", "progress_route"), - ("host_replan", "failed_receipt"), - ("host_repair", "failed_receipt"), - ("host_retry", "failed_receipt"), - ("host_exhausted", "failed_receipt"), - ("task_validation_replan", "failed_receipt"), - ], -) -def test_a_refining_rule_must_precede_the_rule_it_refines(monkeypatch, specific, general): - """Moving the specific rule behind the general one must change a real decision. - - This is what `_RULE_SEQUENCE` protects: each pinned precedence is load - bearing, so the pin is not a restatement of the shipped file. - """ - - contract = deepcopy(controller._LOOP_CONTROLLER_CONTRACT) - ids = [rule["id"] for rule in contract["rules"]] - assert ids.index(specific) < ids.index(general) - moved = next(rule for rule in contract["rules"] if rule["id"] == specific) - rules = [rule for rule in contract["rules"] if rule["id"] != specific] - rules.insert([rule["id"] for rule in rules].index(general) + 1, moved) - contract["rules"] = rules - - shipped = _decisions_for(deepcopy(controller._LOOP_CONTROLLER_CONTRACT), monkeypatch) - observed = _decisions_for(contract, monkeypatch) - assert observed != shipped +# Rule-mutation sensitivity now runs in turn_loop_controller.test.ts, where the +# production TS evaluator consumes this contract. Python retains every semantic cell. diff --git a/tests/test_loop_turn_loop_controller.py b/tests/test_loop_turn_loop_controller.py index bf600dafe5..cb13529427 100644 --- a/tests/test_loop_turn_loop_controller.py +++ b/tests/test_loop_turn_loop_controller.py @@ -322,6 +322,99 @@ def test_no_receipt_with_delivery_decision_runs_now() -> None: _assert_markers(payload, "run_now") +@pytest.mark.parametrize("qualified", [False, True], ids=["mapping", "producer"]) +@pytest.mark.parametrize("unit,repeats", [("G", 128), ("G", 2_420_000), ("界😀", 180_000)], + ids=["small", "large", "escaped-unicode"]) +def test_controller_preserves_complete_goal_over_the_real_transport( + qualified, unit, repeats, +) -> None: + import json + import tempfile + from pathlib import Path + + from loopx.control_plane.quota.turn_envelope import build_turn_controller_input + from loopx.control_plane.turn_driver import build_loopx_turn_plan + from tests.test_turn_envelope import _full_decision + + source = _full_decision() + snapshot_root = Path(tempfile.gettempdir()) + original_text = unit * repeats + source["interaction_contract"]["agent_channel"]["work_context"] = { + "complete": True, + "sources": [{"source": "goal_state", "content": {"text": original_text}}], + } + current = build_turn_controller_input(source) + envelope = current.envelope() + before = set(snapshot_root.glob("loopx-effect-*")) + assert envelope["work_context"]["sources"][0]["content"]["text"] == original_text + assert envelope["action_signature"]["matches"] is True + assert (len(json.dumps(envelope).encode()) > 2 * 1024 * 1024) == (repeats > 128) + if unit != "G": + assert len(current.envelope_json.encode()) < 2 * 1024 * 1024 + plan = build_loopx_turn_plan(envelope, host="codex-cli", execution_mode="interactive-visible") + assert plan["route"]["kind"] == "ready_for_host" + result = _decide_loop_disposition( + turn_receipt=None, quota_decision=current if qualified else envelope, + ) + _assert_markers(result, "run_now") + assert current.envelope() == envelope + assert set(snapshot_root.glob("loopx-effect-*")) == before + + +def test_large_controller_decision_keeps_signature_rejection() -> None: + envelope = _envelope(should_run=True, signature_matches=False) + envelope["work_context"] = {"text": "G" * 2_420_000} + with pytest.raises(ValueError, match="envelope contract"): + _decide_loop_disposition(turn_receipt=None, quota_decision=envelope) + + +@pytest.mark.parametrize("qualified", [False, True], ids=["mapping", "producer"]) +def test_controller_snapshot_limit_rejects_before_sending(monkeypatch, qualified) -> None: + from loopx.control_plane import effect_runtime + from loopx.control_plane.quota.turn_envelope import build_turn_controller_input + from tests.test_turn_envelope import _full_decision + + effect_runtime.effect_runtime_result("runtime.ping", {}) + # Exercise the actual 64 MiB bound, including JSON overhead, without + # admitting a Host, changing a Goal or dispatching an oversized request. + text = "G" * effect_runtime.MAX_LOCAL_SNAPSHOT_BYTES + with monkeypatch.context() as patch: + patch.setattr(effect_runtime.socket, "create_connection", + lambda *_args, **_kwargs: pytest.fail("oversized request was sent")) + with pytest.raises(effect_runtime.EffectRuntimeRejected) as error: + if qualified: + source = _full_decision() + source["interaction_contract"]["agent_channel"]["work_context"] = { + "complete": True, "sources": [{"source": "goal_state", "content": {"text": text}}], + } + decision = build_turn_controller_input(source) + else: + decision = _envelope(should_run=True) + decision["work_context"] = {"text": text} + _decide_loop_disposition(turn_receipt=None, quota_decision=decision) + assert error.value.diagnostic_code == "request_too_large" + + +@pytest.mark.parametrize("rejection", ["invalid_request", "ambiguous"]) +def test_controller_does_not_retry_other_transport_failures(monkeypatch, rejection) -> None: + from loopx.control_plane import effect_runtime + from loopx.control_plane.turn_driver import loop_controller + + failure = (effect_runtime.EffectRuntimeResponseAmbiguous("turn.controller", timeout=1) + if rejection == "ambiguous" else effect_runtime.EffectRuntimeRejected("rejected")) + calls = [] + + def reject(method, params, **kwargs): + calls.append((method, kwargs)) + raise failure + + monkeypatch.setattr(loop_controller, "effect_runtime_result", reject) + with pytest.raises(type(failure)) as error: + _decide_loop_disposition(turn_receipt=None, quota_decision=_envelope(should_run=True)) + assert error.value is failure + assert calls == [("turn.controller", {})] + + def test_no_receipt_with_quiet_decision_waits_no_spend() -> None: payload = decide_loop_disposition( turn_receipt=None, @@ -1352,3 +1445,41 @@ def test_committed_replay_requires_original_complete_settlement_receipts(): invalid["quota_slot_spend_count"] = 0 with pytest.raises((ValueError, TypeError)): ValidatedTurnReceipt.from_execution(invalid) + + +def test_request_receipt_keeps_its_exact_subject_without_granting_todo_lineage(): + import copy + + from loopx.control_plane.effect_program import SettlementIdentity + + identity = SettlementIdentity(goal_id="request-goal", agent_id="request-agent", todo_id=None, + turn_instance_id="request-turn", request_subject={"request_id": "a" * 64, + "request_revision": 1, "request_digest": "b" * 64, "execution_scope_sha256": "c" * 64}) + execution = _execution() + expected = {"goal_id": "request-goal", "agent_id": "request-agent", "todo_id": "", + "request_binding_id": identity.binding_id} + execution["receipt"]["lineage"] = expected + execution["receipt"]["settlement_effect_id"] = identity.effect_id + for receipt in execution["settlement_result"]["receipts"]: + receipt["effect_id"] = identity.effect_id + + with pytest.raises(ValueError, match="goal/agent/todo lineage"): + ValidatedTurnReceipt.from_execution(execution) + assert ValidatedTurnReceipt.from_execution(execution, request_identity=identity).to_mapping()["lineage"] == expected + for field in ["goal_id", "agent_id", "todo_id", "request_binding_id", "extra"]: + invalid = copy.deepcopy(execution) + invalid["receipt"]["lineage"][field] = "foreign" + with pytest.raises(ValueError, match="original request identity"): + ValidatedTurnReceipt.from_execution(invalid, request_identity=identity) + invalid = copy.deepcopy(execution) + invalid["receipt"]["settlement_effect_id"] = "foreign" + with pytest.raises(ValueError, match="original request identity"): + ValidatedTurnReceipt.from_execution(invalid, request_identity=identity) + invalid = copy.deepcopy(execution) + invalid["settlement_result"]["receipts"].pop() + with pytest.raises((ValueError, TypeError), match="ordered validation/writeback/spend"): + ValidatedTurnReceipt.from_execution(invalid, request_identity=identity) + todo_identity = SettlementIdentity(goal_id="request-goal", agent_id="request-agent", + todo_id="todo-existing", turn_instance_id="request-turn") + with pytest.raises(ValueError, match="original request identity"): + ValidatedTurnReceipt.from_execution(execution, request_identity=todo_identity) diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 21716da259..e3b2fb1667 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1327,7 +1327,7 @@ def test_signed_governance_can_run_while_normal_delivery_remains_withheld(hold): @pytest.mark.parametrize("invalid_signature", [False, True]) -def test_turn_plan_missing_todo_does_not_bypass_contract_errors( +def test_turn_plan_without_todo_requires_signature_and_scoped_binding( invalid_signature: bool, ) -> None: envelope = _envelope( @@ -1340,9 +1340,13 @@ def test_turn_plan_missing_todo_does_not_bypass_contract_errors( envelope, host="generic-cli", execution_mode="isolated-headless", ) - assert payload["ok"] is False - assert payload["route"]["kind"] == "contract_error" + assert payload["ok"] is (not invalid_signature) + assert payload["route"]["kind"] == ("contract_error" if invalid_signature else "blocked") assert payload["route"]["would_invoke_host"] is False + assert payload["route"]["host_invocation_allowed"] is False + assert payload["session"]["action"] == "none" + assert payload["transaction"]["status"] == "not_applicable" + assert not any(payload["effects"].values()) def test_turn_plan_preserves_safe_bypass_when_user_action_is_visible() -> None: diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py index de214470ba..5bd9edf654 100644 --- a/tests/test_loopx_turn_managed_step.py +++ b/tests/test_loopx_turn_managed_step.py @@ -23,8 +23,6 @@ from loopx.control_plane.turn_driver.managed_step import ( LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, decide_managed_step, - managed_step_receipt_from_journal, - reconcile_observed_attempt, ) GOAL_ID = "goal-managed-step" @@ -245,6 +243,47 @@ def test_observation_must_be_a_positive_integer() -> None: _decide(journal, observed_attempt=0) with pytest.raises(ValueError, match="positive integer"): _decide(journal, observed_attempt=True) + with pytest.raises(ValueError, match="positive integer"): + _decide(journal, observed_attempt=1.0) + + +@pytest.mark.parametrize("field", ["host_attempt_count", "failure_attempt"]) +def test_json_transport_preserves_python_integer_qualification(field): + journal = _failed_journal() + if field == "host_attempt_count": + journal[field] = 1.0 + else: + journal["host_failure"]["attempt"] = 1.0 + with pytest.raises(ValueError, match="positive integer|invalid attempt"): + _decide(journal) + + +def test_large_attempts_keep_exact_equality_and_readback(): + attempt = 2**60 + 1 + journal = _failed_journal() + # The native producer computes exponential backoff before capping it. This + # fixture tests the transport's integer identity without that unrelated cost. + journal["host_attempt_count"] = attempt + journal["host_failure"]["attempt"] = attempt + journal["host_failure"]["retry"]["backoff_seconds"] = 300 + assert _decide(journal)["attempt"] == attempt + from loopx.control_plane.turn_driver.loop_controller import ( + ValidatedTurnReceipt, decide_loop_disposition, + ) + + receipt = ValidatedTurnReceipt.from_execution({ + "schema_version": "loopx_turn_execution_v0", "status": "failed", + "result_kind": "host_failure", "receipt": journal["receipt"], + "host_failure": journal["host_failure"], + }) + assert receipt.host_failure["attempt"] == attempt + assert decide_loop_disposition( + turn_receipt=receipt, quota_decision=_envelope(), + predecessor_turn_key=journal["turn_key"], + )["disposition"] == "repair" + journal["host_attempt_count"] = attempt - 1 + with pytest.raises(ValueError, match="host_failure attempt disagrees"): + _decide(journal) def test_journal_attempt_divergence_is_refused() -> None: @@ -359,25 +398,16 @@ def test_managed_step_never_spends_or_writes() -> None: def test_reconcile_accepts_omitted_observations() -> None: journal = _failed_journal(kind="provider_capacity", attempt=1) - assert reconcile_observed_attempt( - journal, observed_attempt=None, observed_max_attempts=None - ) is None + assert _decide(journal, observed_attempt=None, observed_max_attempts=None)["disposition"] == "wait" def test_receipt_from_journal_projects_lineage_and_kind() -> None: journal = _failed_journal(kind="provider_capacity", attempt=1) - receipt = managed_step_receipt_from_journal( - journal, - goal_id=GOAL_ID, - agent_id=AGENT_ID, - turn_key=str(journal["turn_key"]), - ) - - assert receipt.result_kind is LoopXTurnResultKind.HOST_FAILURE - assert receipt.lineage["todo_id"] == TODO_ID - assert receipt.host_failure is not None - assert receipt.host_failure["kind"] == "provider_capacity" + payload = _decide(journal) + assert payload["lineage"]["todo_id"] == TODO_ID + assert payload["retry_continuation"]["attempt"] == 1 + assert "provider_capacity" in payload["reason"] def test_journal_without_stored_plan_is_refused() -> None: @@ -491,3 +521,98 @@ def _record_envelope(decision, *, scheduler_execution_context): "both Turn owners must receive the resolved runtime root, not the raw " f"argument that is None when the registry declares common_runtime_root: {roots}" ) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("lost_spend_response", [False, True]) +def test_native_cli_retry_and_saved_result_keep_one_original_settlement( + tmp_path, monkeypatch, provider, lost_spend_response, +): + """Real CLI, authority and journal with a deterministic native subprocess. + + A capacity refusal may be observed without effects, an explicit same-Turn + retry can finish, and a committed spend with a lost response is read back. + No live model, active Goal or production state participates. + """ + import json + import sys + + from loopx.cli_commands import turn_run_once + from loopx.control_plane.turn_driver import load_turn_journal, turn_journal_path + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + from loopx.todos import list_goal_todos + from tests.capabilities.test_pr_wait_observation import _invoke_turn + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from tests.test_loopx_turn_codex_cli import _fake_codex + 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) + executable, invocations = _fake_codex(tmp_path) + source = executable.read_text().replace( + 'output_path.write_text(json.dumps({', + "pathlib.Path('validated-artifact').write_text('independent marker')\n" + 'output_path.write_text(json.dumps({', + ) + executable.write_text(source) + monkeypatch.setenv("FAKE_CODEX_LOG", str(invocations)) + monkeypatch.setenv("FAKE_CODEX_FAIL", "1") + monkeypatch.setenv("FAKE_CODEX_FAILURE_CATEGORY", "capacity-event") + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + run = [*prefix, "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "codex-cli", "--codex-bin", str(executable), "--project", str(project), + "--validation-command-json", json.dumps([sys.executable, "-c", + "from pathlib import Path; assert Path('validated-artifact').read_text() == 'independent marker'"]), + "--no-global-sync", "--execute"] + code, failed = _invoke_turn([*run, "--todo-id", "todo_fixture0001", "--turn-instance-id", "controller-retry"]) + assert code == 1 and failed["result_kind"] == "host_failure", failed + turn_key = failed["resume_turn_key"] + path = turn_journal_path(runtime, goal_id="loopx-turn-fixture", turn_key=turn_key) + snapshot = path.read_bytes() + events_path = rollout_event_log_path(runtime, "loopx-turn-fixture") + events = load_rollout_events(events_path) + task = list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id="todo_fixture0001") + managed = [*prefix, "turn", "managed-step", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--host", "codex-cli", "--turn-key", turn_key] + code, continuation = _invoke_turn([*managed, "--observed-attempt", "1", "--observed-max-attempts", "3"]) + assert code == 0 and continuation["disposition"] == "wait", continuation + assert continuation["retry_continuation"]["same_turn"] is True + assert continuation["retry_continuation"]["retry_after_seconds"] == 30 + code, refused = _invoke_turn([*managed, "--observed-attempt", "2"]) + assert code == 1 and "observed_attempt disagrees" in refused["error"], refused + assert path.read_bytes() == snapshot + assert load_rollout_events(events_path) == events + assert list_goal_todos(registry_path=registry, goal_id="loopx-turn-fixture", todo_id="todo_fixture0001")["todo"] == task["todo"] + assert len(invocations.read_text().splitlines()) == 1 + + monkeypatch.setenv("FAKE_CODEX_FAIL", "0") + original_spend = turn_run_once.spend_quota_slot + lost = [] + + def lose_response(*args, **kwargs): + result = original_spend(*args, **kwargs) + if lost_spend_response and not lost: + assert result["ok"] is True + lost.append(True) + raise OSError("synthetic response lost after durable spend") + return result + + monkeypatch.setattr(turn_run_once, "spend_quota_slot", lose_response) + code, committed = _invoke_turn([*run, "--resume-turn-key", turn_key, "--retry-failed-turn"]) + if lost_spend_response: + assert code == 1 and "synthetic response lost" in json.dumps(committed), committed + assert lost == [True] + code, committed = _invoke_turn([*run, "--resume-turn-key", turn_key]) + assert code == 0 and committed["status"] == "committed", committed + saved = load_turn_journal(path) + assert saved["turn_key"] == turn_key and saved["host_attempt_count"] == 2 + assert [receipt["step_kind"] for receipt in saved["settlement_result"]["receipts"]] == [ + "validation", "durable_writeback", "quota_spend"] + identity = saved["receipt"]["settlement_effect_id"] + assert all(receipt["effect_id"] == identity for receipt in saved["settlement_result"]["receipts"]) + code, replay = _invoke_turn([*run, "--resume-turn-key", turn_key]) + assert code == 0 and replay["replayed"] is True, replay + assert replay["effects"]["host_invoked"] is False + assert replay["effects"]["state_written"] is False and replay["effects"]["quota_spent"] is False + assert len(invocations.read_text().splitlines()) == 2 + assert sum(event["event_kind"] == "quota_spend" for event in load_rollout_events(events_path)) == 1 diff --git a/tests/test_turn_controller_projection.py b/tests/test_turn_controller_projection.py index 1296b1af8a..2e42fb6ff4 100644 --- a/tests/test_turn_controller_projection.py +++ b/tests/test_turn_controller_projection.py @@ -10,7 +10,7 @@ @pytest.mark.parametrize("user_action_required", [False, True]) -def test_managed_projection_reuses_route_without_trusting_a_mutable_wire( +def test_managed_projection_uses_one_controller_call_without_trusting_a_mutable_wire( monkeypatch, user_action_required, ) -> None: from dataclasses import FrozenInstanceError, replace @@ -57,15 +57,23 @@ def record(operation, **params): 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"]) + controller_calls = [] + original_runtime = loop_controller.effect_runtime_result + + def record_runtime(method, params, **options): + controller_calls.append((method, params["operation"])) + return original_runtime(method, params, **options) + + monkeypatch.setattr(loop_controller, "effect_runtime_result", record_runtime) + actual = decide_managed_step(journal, current, goal_id=GOAL_ID, + agent_id=AGENT_ID, turn_key=journal["turn_key"]) assert actual == expected assert calls == [] + assert controller_calls == [("turn.controller", "managed")] with pytest.raises(ValueError, match="shared envelope contract"): _decide(journal, envelope=changed) - assert calls == ["driver_route"] + assert calls == [] + assert controller_calls == [("turn.controller", "managed")] * 2 @pytest.mark.parametrize("provider", ["file", "sqlite"])