From 014d01571f1e256706a66163215fb636d12e8fff Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 09:52:51 +0800 Subject: [PATCH] Bound the optional quota context source transport Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/capability_hooks.py | 11 ++++++- loopx/control_plane/quota/turn_envelope.py | 27 ++++++++++++++-- loopx/control_plane/quota/turn_envelope.ts | 31 +++++++++++++++++++ loopx/control_plane/reward_memory.py | 4 +-- .../turn_driver/capability_context.ts | 7 +++-- .../turn_capability_context.test.ts | 24 +++++++++++++- 6 files changed, 95 insertions(+), 9 deletions(-) diff --git a/loopx/control_plane/capability_hooks.py b/loopx/control_plane/capability_hooks.py index 500153872e..d637cc6b53 100644 --- a/loopx/control_plane/capability_hooks.py +++ b/loopx/control_plane/capability_hooks.py @@ -274,12 +274,21 @@ def dispatch_turn_context_hooks( and result.get("instructions") == [] and result.get("result_fields") == {} ) + binding: dict[str, Any] = ( + {"readback_only": True} if readback_only else {"source": source} + ) + if not readback_only and isinstance(source.get("heartbeat_receipt"), Mapping): + from .quota.turn_envelope import pack_turn_source + + transport = pack_turn_source(source) + if transport is not None: + binding = {"source_transport": transport} contribution = effect_runtime_result( "capability_hook.turn_context.validate", { "registration": registration.contract(), "result": result, - **({"readback_only": True} if readback_only else {"source": source}), + **binding, }, ) readback = contribution.pop("readback") diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 81f0f520d1..2e78dd4352 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -2,13 +2,15 @@ from __future__ import annotations +import base64 from collections.abc import Mapping from copy import deepcopy import json from dataclasses import dataclass from typing import Any, cast +import zlib -from ..effect_runtime import effect_runtime_projection_result, source_number_tokens +from ..effect_runtime import _encode_rpc_json, effect_runtime_projection_result, source_number_tokens from ..scheduler.execution_context import ( SchedulerExecutionContextResolution, render_scheduler_execution_args, @@ -93,6 +95,16 @@ def _apply_context_projection_changes( return cast(dict[str, Any], projected) +def pack_turn_source(source: Mapping[str, Any]) -> dict[str, str] | None: + """Use the lossless wire for bounded sources that compress well.""" + raw = _encode_rpc_json(source) + if 16 * 1024 <= len(raw) <= 1024 * 1024: + data = base64.b64encode(zlib.compress(raw, level=1)).decode("ascii") + if len(data) + 128 < len(raw): + return {"encoding": "deflate-base64-json-v0", "data": data} + return None + + def _turn_envelope_result( operation: str, **params: object, @@ -103,7 +115,8 @@ def _turn_envelope_result( payload = params.get("payload") complete = isinstance(payload, Mapping) and payload.get("schema_version") == "loopx_turn_envelope_v1" staged_replan = ( - operation in {"build", "build_controller_input", "quota_signature", "replan_cli_view", "bind_capture"} + operation in {"build", "build_controller_input", "quota_signature", "replan_cli_view", "bind_capture", + "capability_automation_source"} and isinstance(payload, Mapping) and ( (operation == "replan_cli_view" and params.get("source_edits") is True) @@ -118,6 +131,11 @@ def _turn_envelope_result( tokens = source_number_tokens(params) if tokens: request["source_number_tokens"] = tokens + if operation == "capability_automation_source" and set(params) == {"payload"} and isinstance(payload, Mapping): + transport = pack_turn_source(payload) + if transport is not None: + del request["payload"] + request["source_transport"] = transport result = effect_runtime_projection_result("quota.turn_envelope.evaluate", request) if not isinstance(result, Mapping): raise RuntimeError("TypeScript Turn envelope result must be an object") @@ -261,6 +279,11 @@ def quota_action_signature_document(payload: Mapping[str, Any]) -> dict[str, Any ) +def quota_capability_automation_source(payload: Mapping[str, Any]) -> dict[str, Any]: + """Read the quota capability gate without returning the action document.""" + return _turn_envelope_result("capability_automation_source", payload=dict(payload)) + + def turn_envelope_replan_settlement_identity( envelope: Mapping[str, Any], *, turn_instance_id: str | None, ) -> dict[str, Any] | None: diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 649ffaad59..79b3830af9 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -5,6 +5,7 @@ import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated import { EFFECTIVE_ACTIONS, EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; import { AGENT_SCOPE_FRONTIER_ACTIONS } from "../agents/agent_scope_frontier.generated.ts"; import { createHash } from "node:crypto"; +import { inflateSync } from "node:zlib"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; import {omitUnboundExploreResultWriteback, referenceExploreResultWriteback} from "../capabilities/explore_turn_context.ts"; import {prepareContextDelivery, restoreContextDelivery} from "./context_delivery.ts"; @@ -2350,6 +2351,25 @@ function contextPresentationPlan(request: JsonObject): JsonObject { return progressiveContextPlan({payload: context, surface: request.surface}, progressivePreparationScope(context)); } +export function decodeTurnSource(request: JsonObject, inlineField: "payload" | "source"): JsonObject { + const transport = requireJsonObject(request.source_transport, "Turn composition source transport"); + const {encoding, data} = transport; + const maxSourceBytes = 1024 * 1024; + if (Object.hasOwn(request, inlineField) || encoding !== "deflate-base64-json-v0" || + typeof data !== "string" || data.length > maxSourceBytes || + data.length % 4 !== 0 || !/^[A-Za-z0-9+/]+={0,2}$/.test(data)) { + throw new EffectRuntimeRequestError("invalid Turn composition source transport"); + } + let source: unknown; + try { + const decoded = inflateSync(Buffer.from(data, "base64"), {maxOutputLength: maxSourceBytes}); + source = JSON.parse(new TextDecoder("utf-8", {fatal: true, ignoreBOM: true}).decode(decoded)); + } catch { + throw new EffectRuntimeRequestError("Turn composition source must be valid compressed JSON within 1 MiB"); + } + return requireJsonObject(source, "complete Turn decision source"); +} + export function evaluateTurnEnvelope(value: unknown): JsonObject { const request = requireJsonObject(value, "turn envelope request"); const operation = request.operation; @@ -2358,6 +2378,17 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { const {decision, projection} = completeHostSource(request); return {goal_id: decision.goal_id ?? null, ...projection}; } + if (operation === "capability_automation_source") { + if (Object.hasOwn(request, "complete_host_request") || Object.hasOwn(request, "envelope")) { + throw new EffectRuntimeRequestError("invalid Turn composition source transport"); + } + const payload = Object.hasOwn(request, "source_transport") ? decodeTurnSource(request, "payload") + : requireJsonObject(request.payload, "quota payload"); + const projection = turnActionProjection(payload, replanReadPrecision(payload, request.source_number_tokens)); + assertSupportedQuotaPacket(projection); + return {goal_id: payload.goal_id ?? null, agent_id: projection.agent_id ?? null, + boundary: {capabilities: object(projection.boundary).capabilities ?? null}}; + } if (operation === "host_context_binding") { const decision = requireCompleteTurnDecision(request.payload, request.source_number_tokens); const header = requireCompleteHostHeader(request, decision); diff --git a/loopx/control_plane/reward_memory.py b/loopx/control_plane/reward_memory.py index 45eb52f0af..b6fcbf02c4 100644 --- a/loopx/control_plane/reward_memory.py +++ b/loopx/control_plane/reward_memory.py @@ -60,9 +60,9 @@ def reward_memory_automation_enabled( return False # Reuse the typed quota projection rather than interpreting another # copy of the legacy quota boundary in this compatibility adapter. - from .quota.turn_envelope import quota_action_signature_document + from .quota.turn_envelope import quota_capability_automation_source - boundary = quota_action_signature_document(payload).get("boundary") + boundary = quota_capability_automation_source(payload).get("boundary") else: return False try: diff --git a/loopx/control_plane/turn_driver/capability_context.ts b/loopx/control_plane/turn_driver/capability_context.ts index 166475dc8b..5a64f696ff 100644 --- a/loopx/control_plane/turn_driver/capability_context.ts +++ b/loopx/control_plane/turn_driver/capability_context.ts @@ -2,7 +2,7 @@ import type {JsonObject} from "../effect_program.ts"; import {ENVELOPED_SHA256_PATTERN} from "../content_digest.ts"; import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts"; import {requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts"; -import {quotaActionSignatureDocument, turnEnvelopeActionSignatureDocument} from "../quota/turn_envelope.ts"; +import {decodeTurnSource, quotaActionSignatureDocument, turnEnvelopeActionSignatureDocument} from "../quota/turn_envelope.ts"; // This is a local host-IO contract, not a new capability or authority vocabulary. // Capability providers own enablement. These rules only bound their contributions. @@ -86,13 +86,14 @@ export function validateTurnContextContribution(input: JsonObject): JsonObject { if (input.readback_only !== undefined && requireBoolean(input.readback_only, "readback_only")) { // A status-only result is returned to the guard and is never a reusable // context contribution. Keep its registration, slot and budget checks. - if (Object.hasOwn(input, "source") || Object.keys(context).length || + if (Object.hasOwn(input, "source") || Object.hasOwn(input, "source_transport") || Object.keys(context).length || instructions.length || Object.keys(resultFields).length) { throw new Error("readback-only context validation cannot carry a source or a contribution"); } return normalized; } - return {...normalized, basis_digest: canonicalAuthoritySha256({source: contextSource(input.source), ...normalized})}; + const source = Object.hasOwn(input, "source_transport") ? decodeTurnSource(input, "source") : input.source; + return {...normalized, basis_digest: canonicalAuthoritySha256({source: contextSource(source), ...normalized})}; } export function projectTurnContextContribution(input: JsonObject): JsonObject { diff --git a/tests/control_plane_ts/turn_capability_context.test.ts b/tests/control_plane_ts/turn_capability_context.test.ts index 11807cfb10..9704f8f3ec 100644 --- a/tests/control_plane_ts/turn_capability_context.test.ts +++ b/tests/control_plane_ts/turn_capability_context.test.ts @@ -1,7 +1,8 @@ import assert from "node:assert/strict"; import test from "node:test"; +import {deflateSync} from "node:zlib"; import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts"; -import {quotaActionSignatureDocument} from "../../loopx/control_plane/quota/turn_envelope.ts"; +import {evaluateTurnEnvelope, quotaActionSignatureDocument} from "../../loopx/control_plane/quota/turn_envelope.ts"; import {validateTurnContextRegistration, validateTurnContextContribution, projectTurnContextContribution} from "../../loopx/control_plane/turn_driver/capability_context.ts"; @@ -45,6 +46,27 @@ const quotaSource = {goal_id: "fixture-goal", agent_id: "fixture-agent", enabled: true, configured_for_agent: true, experiment_available: true, automatic_recall: true, automatic_ingest: false}}}}; +test("the shared quota source transport preserves capability gates and contribution binding", () => { + const source = {...quotaSource, goal_boundary: {...quotaSource.goal_boundary, stop_condition: "保留原停止条件"}}; + const source_transport = {encoding: "deflate-base64-json-v0", + data: deflateSync(Buffer.from(JSON.stringify(source))).toString("base64")}; + const operation = "capability_automation_source"; + const inline = evaluateTurnEnvelope({operation, payload: source}); + assert.deepEqual(evaluateTurnEnvelope({operation, source_transport}), inline); + const contribution = validateTurnContextContribution({registration, result, source}); + assert.deepEqual(validateTurnContextContribution({registration, result, source_transport}), contribution); + assert.throws(() => projectTurnContextContribution({contribution, readback: result.readback, + source: {...source, goal_boundary: {...source.goal_boundary, stop_condition: "改动停止条件"}}})); + assert.throws(() => evaluateTurnEnvelope({operation, payload: source, source_transport}), /invalid Turn/); + assert.throws(() => evaluateTurnEnvelope({operation, envelope: {}, source_transport}), /invalid Turn/); + assert.throws(() => validateTurnContextContribution({registration, result, source, source_transport}), /invalid Turn/); + const status = {...result, context_fields: {}, instructions: [], result_fields: {}}; + assert.throws(() => validateTurnContextContribution({registration, result: status, readback_only: true, + source_transport}), /readback-only/); + assert.throws(() => validateTurnContextContribution({registration, result, source_transport: { + ...source_transport, data: "AAAA"}}), /valid compressed JSON/); +}); + test("an existing quota contribution can replay with its exact legacy reader binding", () => { const source = {...quotaSource, replan_read_path: "fixture/decision.json", replan_action_packet: {obligation_id: "fixture-replan", decision: "replan_required"}};