diff --git a/internal/pluginhost/adapters_test.go b/internal/pluginhost/adapters_test.go index 63e30d62cc8..6cc495ec448 100644 --- a/internal/pluginhost/adapters_test.go +++ b/internal/pluginhost/adapters_test.go @@ -2540,6 +2540,101 @@ func TestInterceptorsDoNotMutateInputs(t *testing.T) { }) } +// TestResponseTransformRequestBodiesBySchemaVersion locks the request-body +// delivery rules for the response normalize/translate hooks. +// +// The host invokes those hooks once per SSE frame, so shipping the client and +// upstream request bodies on every frame made streaming cost +// O(frames * request size). Plugins that opt out on schema v10 must therefore +// see empty bodies on streaming calls while still receiving them on +// non-streaming calls, and plugins that do not opt out (or sit below v11) must +// keep the previous behaviour on both paths. +func TestResponseTransformRequestBodiesBySchemaVersion(t *testing.T) { + cases := []struct { + name string + schemaVersion uint32 + omitRequestBody bool + stream bool + wantOriginal string + wantTranslated string + }{ + { + name: "schema 11 opt-out streaming omits request bodies", + schemaVersion: pluginabi.SchemaVersionResponseTransformOmitRequestBody, + omitRequestBody: true, + stream: true, + }, + { + name: "schema 11 opt-out non-streaming keeps request bodies", + schemaVersion: pluginabi.SchemaVersionResponseTransformOmitRequestBody, + omitRequestBody: true, + stream: false, + wantOriginal: "original", + wantTranslated: "translated", + }, + { + name: "schema 11 without opt-out keeps request bodies", + schemaVersion: pluginabi.SchemaVersionResponseTransformOmitRequestBody, + stream: true, + wantOriginal: "original", + wantTranslated: "translated", + }, + { + name: "schema below v11 opt-out keeps request bodies", + schemaVersion: pluginabi.SchemaVersionResponseTransformOmitRequestBody - 1, + omitRequestBody: true, + stream: true, + wantOriginal: "original", + wantTranslated: "translated", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var beforeGot, translatorGot, afterGot pluginapi.ResponseTransformRequest + host := newHostWithRecords(capabilityRecord{ + id: "transform", + plugin: pluginapi.Plugin{ + SchemaVersion: tc.schemaVersion, + Capabilities: pluginapi.Capabilities{ + ResponseTransformOmitRequestBody: tc.omitRequestBody, + ResponseBeforeTranslator: responseNormalizerFunc(func(_ context.Context, req pluginapi.ResponseTransformRequest) (pluginapi.PayloadResponse, error) { + beforeGot = req + return pluginapi.PayloadResponse{Body: req.Body}, nil + }), + ResponseTranslator: responseTranslatorFunc(func(_ context.Context, req pluginapi.ResponseTransformRequest) (pluginapi.PayloadResponse, error) { + translatorGot = req + return pluginapi.PayloadResponse{Body: req.Body}, nil + }), + ResponseAfterTranslator: responseNormalizerFunc(func(_ context.Context, req pluginapi.ResponseTransformRequest) (pluginapi.PayloadResponse, error) { + afterGot = req + return pluginapi.PayloadResponse{Body: req.Body}, nil + }), + }, + }, + }) + assertBodies := func(hook string, got pluginapi.ResponseTransformRequest) { + t.Helper() + if string(got.Body) != "body" { + t.Fatalf("%s did not receive the response payload, body = %q", hook, got.Body) + } + if string(got.OriginalRequest) != tc.wantOriginal || string(got.TranslatedRequest) != tc.wantTranslated { + t.Fatalf("%s OriginalRequest/TranslatedRequest = %q/%q, want %q/%q", hook, got.OriginalRequest, got.TranslatedRequest, tc.wantOriginal, tc.wantTranslated) + } + } + + host.NormalizeResponseBefore(context.Background(), sdktranslator.FormatOpenAI, sdktranslator.FormatClaude, "model", []byte("original"), []byte("translated"), []byte("body"), tc.stream) + assertBodies("ResponseBeforeTranslator", beforeGot) + if _, ok := host.TranslateResponse(context.Background(), sdktranslator.FormatOpenAI, sdktranslator.FormatClaude, "model", []byte("original"), []byte("translated"), []byte("body"), tc.stream); !ok { + t.Fatal("TranslateResponse() = false, want a translated payload") + } + assertBodies("ResponseTranslator", translatorGot) + host.NormalizeResponseAfter(context.Background(), sdktranslator.FormatOpenAI, sdktranslator.FormatClaude, "model", []byte("original"), []byte("translated"), []byte("body"), tc.stream) + assertBodies("ResponseAfterTranslator", afterGot) + }) + } +} + func TestResponseHooksKeepPayloadOrTryNextOnErrorAndEmptyBody(t *testing.T) { normalizerHost := newHostWithRecords( capabilityRecord{ diff --git a/internal/pluginhost/adapters_usage_translation.go b/internal/pluginhost/adapters_usage_translation.go index 77236eb2103..76789b84ba1 100644 --- a/internal/pluginhost/adapters_usage_translation.go +++ b/internal/pluginhost/adapters_usage_translation.go @@ -11,6 +11,7 @@ import ( "github.com/router-for-me/CLIProxyAPI/v8/internal/registry" "github.com/router-for-me/CLIProxyAPI/v8/internal/thinking" coreusage "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/usage" + "github.com/router-for-me/CLIProxyAPI/v8/sdk/pluginabi" "github.com/router-for-me/CLIProxyAPI/v8/sdk/pluginapi" sdktranslator "github.com/router-for-me/CLIProxyAPI/v8/sdk/translator" log "github.com/sirupsen/logrus" @@ -374,6 +375,25 @@ func (h *Host) callRequestTranslator(ctx context.Context, record capabilityRecor return bytes.Clone(resp.Body), true } +// responseTransformRequestBodies returns the request bodies delivered to a +// response normalize/translate hook. +// +// The host calls those hooks once per SSE frame, so delivering the client and +// upstream request bodies on every frame made streaming cost +// O(frames * request size): both multi-megabyte bodies were cloned and Base64 +// encoded into every RPC payload for plugins that only inspect the response. +// Plugins that declare Capabilities.ResponseTransformOmitRequestBody on schema +// version SchemaVersionResponseTransformOmitRequestBody receive neither body on +// streaming calls. Non-streaming calls run once per response and keep carrying +// both, because a converter may legitimately need the request there. +func responseTransformRequestBodies(record capabilityRecord, stream bool, originalRequest, translatedRequest []byte) (originalBody, translatedBody []byte) { + caps := record.plugin.Capabilities + if stream && caps.ResponseTransformOmitRequestBody && record.plugin.SchemaVersion >= pluginabi.SchemaVersionResponseTransformOmitRequestBody { + return nil, nil + } + return bytes.Clone(originalRequest), bytes.Clone(translatedRequest) +} + func (h *Host) callResponseNormalizer(ctx context.Context, record capabilityRecord, method string, normalizer pluginapi.ResponseNormalizer, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) (out []byte, ok bool) { if h == nil || normalizer == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) { return nil, false @@ -385,13 +405,14 @@ func (h *Host) callResponseNormalizer(ctx context.Context, record capabilityReco ok = false } }() + originalBody, translatedBody := responseTransformRequestBodies(record, stream, originalRequestRawJSON, requestRawJSON) resp, errNormalizeResponse := normalizer.NormalizeResponse(ctx, pluginapi.ResponseTransformRequest{ FromFormat: from.String(), ToFormat: to.String(), Model: model, Stream: stream, - OriginalRequest: bytes.Clone(originalRequestRawJSON), - TranslatedRequest: bytes.Clone(requestRawJSON), + OriginalRequest: originalBody, + TranslatedRequest: translatedBody, Body: bytes.Clone(body), }) if errNormalizeResponse != nil || len(resp.Body) == 0 { @@ -411,13 +432,14 @@ func (h *Host) callResponseTranslator(ctx context.Context, record capabilityReco ok = false } }() + originalBody, translatedBody := responseTransformRequestBodies(record, stream, originalRequestRawJSON, requestRawJSON) resp, errTranslateResponse := translator.TranslateResponse(ctx, pluginapi.ResponseTransformRequest{ FromFormat: from.String(), ToFormat: to.String(), Model: model, Stream: stream, - OriginalRequest: bytes.Clone(originalRequestRawJSON), - TranslatedRequest: bytes.Clone(requestRawJSON), + OriginalRequest: originalBody, + TranslatedRequest: translatedBody, Body: bytes.Clone(body), }) if errTranslateResponse != nil || len(resp.Body) == 0 { diff --git a/internal/pluginhost/rpc_client.go b/internal/pluginhost/rpc_client.go index 9916a048c25..619e5f8a0e7 100644 --- a/internal/pluginhost/rpc_client.go +++ b/internal/pluginhost/rpc_client.go @@ -194,6 +194,10 @@ func registerRPCPlugin(ctx context.Context, host *Host, id string, client plugin if resp.Capabilities.RequestLifecyclePlugin { plugin.Capabilities.RequestLifecyclePlugin = adapter } + // One flag covers ResponseTranslator, ResponseBeforeTranslator and + // ResponseAfterTranslator, so it is copied independently of which hooks the + // plugin declares. The schema gate is applied where the bodies are built. + plugin.Capabilities.ResponseTransformOmitRequestBody = resp.Capabilities.ResponseTransformOmitRequestBody if resp.Capabilities.ResponseTranslator { plugin.Capabilities.ResponseTranslator = adapter } diff --git a/internal/pluginhost/rpc_schema.go b/internal/pluginhost/rpc_schema.go index 60c51bd2601..bfa8f913f4c 100644 --- a/internal/pluginhost/rpc_schema.go +++ b/internal/pluginhost/rpc_schema.go @@ -46,6 +46,7 @@ type rpcCapabilities struct { ResponseTranslator bool `json:"response_translator"` ResponseBeforeTranslator bool `json:"response_before_translator"` ResponseAfterTranslator bool `json:"response_after_translator"` + ResponseTransformOmitRequestBody bool `json:"response_transform_omit_request_body,omitempty"` ResponseInterceptor bool `json:"response_interceptor"` StreamChunkInterceptor bool `json:"response_stream_interceptor"` WebSocketResponseObserver bool `json:"websocket_response_observer"` @@ -246,6 +247,7 @@ func rpcCapabilitiesFromPlugin(plugin pluginapi.Plugin) rpcCapabilities { ResponseTranslator: caps.ResponseTranslator != nil, ResponseBeforeTranslator: caps.ResponseBeforeTranslator != nil, ResponseAfterTranslator: caps.ResponseAfterTranslator != nil, + ResponseTransformOmitRequestBody: caps.ResponseTransformOmitRequestBody, ResponseInterceptor: caps.ResponseInterceptor != nil, StreamChunkInterceptor: caps.StreamChunkInterceptor != nil, WebSocketResponseObserver: caps.WebSocketResponseObserver != nil, diff --git a/internal/pluginhost/rpc_schema_test.go b/internal/pluginhost/rpc_schema_test.go index cf1350c793a..e93c47bccaf 100644 --- a/internal/pluginhost/rpc_schema_test.go +++ b/internal/pluginhost/rpc_schema_test.go @@ -371,6 +371,34 @@ func TestRPCCapabilitiesIncludeRequestInterceptorOmitBody(t *testing.T) { } } +func TestRPCCapabilitiesIncludeResponseTransformOmitRequestBody(t *testing.T) { + plugin := pluginapi.Plugin{ + Capabilities: pluginapi.Capabilities{ + ResponseTranslator: responseTranslatorFunc(func(context.Context, pluginapi.ResponseTransformRequest) (pluginapi.PayloadResponse, error) { + return pluginapi.PayloadResponse{}, nil + }), + ResponseTransformOmitRequestBody: true, + }, + } + + caps := rpcCapabilitiesFromPlugin(plugin) + if !caps.ResponseTransformOmitRequestBody { + t.Fatal("ResponseTransformOmitRequestBody = false, want true") + } + + raw, errMarshal := json.Marshal(caps) + if errMarshal != nil { + t.Fatalf("Marshal() error = %v", errMarshal) + } + var decoded map[string]any + if errUnmarshal := json.Unmarshal(raw, &decoded); errUnmarshal != nil { + t.Fatalf("Unmarshal() error = %v", errUnmarshal) + } + if decoded["response_transform_omit_request_body"] != true { + t.Fatalf("response_transform_omit_request_body = %#v, want true", decoded["response_transform_omit_request_body"]) + } +} + func TestRPCRequestInterceptorOmitsBodyOnTheWireForSchema7(t *testing.T) { for _, tc := range []struct { name string diff --git a/sdk/pluginabi/types.go b/sdk/pluginabi/types.go index d1bb79b85c0..9b3c95bc9cf 100644 --- a/sdk/pluginabi/types.go +++ b/sdk/pluginabi/types.go @@ -33,7 +33,12 @@ const ( // publish model catalogs for multiple providers in one response via Registrations. // Version 10 lets a plugin expose multiple OAuth/auth provider identifiers via // auth.identifier (identifiers) or Capabilities.AuthProviders / MultiAuthProvider. - SchemaVersion uint32 = 10 + // Version 11 lets response translators and normalizers opt out of receiving the + // client and upstream request bodies on streaming response transform calls via + // the response_transform_omit_request_body capability. The host invokes those + // hooks once per SSE frame, so shipping both bodies per frame made the cost of + // one stream O(frames * request size). Non-streaming calls still carry them. + SchemaVersion uint32 = 11 // SchemaVersionStreamChunkOmitRequestBody is the first schema version that omits // request bodies on payload stream-chunk interceptor calls. SchemaVersionStreamChunkOmitRequestBody uint32 = 3 @@ -61,6 +66,10 @@ const ( // SchemaVersionMultiAuthProviders is the first schema version where a plugin // may register multiple auth provider identifiers. SchemaVersionMultiAuthProviders uint32 = 10 + // SchemaVersionResponseTransformOmitRequestBody is the first schema version where + // response translators and normalizers may opt out of receiving + // OriginalRequest/TranslatedRequest on streaming response transform calls. + SchemaVersionResponseTransformOmitRequestBody uint32 = 11 ) const ( diff --git a/sdk/pluginabi/types_test.go b/sdk/pluginabi/types_test.go index d79c4c04349..ae7811bed87 100644 --- a/sdk/pluginabi/types_test.go +++ b/sdk/pluginabi/types_test.go @@ -28,8 +28,8 @@ func TestEnvelopeRoundTrip(t *testing.T) { } func TestMethodNamesAreStable(t *testing.T) { - if SchemaVersion != 10 { - t.Fatalf("SchemaVersion = %d, want 10", SchemaVersion) + if SchemaVersion != 11 { + t.Fatalf("SchemaVersion = %d, want 11", SchemaVersion) } if SchemaVersionMultiAuthProviders != 10 { t.Fatalf("SchemaVersionMultiAuthProviders = %d, want 10", SchemaVersionMultiAuthProviders) @@ -58,6 +58,9 @@ func TestMethodNamesAreStable(t *testing.T) { if SchemaVersionMultiProviderModels != 9 { t.Fatalf("SchemaVersionMultiProviderModels = %d, want 9", SchemaVersionMultiProviderModels) } + if SchemaVersionResponseTransformOmitRequestBody != 11 { + t.Fatalf("SchemaVersionResponseTransformOmitRequestBody = %d, want 11", SchemaVersionResponseTransformOmitRequestBody) + } if MethodPluginRegister != "plugin.register" { t.Fatalf("MethodPluginRegister = %q", MethodPluginRegister) } diff --git a/sdk/pluginapi/types.go b/sdk/pluginapi/types.go index fe6d87aa27b..038a56db892 100644 --- a/sdk/pluginapi/types.go +++ b/sdk/pluginapi/types.go @@ -121,6 +121,13 @@ type Capabilities struct { ResponseBeforeTranslator ResponseNormalizer // ResponseAfterTranslator normalizes translated responses before delivery. ResponseAfterTranslator ResponseNormalizer + // ResponseTransformOmitRequestBody opts out of receiving + // ResponseTransformRequest.OriginalRequest and ResponseTransformRequest.TranslatedRequest + // in streaming TranslateResponse and NormalizeResponse calls. It is honored only on + // schema version 11+; the host then skips cloning and Base64-encoding both bodies on + // every SSE frame. Non-streaming calls still carry them because a converter may need + // the request once per response. + ResponseTransformOmitRequestBody bool // RequestInterceptor rewrites execution requests before and after credential selection. RequestInterceptor RequestInterceptor // RequestInterceptorOmitBodyBefore opts out of receiving RequestInterceptRequest.Body