Compress large complete Turn sources on the local bridge

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")

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);

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]