diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 81f0f520d1..11175d2238 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, @@ -118,6 +120,15 @@ def _turn_envelope_result( tokens = source_number_tokens(params) if tokens: request["source_number_tokens"] = tokens + if operation == "compose_complete" and request.get("source_edits") is True and isinstance(payload, Mapping): + # Keep the original source for applying edits. Larger inputs retain + # the existing snapshot route. + raw = _encode_rpc_json(payload) + if 16 * 1024 <= len(raw) <= 1024 * 1024: + data = base64.b64encode(zlib.compress(raw, level=1)).decode("ascii") + if len(data) + 128 < len(raw): + del request["payload"] + request["source_transport"] = {"encoding": "deflate-base64-json-v0", "data": data} 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") diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 649ffaad59..dc017deb95 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)); } +function decodeCompositionSource(request: JsonObject): JsonObject { + const transport = requireJsonObject(request.source_transport, "Turn composition source transport"); + const {encoding, data} = transport; + const maxSourceBytes = 1024 * 1024; + if (Object.hasOwn(request, "payload") || 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; @@ -2388,9 +2408,11 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { return sourceEdits ? contextProjectionChanges(payload, view, precision?.values) : view; } if (operation === "compose_complete") { - const result = composeCompleteTurnDecision(request); + const input = request.source_transport === undefined ? request + : {...request, payload: decodeCompositionSource(request)}; + const result = composeCompleteTurnDecision(input); return request.source_edits === true - ? contextProjectionChanges(requireJsonObject(request.payload, "complete Turn source"), result) : result; + ? contextProjectionChanges(requireJsonObject(input.payload, "complete Turn source"), result) : result; } if (operation === "complete_consumer") { requireCompleteTurnDecision(request.payload, request.source_number_tokens); diff --git a/tests/control_plane/test_complete_numeric_transport.py b/tests/control_plane/test_complete_numeric_transport.py index e0b6048643..497fa0769b 100644 --- a/tests/control_plane/test_complete_numeric_transport.py +++ b/tests/control_plane/test_complete_numeric_transport.py @@ -1,6 +1,8 @@ """Complete consumer costs and the explicit experimental hash transition.""" +import base64 from copy import deepcopy import json +import zlib import pytest @@ -42,6 +44,48 @@ def test_checked_complete_progressive_projection_keeps_sparse_numeric_source(tmp turn.assert_complete_turn_decision(contract) +def test_composed_source_uses_lossless_bounded_transport(tmp_path, monkeypatch): + isolate_sqlite_runtime(tmp_path, monkeypatch) + source = _quota_payload() + source["observations"] = [1.0, -0.0, 2**60 + 1, 1e-7] + source["fixture_material"] = "Full evidence / 完整证据. " * 1400 + original_rpc = turn.effect_runtime_projection_result + requests = [] + + def observe(method, params, **kwargs): + if params["operation"] == "compose_complete": + requests.append((method, params)) + return original_rpc(method, params, **kwargs) + + monkeypatch.setattr(turn, "effect_runtime_projection_result", observe) + complete = turn.compose_turn_decision(source) + (method, packed), = requests + assert "payload" not in packed and len(_json(packed).encode()) < len(_json(source).encode()) + transport = packed["source_transport"] + raw = {key: value for key, value in packed.items() if key != "source_transport"} + assert original_rpc(method, packed) == original_rpc(method, {**raw, "payload": source}) + assert _json(complete["observations"]) == _json(source["observations"]) + assert complete["fixture_material"] == source["fixture_material"] + turn.assert_complete_turn_decision(complete) + + with pytest.raises(RuntimeError, match="invalid Turn composition source transport"): + original_rpc(method, {**packed, "payload": source}) + invalid_utf8 = base64.b64encode(zlib.compress(b'{"body":"\xff"}')).decode("ascii") + oversized = base64.b64encode(zlib.compress(b" " * (1024 * 1024 + 1))).decode("ascii") + for invalid in ("AAAA", invalid_utf8, oversized): + with pytest.raises(RuntimeError, match="valid compressed JSON within 1 MiB"): + original_rpc(method, {**packed, "source_transport": {**transport, "data": invalid}}) + + small_source = {key: source[key] for key in ("ok", "goal_id", "effective_action", "agent_identity", "interaction_contract")} + for original in ({**small_source, "fixture_material": "short text"}, + {**source, "fixture_material": "x" * (1024 * 1024 + 1)}): + requests.clear() + decision = turn.compose_turn_decision(original) + (_, uncompressed), = requests + assert "source_transport" not in uncompressed and uncompressed["payload"] == original + assert decision["fixture_material"] == original["fixture_material"] + + def test_prior_complete_numeric_binding_needs_its_pinned_reader(): source = _quota_payload() source["observations"] = [0.5, 1.0, -0.0, 2**60 + 1, 1e-7]