loopx-agent · Nov 9, 2026 · loopx-agent · Nov 9, 2026
@@ -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")

@@ -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:

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

@@ -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:

@@ -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 {

@@ -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"}};