1195 lines
48 KiB
Go
1195 lines
48 KiB
Go
package openai
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"iop/packages/go/config"
|
|
"iop/packages/go/streamgate"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
func TestRepeatGuardContinuationTemperatureCandidates(t *testing.T) {
|
|
for attempt, want := range []float64{0.2, 0.4, 0.6} {
|
|
got := continuationTemperature(attempt+1, nil)
|
|
if got == nil || *got != want {
|
|
t.Fatalf("attempt %d temperature = %v, want %.1f", attempt+1, got, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRepeatGuardPreservesCallerTemperature(t *testing.T) {
|
|
callerTemperature := 1.3
|
|
for attempt := 1; attempt <= 3; attempt++ {
|
|
got := continuationTemperature(attempt, &callerTemperature)
|
|
if got == nil || *got != callerTemperature {
|
|
t.Fatalf("attempt %d temperature = %v, want caller value %.1f", attempt, got, callerTemperature)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestStreamGateChatConfiguredOutputFiltersCleanStreamSingleTerminal runs a
|
|
// clean chat stream through the production registry built from a configured
|
|
// output-filter policy (repeat rolling text + terminal-gated repeat action
|
|
// sibling + schema terminal-gate + provider-error none). It proves S14's
|
|
// pipeline contract end to end: nothing is committed before the all-complete
|
|
// outcome set (single WriteHeader), the terminal-gate participants release a
|
|
// clean stream once, and no filter fires a recovery (single [DONE], zero
|
|
// re-dispatch).
|
|
func TestStreamGateChatConfiguredOutputFiltersCleanStreamSingleTerminal(t *testing.T) {
|
|
srv, dc, initial, fake := buildStreamGateChatFixture(t, true, 3)
|
|
initial.events = bufferedRunEvents(
|
|
&iop.RunEvent{Type: "delta", Delta: "hello world"},
|
|
&iop.RunEvent{Type: "complete"},
|
|
)
|
|
|
|
gateCfg := outputFilterGateCfg(config.StreamGateFilterEnforcementBlocking)
|
|
snapRef, err := dc.ingress.recoveryRef()
|
|
if err != nil {
|
|
t.Fatalf("recoveryRef: %v", err)
|
|
}
|
|
// hasScheme=true registers all three filters (including the blocking
|
|
// terminal-gate schema participant).
|
|
fctx := openAIOutputFilterContext{endpoint: openAIRebuildEndpointChat, hasScheme: true, requestRef: snapRef.SnapshotRef()}
|
|
registry, err := openAIStreamGateRegistrySnapshotFor(gateCfg, fctx)
|
|
if err != nil {
|
|
t.Fatalf("openAIStreamGateRegistrySnapshotFor: %v", err)
|
|
}
|
|
|
|
w := newRecordingResponseWriter()
|
|
rt, _, err := srv.buildOpenAIChatStreamGateRuntime(dc, initial, newOpenAIChatSSEReleaseSink(w, w, "chatcmpl-run-attempt-1", time.Now().Unix(), "client-model"), registry)
|
|
if err != nil {
|
|
t.Fatalf("buildOpenAIChatStreamGateRuntime: %v", err)
|
|
}
|
|
if err := rt.Run(context.Background()); err != nil {
|
|
t.Fatalf("rt.Run: %v", err)
|
|
}
|
|
|
|
if got := len(fake.reqsSnapshot()); got != 0 {
|
|
t.Fatalf("clean stream must not dispatch a recovery: got %d dispatches", got)
|
|
}
|
|
if got := w.headerCallCount(); got != 1 {
|
|
t.Fatalf("WriteHeader call count: got %d, want 1 (no eager commit before all-complete)", got)
|
|
}
|
|
chunks := parseSSEChatChunks(t, w.body.String())
|
|
if got := joinedContent(chunks); got != "hello world" {
|
|
t.Fatalf("joined content: got %q, want %q", got, "hello world")
|
|
}
|
|
if n := strings.Count(w.body.String(), "data: [DONE]"); n != 1 {
|
|
t.Fatalf("[DONE] sentinel count: got %d, want 1", n)
|
|
}
|
|
roleCount := 0
|
|
for _, c := range chunks {
|
|
if len(c.Choices) > 0 && c.Choices[0].Delta.Role == "assistant" {
|
|
roleCount++
|
|
}
|
|
}
|
|
if roleCount != 1 {
|
|
t.Fatalf("assistant role chunk count: got %d, want 1 (single opening)", roleCount)
|
|
}
|
|
}
|
|
|
|
func TestStreamGateEndpointPathMatrix(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
endpoint string
|
|
path string
|
|
wantKind streamgate.EventKind
|
|
}{
|
|
{name: "chat normalized", endpoint: openAIRebuildEndpointChat, path: "normalized", wantKind: streamgate.EventKindTextDelta},
|
|
{name: "responses normalized", endpoint: openAIRebuildEndpointResponses, path: "normalized", wantKind: streamgate.EventKindTextDelta},
|
|
{name: "chat tunnel", endpoint: openAIRebuildEndpointChat, path: "tunnel", wantKind: streamgate.EventKindTextDelta},
|
|
{name: "responses tunnel", endpoint: openAIRebuildEndpointResponses, path: "tunnel", wantKind: streamgate.EventKindTextDelta},
|
|
}
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
if tc.path == "tunnel" {
|
|
state := &openAITunnelCodecState{}
|
|
codec := newOpenAITunnelEndpointCodec(tc.endpoint, state)
|
|
frame := []byte("data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n")
|
|
if tc.endpoint == openAIRebuildEndpointResponses {
|
|
frame = []byte("data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\n")
|
|
}
|
|
events, err := codec.decode(frame, false)
|
|
if err != nil {
|
|
t.Fatalf("decode: %v", err)
|
|
}
|
|
if len(events) != 1 || events[0].Kind() != tc.wantKind {
|
|
t.Fatalf("events=%v", eventKinds(events))
|
|
}
|
|
wire, ok := state.popRelease()
|
|
if !ok || string(wire) != string(frame) {
|
|
t.Fatalf("wire=%q, want exact frame", wire)
|
|
}
|
|
return
|
|
}
|
|
events := bufferedRunEvents(&iop.RunEvent{Type: "delta", Delta: "hello"}, &iop.RunEvent{Type: "complete"})
|
|
handle := &fakeRunResult{dispatch: edgeservice.RunDispatch{RunID: "run-matrix", Target: "served", Adapter: "test"}, events: events}
|
|
var source streamgate.NormalizedEventSource
|
|
if tc.endpoint == openAIRebuildEndpointResponses {
|
|
dc := &responsesDispatchContext{req: responsesRequest{Model: "alias"}}
|
|
source = newOpenAIResponsesEventSource(dc, handle, &openAIResponsesResultHolder{}, &openAIStreamGateUsageHolder{})
|
|
} else {
|
|
source = newOpenAIRunEventSource(handle.Stream(), handle.WaitTimeout(), &openAIStreamGateUsageHolder{})
|
|
}
|
|
if start, err := source.NextEvent(t.Context()); err != nil || start.Kind() != streamgate.EventKindResponseStart {
|
|
t.Fatalf("start=(%v,%v)", start.Kind(), err)
|
|
}
|
|
if event, err := source.NextEvent(t.Context()); err != nil || event.Kind() != tc.wantKind {
|
|
t.Fatalf("content=(%v,%v)", event.Kind(), err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func eventKinds(events []streamgate.NormalizedEvent) []streamgate.EventKind {
|
|
out := make([]streamgate.EventKind, len(events))
|
|
for i := range events {
|
|
out[i] = events[i].Kind()
|
|
}
|
|
return out
|
|
}
|
|
|
|
func TestResponsesStreamGateEventShapeAndPathSwitch(t *testing.T) {
|
|
state := &openAITunnelCodecState{}
|
|
codec := newOpenAITunnelEndpointCodec(openAIRebuildEndpointResponses, state)
|
|
frames := [][]byte{
|
|
[]byte("data: {\"type\":\"response.output_text.delta\",\"delta\":\"answer\"}\n\n"),
|
|
[]byte("data: {\"type\":\"response.reasoning_text.delta\",\"delta\":\"think\"}\n\n"),
|
|
[]byte("data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc-1\",\"name\":\"lookup\",\"delta\":\"{}\"}\n\n"),
|
|
[]byte("data: {\"type\":\"response.completed\"}\n\n"),
|
|
}
|
|
var events []streamgate.NormalizedEvent
|
|
for _, frame := range frames {
|
|
decoded, err := codec.decode(frame, false)
|
|
if err != nil {
|
|
t.Fatalf("decode: %v", err)
|
|
}
|
|
events = append(events, decoded...)
|
|
}
|
|
decoded, err := codec.finishTransport(nil, false)
|
|
if err != nil {
|
|
t.Fatalf("finishTransport: %v", err)
|
|
}
|
|
events = append(events, decoded...)
|
|
wantKinds := []streamgate.EventKind{streamgate.EventKindTextDelta, streamgate.EventKindReasoningDelta, streamgate.EventKindToolCallFragment, streamgate.EventKindTerminal}
|
|
if got := eventKinds(events); len(got) != len(wantKinds) {
|
|
t.Fatalf("kinds=%v", got)
|
|
} else {
|
|
for i := range got {
|
|
if got[i] != wantKinds[i] {
|
|
t.Fatalf("kinds=%v, want=%v", got, wantKinds)
|
|
}
|
|
}
|
|
}
|
|
|
|
w := newRecordingResponseWriter()
|
|
selector := newOpenAIStreamGateCodecSelector(openAIStreamGateCodecNormalized)
|
|
normalized := newOpenAIResponsesReleaseSink(&Server{}, w, &responsesDispatchContext{}, &openAIResponsesResultHolder{})
|
|
tunnel := newOpenAITunnelReleaseSink(w, w)
|
|
// Move the exact decoded wire queues to the tunnel sink and switch before
|
|
// the first commit. The composite must freeze tunnel framing exactly once.
|
|
tunnel.codec = state
|
|
selector.set(openAIStreamGateCodecTunnel)
|
|
composite := newOpenAICompositeReleaseSink(selector, normalized, tunnel)
|
|
start, _ := streamgate.NewResponseStartEvent(streamGateChannelDefault, http.StatusOK, map[string]string{"Content-Type": "text/event-stream"}, time.Now())
|
|
responseStart, _ := start.AsResponseStart()
|
|
if _, err := composite.CommitResponseStart(t.Context(), responseStart); err != nil {
|
|
t.Fatalf("CommitResponseStart: %v", err)
|
|
}
|
|
for _, event := range events[:3] {
|
|
var release streamgate.ReleaseEvent
|
|
var err error
|
|
switch event.Kind() {
|
|
case streamgate.EventKindTextDelta:
|
|
value, _ := event.AsTextDelta()
|
|
release, err = streamgate.NewReleaseTextDeltaEvent(event.Channel(), value, event.Timestamp())
|
|
case streamgate.EventKindReasoningDelta:
|
|
value, _ := event.AsReasoningDelta()
|
|
release, err = streamgate.NewReleaseReasoningDeltaEvent(event.Channel(), value, event.Timestamp())
|
|
case streamgate.EventKindToolCallFragment:
|
|
value, _ := event.AsToolCallFragment()
|
|
release, err = streamgate.NewReleaseToolCallFragmentEvent(event.Channel(), value.ID, value.Name, value.Arguments, event.Timestamp())
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("release event: %v", err)
|
|
}
|
|
if _, err := composite.Release(t.Context(), release); err != nil {
|
|
t.Fatalf("Release: %v", err)
|
|
}
|
|
}
|
|
terminal, _ := events[3].AsTerminal()
|
|
if _, err := composite.CommitTerminal(t.Context(), terminal); err != nil {
|
|
t.Fatalf("CommitTerminal: %v", err)
|
|
}
|
|
var want strings.Builder
|
|
for _, frame := range frames {
|
|
want.Write(frame)
|
|
}
|
|
if w.body.String() != want.String() {
|
|
t.Fatalf("released wire=%q, want=%q", w.body.String(), want.String())
|
|
}
|
|
if composite.resolvedCodec() != openAIStreamGateCodecTunnel {
|
|
t.Fatalf("resolved codec=%q", composite.resolvedCodec())
|
|
}
|
|
|
|
t.Run("production normalized to tunnel recovery", func(t *testing.T) {
|
|
providerBody := []byte(`{"id":"resp-recovered","object":"response","model":"served-b","output_text":"recovered","output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":"recovered"}]}]}`)
|
|
service := newScriptedPoolRunService(
|
|
scriptedPoolAttempt{
|
|
path: string(edgeservice.ProviderPoolPathNormalized), runID: "responses-path-1",
|
|
provider: "prov-a", target: "served-a",
|
|
runEvents: bufferedRunEvents(
|
|
&iop.RunEvent{Type: "delta", Delta: "discarded"},
|
|
&iop.RunEvent{Type: "complete"},
|
|
),
|
|
},
|
|
scriptedPoolAttempt{
|
|
path: string(edgeservice.ProviderPoolPathTunnel), runID: "responses-path-2",
|
|
provider: "prov-b", target: "served-b",
|
|
frames: bufferedTunnelFrames(
|
|
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "application/json"}},
|
|
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: providerBody},
|
|
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
|
|
),
|
|
},
|
|
)
|
|
fault := 3
|
|
srv := NewServer(config.EdgeOpenAIConf{
|
|
TimeoutSec: 15,
|
|
StreamEvidenceGate: config.StreamEvidenceGateConf{
|
|
Enabled: true, MaxRequestFaultRecovery: &fault,
|
|
},
|
|
}, service, nil)
|
|
rawBody := []byte(`{"model":"client-model","input":"hi","stream":false}`)
|
|
route := routeDispatch{ProviderPool: true, TimeoutSec: 15}
|
|
base := newTestRequestContext(t, route, rawBody)
|
|
base.endpoint = usageEndpointResponses
|
|
requestCtx := &responsesRequestContext{
|
|
openAIRequestContext: base,
|
|
envelope: responsesEnvelope{Model: "client-model"},
|
|
}
|
|
var req responsesRequest
|
|
if err := json.Unmarshal(rawBody, &req); err != nil {
|
|
t.Fatalf("decode request: %v", err)
|
|
}
|
|
dc, err := srv.newResponsesDispatchContext(requestCtx, req)
|
|
if err != nil {
|
|
t.Fatalf("newResponsesDispatchContext: %v", err)
|
|
}
|
|
poolReq := edgeservice.ProviderPoolDispatchRequest{
|
|
Run: dc.submitReq,
|
|
Tunnel: edgeservice.SubmitProviderTunnelRequest{
|
|
ModelGroupKey: "client-model", Method: http.MethodPost, Path: "/v1/responses",
|
|
TimeoutSec: 15, Metadata: dc.runMetadata, ProviderPool: true,
|
|
},
|
|
PrepareTunnel: func(req edgeservice.SubmitProviderTunnelRequest) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
return req, nil
|
|
},
|
|
PrepareRun: func(req edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) { return req, nil },
|
|
}
|
|
poolReq.Tunnel.BuildBody = func(target string) ([]byte, error) { return rewriteResponsesModel(rawBody, target) }
|
|
dc = dc.withPoolDispatch(poolReq)
|
|
initial, err := service.SubmitProviderPool(t.Context(), poolReq)
|
|
if err != nil || initial.Run == nil {
|
|
t.Fatalf("initial pool admission=(%v,%v)", initial, err)
|
|
}
|
|
|
|
snapRef, err := dc.ingress.recoveryRef()
|
|
if err != nil {
|
|
t.Fatalf("recoveryRef: %v", err)
|
|
}
|
|
violation := newInjectedViolationFilter(t, "test.responses_path_switch", 1, snapRef.SnapshotRef())
|
|
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
|
|
if err != nil {
|
|
t.Fatalf("NewFilterRegistration: %v", err)
|
|
}
|
|
productionWriter := newRecordingResponseWriter()
|
|
holder := &openAIResponsesResultHolder{}
|
|
productionNormalized := newOpenAIResponsesReleaseSink(srv, productionWriter, dc, holder)
|
|
productionSelector := newOpenAIStreamGateCodecSelector(openAIStreamGateCodecNormalized)
|
|
productionTunnel := newOpenAIBufferedTunnelReleaseSink(productionWriter, nil, "")
|
|
productionSink := newOpenAICompositeReleaseSink(productionSelector, productionNormalized, productionTunnel)
|
|
runtime, _, err := srv.buildOpenAIResponsesStreamGateRuntime(dc, initial.Run, productionSink, streamGateTestRegistry(t, reg))
|
|
if err != nil {
|
|
t.Fatalf("buildOpenAIResponsesStreamGateRuntime: %v", err)
|
|
}
|
|
runErr := runtime.Run(t.Context())
|
|
_ = runtime.CloseRequestResources(t.Context(), runErr == nil)
|
|
if runErr != nil {
|
|
t.Fatalf("runtime.Run: %v", runErr)
|
|
}
|
|
if productionWriter.code != http.StatusOK || productionWriter.headerCallCount() != 1 {
|
|
t.Fatalf("response start=(status=%d headers=%d), want one 200", productionWriter.code, productionWriter.headerCallCount())
|
|
}
|
|
if got := productionWriter.body.Bytes(); string(got) != string(providerBody) {
|
|
t.Fatalf("released body=%q, want exact recovered Responses body=%q", got, providerBody)
|
|
}
|
|
if strings.Contains(productionWriter.body.String(), "discarded") || productionSink.resolvedCodec() != openAIStreamGateCodecTunnel {
|
|
t.Fatalf("path switch leaked initial output or wrong codec: codec=%q body=%q", productionSink.resolvedCodec(), productionWriter.body.String())
|
|
}
|
|
if service.poolSubmits() != 2 {
|
|
t.Fatalf("pool admissions=%d, want initial + one recovery", service.poolSubmits())
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestTunnelSchemaContextPreserved(t *testing.T) {
|
|
srv, dc, _, _ := buildStreamGateChatFixture(t, true, 1)
|
|
dc.req.Metadata = json.RawMessage(`{"scheme":{"type":"object"}}`)
|
|
req := srv.openAIChatTunnelStreamGateRequest(dc)
|
|
fctx, err := srv.openAITunnelOutputFilterContext(req)
|
|
if err != nil {
|
|
t.Fatalf("openAITunnelOutputFilterContext: %v", err)
|
|
}
|
|
if !fctx.hasScheme {
|
|
t.Fatal("tunnel context dropped metadata.scheme")
|
|
}
|
|
gateCfg := config.StreamEvidenceGateConf{Filters: []config.StreamGateFilterPolicyConf{{Filter: config.StreamGateFilterSchemaGate}}}
|
|
registry, err := openAIStreamGateRegistrySnapshotFor(gateCfg, fctx)
|
|
if err != nil {
|
|
t.Fatalf("registry: %v", err)
|
|
}
|
|
reqCtx, _ := openAIOutputFilterRequestContext(fctx)
|
|
reqSnap, err := registry.BeginRequest(reqCtx)
|
|
if err != nil {
|
|
t.Fatalf("BeginRequest: %v", err)
|
|
}
|
|
target, _ := streamgate.NewAttemptTarget(fctx.modelGroup, "served", "prov", string(edgeservice.ProviderPoolPathTunnel), []string{"output.schema_gate", streamGateNoopCapability})
|
|
resolved, err := reqSnap.ResolveAttempt(target)
|
|
if err != nil {
|
|
t.Fatalf("ResolveAttempt: %v", err)
|
|
}
|
|
if !resolvedFilterIDs(resolved)[openAISchemaGateFilterID] {
|
|
t.Fatal("schema gate not resolved for tunnel path")
|
|
}
|
|
}
|
|
|
|
// TestOpenAITunnelCodecTerminalWire is the S14/S18 production-codec fixture:
|
|
// protocol finish remains releaseable wire, while [DONE] or END emits exactly
|
|
// one Core terminal and preserves every provider frame byte-for-byte.
|
|
func TestOpenAITunnelCodecTerminalWire(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
endpoint string
|
|
frames [][]byte
|
|
}{
|
|
{
|
|
name: "chat finish then done",
|
|
endpoint: openAIRebuildEndpointChat,
|
|
frames: [][]byte{
|
|
[]byte("data: {\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n"),
|
|
[]byte("data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n"),
|
|
[]byte("data: [DONE]\n\n"),
|
|
},
|
|
},
|
|
{
|
|
name: "responses completed then done",
|
|
endpoint: openAIRebuildEndpointResponses,
|
|
frames: [][]byte{
|
|
[]byte("data: {\"type\":\"response.output_text.delta\",\"delta\":\"answer\"}\n\n"),
|
|
[]byte("data: {\"type\":\"response.completed\"}\n\n"),
|
|
[]byte("data: [DONE]\n\n"),
|
|
},
|
|
},
|
|
{
|
|
name: "transport end only",
|
|
endpoint: openAIRebuildEndpointChat,
|
|
frames: [][]byte{
|
|
[]byte("data: {\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n"),
|
|
},
|
|
},
|
|
}
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
state := &openAITunnelCodecState{}
|
|
codec := newOpenAITunnelEndpointCodec(tc.endpoint, state)
|
|
var events []streamgate.NormalizedEvent
|
|
for _, frame := range tc.frames {
|
|
decoded, err := codec.decode(frame, false)
|
|
if err != nil {
|
|
t.Fatalf("decode: %v", err)
|
|
}
|
|
events = append(events, decoded...)
|
|
}
|
|
decoded, err := codec.finishTransport(nil, false)
|
|
if err != nil {
|
|
t.Fatalf("finishTransport: %v", err)
|
|
}
|
|
events = append(events, decoded...)
|
|
terminals := 0
|
|
for _, event := range events {
|
|
if event.Kind() == streamgate.EventKindTerminal {
|
|
terminals++
|
|
}
|
|
}
|
|
if terminals != 1 {
|
|
t.Fatalf("terminal event count=%d, want 1", terminals)
|
|
}
|
|
got := releaseTunnelCodecEvents(t, state, events)
|
|
want := strings.Join(byteFramesToStrings(tc.frames), "")
|
|
if got != want {
|
|
t.Fatalf("released wire=%q, want=%q", got, want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestOpenAITunnelCodecSemanticFrames proves metadata is wire-only evidence and
|
|
// split Chat/Responses function calls retain the first frame's stable identity.
|
|
func TestOpenAITunnelCodecSemanticFrames(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
endpoint string
|
|
frames [][]byte
|
|
wantID string
|
|
wantName string
|
|
}{
|
|
{
|
|
name: "chat split tool identity",
|
|
endpoint: openAIRebuildEndpointChat,
|
|
frames: [][]byte{
|
|
[]byte("data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-chat\",\"function\":{\"name\":\"lookup\"}}]}}]}\n\n"),
|
|
[]byte("data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{ }\"}}]}}]}\n\n"),
|
|
[]byte("data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n"),
|
|
[]byte("data: [DONE]\n\n"),
|
|
},
|
|
wantID: "call-chat", wantName: "lookup",
|
|
},
|
|
{
|
|
name: "responses split tool identity",
|
|
endpoint: openAIRebuildEndpointResponses,
|
|
frames: [][]byte{
|
|
[]byte("data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc-1\",\"call_id\":\"call-responses\",\"name\":\"lookup\"}}\n\n"),
|
|
[]byte("data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc-1\",\"output_index\":0,\"delta\":\"{ }\"}\n\n"),
|
|
[]byte("data: {\"type\":\"response.completed\"}\n\n"),
|
|
[]byte("data: [DONE]\n\n"),
|
|
},
|
|
wantID: "call-responses", wantName: "lookup",
|
|
},
|
|
}
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
state := &openAITunnelCodecState{}
|
|
codec := newOpenAITunnelEndpointCodec(tc.endpoint, state)
|
|
var events []streamgate.NormalizedEvent
|
|
for _, frame := range tc.frames {
|
|
decoded, err := codec.decode(frame, false)
|
|
if err != nil {
|
|
t.Fatalf("decode: %v", err)
|
|
}
|
|
events = append(events, decoded...)
|
|
}
|
|
toolCount := 0
|
|
for _, event := range events {
|
|
if event.Kind() != streamgate.EventKindToolCallFragment {
|
|
continue
|
|
}
|
|
toolCount++
|
|
fragment, err := event.AsToolCallFragment()
|
|
if err != nil {
|
|
t.Fatalf("AsToolCallFragment: %v", err)
|
|
}
|
|
if fragment.ID != tc.wantID || fragment.Name != tc.wantName || fragment.Arguments != "{ }" {
|
|
t.Fatalf("tool fragment=%+v", fragment)
|
|
}
|
|
}
|
|
if toolCount != 1 {
|
|
t.Fatalf("tool fragment count=%d, want 1", toolCount)
|
|
}
|
|
got := releaseTunnelCodecEvents(t, state, append(events, mustFinishTunnelCodec(t, codec)...))
|
|
want := strings.Join(byteFramesToStrings(tc.frames), "")
|
|
if got != want {
|
|
t.Fatalf("released wire=%q, want=%q", got, want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
type synchronizedTunnelLifecycleWriter struct {
|
|
mu sync.Mutex
|
|
header http.Header
|
|
body []byte
|
|
status int
|
|
headerCalls int
|
|
}
|
|
|
|
func newSynchronizedTunnelLifecycleWriter() *synchronizedTunnelLifecycleWriter {
|
|
return &synchronizedTunnelLifecycleWriter{header: make(http.Header)}
|
|
}
|
|
|
|
func (w *synchronizedTunnelLifecycleWriter) Header() http.Header {
|
|
return w.header
|
|
}
|
|
|
|
func (w *synchronizedTunnelLifecycleWriter) WriteHeader(status int) {
|
|
w.mu.Lock()
|
|
w.status = status
|
|
w.headerCalls++
|
|
w.mu.Unlock()
|
|
}
|
|
|
|
func (w *synchronizedTunnelLifecycleWriter) Write(payload []byte) (int, error) {
|
|
w.mu.Lock()
|
|
w.body = append(w.body, payload...)
|
|
w.mu.Unlock()
|
|
return len(payload), nil
|
|
}
|
|
|
|
func (w *synchronizedTunnelLifecycleWriter) Flush() {}
|
|
|
|
func (w *synchronizedTunnelLifecycleWriter) snapshot() (int, int, []byte) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
return w.status, w.headerCalls, append([]byte(nil), w.body...)
|
|
}
|
|
|
|
func waitForTunnelCodecReleases(t *testing.T, state *openAITunnelCodecState, want int) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for {
|
|
state.mu.Lock()
|
|
got := len(state.releases)
|
|
state.mu.Unlock()
|
|
if got >= want {
|
|
return
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("tunnel codec releases=%d, want at least %d", got, want)
|
|
}
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
|
|
func repeatActionTunnelHistory(endpoint string) []byte {
|
|
if endpoint == openAIRebuildEndpointResponses {
|
|
return []byte(`{"model":"client-model","stream":true,"input":[
|
|
{"type":"function_call","call_id":"history-1","name":"lookup","arguments":"{\"id\":1}"},
|
|
{"type":"function_call_output","call_id":"history-1","output":"same failure"},
|
|
{"type":"function_call","call_id":"history-2","name":"lookup","arguments":"{\"id\":1}"},
|
|
{"type":"function_call_output","call_id":"history-2","output":"same failure"}
|
|
]}`)
|
|
}
|
|
return []byte(`{"model":"client-model","stream":true,"messages":[
|
|
{"role":"assistant","tool_calls":[{"id":"history-1","type":"function","function":{"name":"lookup","arguments":"{\"id\":1}"}}]},
|
|
{"role":"tool","tool_call_id":"history-1","content":"same failure"},
|
|
{"role":"assistant","tool_calls":[{"id":"history-2","type":"function","function":{"name":"lookup","arguments":"{\"id\":1}"}}]},
|
|
{"role":"tool","tool_call_id":"history-2","content":"same failure"}
|
|
]}`)
|
|
}
|
|
|
|
func splitActionTunnelFrames(endpoint string, repeated bool) ([][]byte, [][]byte) {
|
|
id := 2
|
|
if repeated {
|
|
id = 1
|
|
}
|
|
if endpoint == openAIRebuildEndpointResponses {
|
|
return [][]byte{
|
|
[]byte("data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc-live\",\"call_id\":\"call-responses\",\"name\":\"lookup\"}}\n\n"),
|
|
[]byte("data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc-live\",\"output_index\":0,\"delta\":\"{\\\"id\\\":\"}\n\n"),
|
|
[]byte(fmt.Sprintf("data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc-live\",\"output_index\":0,\"delta\":\"%d}\"}\n\n", id)),
|
|
}, [][]byte{
|
|
[]byte("data: {\"type\":\"response.completed\"}\n\n"),
|
|
[]byte("data: [DONE]\n\n"),
|
|
}
|
|
}
|
|
return [][]byte{
|
|
[]byte("data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-chat\",\"function\":{\"name\":\"lookup\",\"arguments\":\"{\\\"id\\\":\"}}]}}]}\n\n"),
|
|
[]byte(fmt.Sprintf("data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"%d}\"}}]}}]}\n\n", id)),
|
|
}, [][]byte{
|
|
[]byte("data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n"),
|
|
[]byte("data: [DONE]\n\n"),
|
|
}
|
|
}
|
|
|
|
// TestStreamGateConfiguredRepeatActionSplitLifecycle drives raw split Chat and
|
|
// Responses action frames through the configured production registry, Core,
|
|
// endpoint codec, and tunnel release sink. It proves the terminal action gate
|
|
// holds every fragment, blocks a repeated action without provider tool wire,
|
|
// and releases a distinct completed lifecycle exactly once after evaluation.
|
|
func TestStreamGateConfiguredRepeatActionSplitLifecycle(t *testing.T) {
|
|
for _, endpoint := range []string{openAIRebuildEndpointChat, openAIRebuildEndpointResponses} {
|
|
for _, repeated := range []bool{true, false} {
|
|
name := endpoint + "/distinct"
|
|
if repeated {
|
|
name = endpoint + "/repeated"
|
|
}
|
|
t.Run(name, func(t *testing.T) {
|
|
fault := 3
|
|
gateCfg := config.StreamEvidenceGateConf{
|
|
Enabled: true,
|
|
MaxRequestFaultRecovery: &fault,
|
|
Filters: []config.StreamGateFilterPolicyConf{{
|
|
Filter: config.StreamGateFilterRepeatGuard,
|
|
Enforcement: config.StreamGateFilterEnforcementBlocking,
|
|
Priority: 10,
|
|
HoldEvidenceRunes: 500,
|
|
}},
|
|
}
|
|
service := &providerFakeRunService{}
|
|
srv := NewServer(config.EdgeOpenAIConf{
|
|
Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15,
|
|
StreamEvidenceGate: gateCfg,
|
|
}, service, nil)
|
|
observations := &recordingOpenAIObservationSink{}
|
|
srv.SetObservationSink(observations)
|
|
|
|
rawRequest := repeatActionTunnelHistory(endpoint)
|
|
route := routeDispatch{Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15}
|
|
requestCtx := newTestRequestContext(t, route, rawRequest)
|
|
req := openAITunnelStreamGateRequest{
|
|
route: route, ingress: requestCtx.ingress, endpoint: endpoint,
|
|
method: http.MethodPost, path: "/v1/" + endpoint, stream: true,
|
|
modelGroupKey: "client-model",
|
|
authorize: func(context.Context) (map[string]string, error) { return nil, nil },
|
|
rewriteBody: func(body []byte, _ string) ([]byte, error) { return body, nil },
|
|
}
|
|
frames := make(chan *iop.ProviderTunnelFrame)
|
|
handle := &fakeTunnelHandle{
|
|
dispatch: edgeservice.RunDispatch{
|
|
RunID: "split-" + endpoint, ModelGroupKey: "client-model",
|
|
Adapter: "openai-compat", Target: "served-model",
|
|
ProviderID: "provider-a", ExecutionPath: string(edgeservice.ProviderPoolPathTunnel),
|
|
},
|
|
frames: frames,
|
|
}
|
|
fctx, err := srv.openAITunnelOutputFilterContext(req)
|
|
if err != nil {
|
|
t.Fatalf("openAITunnelOutputFilterContext: %v", err)
|
|
}
|
|
repeatedFingerprint, _ := openAIRepeatActionFingerprint("lookup", json.RawMessage(`{"id":1}`))
|
|
if _, found := fctx.history.repeatedAction(repeatedFingerprint); !found {
|
|
t.Fatal("production output-filter context lost repeated action history")
|
|
}
|
|
registry, err := openAIStreamGateRegistrySnapshotFor(gateCfg, fctx)
|
|
if err != nil {
|
|
t.Fatalf("openAIStreamGateRegistrySnapshotFor: %v", err)
|
|
}
|
|
writer := newSynchronizedTunnelLifecycleWriter()
|
|
sink := newOpenAITunnelReleaseSink(writer, writer)
|
|
runtime, _, err := srv.buildOpenAITunnelStreamGateRuntime(req, handle, sink, registry)
|
|
if err != nil {
|
|
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
|
|
}
|
|
runDone := make(chan error, 1)
|
|
go func() {
|
|
runDone <- runtime.Run(t.Context())
|
|
}()
|
|
|
|
frames <- &iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START,
|
|
StatusCode: http.StatusOK,
|
|
Headers: map[string]string{"Content-Type": "text/event-stream"},
|
|
}
|
|
actionFrames, terminalFrames := splitActionTunnelFrames(endpoint, repeated)
|
|
for _, frame := range actionFrames {
|
|
frames <- &iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
|
|
Body: frame,
|
|
}
|
|
}
|
|
// Responses emits metadata plus two semantic argument
|
|
// fragments; Chat emits the same two semantic fragments.
|
|
waitForTunnelCodecReleases(t, sink.codec, 2)
|
|
if status, headers, body := writer.snapshot(); status != 0 || headers != 0 || len(body) != 0 {
|
|
t.Fatalf("pre-terminal release = (status=%d headers=%d body=%q), want no caller wire", status, headers, body)
|
|
}
|
|
|
|
for _, frame := range terminalFrames {
|
|
frames <- &iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
|
|
Body: frame,
|
|
}
|
|
}
|
|
close(frames)
|
|
select {
|
|
case runErr := <-runDone:
|
|
if runErr != nil {
|
|
t.Fatalf("runtime.Run: %v", runErr)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("runtime did not finish after terminal evaluation")
|
|
}
|
|
if closeErr := runtime.CloseRequestResources(t.Context(), !repeated); closeErr != nil {
|
|
t.Fatalf("CloseRequestResources: %v", closeErr)
|
|
}
|
|
|
|
status, headerCalls, body := writer.snapshot()
|
|
if headerCalls != 1 {
|
|
t.Fatalf("response-start commits=%d, want 1", headerCalls)
|
|
}
|
|
terminalCommitted, terminalSuccess := sink.terminalStatus()
|
|
if !terminalCommitted || terminalSuccess == repeated {
|
|
t.Fatalf("terminal status=(committed=%v success=%v), repeated=%v", terminalCommitted, terminalSuccess, repeated)
|
|
}
|
|
if repeated {
|
|
if status != http.StatusOK && status != http.StatusBadGateway {
|
|
t.Fatalf("repeated action status=%d, want a single endpoint terminal status", status)
|
|
}
|
|
for _, fragment := range actionFrames {
|
|
if strings.Contains(string(body), string(fragment)) {
|
|
t.Fatalf("repeated provider tool wire was released: %q", body)
|
|
}
|
|
}
|
|
} else {
|
|
if status != http.StatusOK {
|
|
t.Fatalf("distinct action status=%d, want %d", status, http.StatusOK)
|
|
}
|
|
wantBody := strings.Join(append(byteFramesToStrings(actionFrames), byteFramesToStrings(terminalFrames)...), "")
|
|
if string(body) != wantBody {
|
|
t.Fatalf("distinct action wire=%q, want exact lifecycle=%q", body, wantBody)
|
|
}
|
|
}
|
|
|
|
filterEvaluations := 0
|
|
terminalCommits := 0
|
|
for _, observation := range observations.Snapshot() {
|
|
if observation.Kind() == streamgate.ObservationKindTerminalCommitted {
|
|
terminalCommits++
|
|
}
|
|
attribution := observation.Attribution()
|
|
if observation.Kind() != streamgate.ObservationKindFilterEvaluated ||
|
|
attribution == nil || attribution.FilterID() != openAIRepeatActionGuardFilterID {
|
|
continue
|
|
}
|
|
filterEvaluations++
|
|
descriptor := observation.Evidence().DescriptorCode()
|
|
wantDescriptor := "repeat_action_clear"
|
|
if repeated {
|
|
wantDescriptor = "repeated_action_no_progress"
|
|
}
|
|
if descriptor != wantDescriptor {
|
|
t.Fatalf("action descriptor=%q, want %q", descriptor, wantDescriptor)
|
|
}
|
|
}
|
|
if filterEvaluations != 1 || terminalCommits != 1 {
|
|
t.Fatalf("action evaluations=%d terminal commits=%d, want 1/1", filterEvaluations, terminalCommits)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestOpenAITunnelHTTPErrorLifecycle ensures non-2xx raw bodies are not text
|
|
// evidence, retain their response status, and enter the provider-error path.
|
|
func TestOpenAITunnelHTTPErrorLifecycle(t *testing.T) {
|
|
frames := make(chan *iop.ProviderTunnelFrame, 3)
|
|
rawBody := []byte(`{"error":{"message":"upstream failed"}}`)
|
|
frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusInternalServerError, Headers: map[string]string{"Content-Type": "application/json"}}
|
|
frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: rawBody}
|
|
frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true}
|
|
close(frames)
|
|
state := &openAITunnelCodecState{}
|
|
source := newOpenAITunnelEndpointEventSource(edgeservice.ProviderTunnelStream{Frames: frames}, time.Second, nil, nil, openAIRebuildEndpointChat, state)
|
|
start, err := source.NextEvent(t.Context())
|
|
if err != nil || start.Kind() != streamgate.EventKindResponseStart {
|
|
t.Fatalf("start=(%v,%v)", start.Kind(), err)
|
|
}
|
|
responseStart, err := start.AsResponseStart()
|
|
if err != nil || responseStart.Status() != http.StatusInternalServerError {
|
|
t.Fatalf("response start=(%v,%v)", responseStart.Status(), err)
|
|
}
|
|
terminal, err := source.NextEvent(t.Context())
|
|
if err != nil || terminal.Kind() != streamgate.EventKindProviderError {
|
|
t.Fatalf("terminal=(%v,%v)", terminal.Kind(), err)
|
|
}
|
|
response, ok := state.popErrorResponse()
|
|
if !ok || response.status != http.StatusInternalServerError || string(response.body) != string(rawBody) {
|
|
t.Fatalf("error response=(ok=%v status=%d body=%q), want 500 and raw body=%q", ok, response.status, response.body, rawBody)
|
|
}
|
|
if got := response.headers["Content-Type"]; got != "application/json" {
|
|
t.Fatalf("error response content type=%q, want application/json", got)
|
|
}
|
|
}
|
|
|
|
// TestOpenAITunnelHTTPErrorRawPassthroughRuntime closes the production
|
|
// source -> Core -> release-sink gap for unmatched provider errors. A non-2xx
|
|
// body stays opaque even when it looks like endpoint success/tool evidence,
|
|
// and the caller receives the original response exactly once without recovery.
|
|
func TestOpenAITunnelHTTPErrorRawPassthroughRuntime(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
endpoint string
|
|
stream bool
|
|
body []byte
|
|
}{
|
|
{
|
|
name: "chat stream", endpoint: openAIRebuildEndpointChat, stream: true,
|
|
body: []byte(`{"choices":[{"delta":{"content":"opaque-chat","tool_calls":[{"index":0,"id":"call-opaque","function":{"name":"must_not_run","arguments":"{}"}}]}}],"error":{"message":"upstream failed"}}`),
|
|
},
|
|
{
|
|
name: "chat buffered", endpoint: openAIRebuildEndpointChat, stream: false,
|
|
body: []byte(`{"choices":[{"message":{"content":"opaque-chat"}}],"error":{"message":"upstream failed"}}`),
|
|
},
|
|
{
|
|
name: "responses stream", endpoint: openAIRebuildEndpointResponses, stream: true,
|
|
body: []byte(`{"output_text":"opaque-responses","output":[{"type":"function_call","call_id":"call-opaque","name":"must_not_run","arguments":"{}"}],"error":{"message":"upstream failed"}}`),
|
|
},
|
|
{
|
|
name: "responses buffered", endpoint: openAIRebuildEndpointResponses, stream: false,
|
|
body: []byte(`{"output_text":"opaque-responses","output":[{"type":"message","content":[{"type":"output_text","text":"opaque-responses"}]}],"error":{"message":"upstream failed"}}`),
|
|
},
|
|
}
|
|
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
fault := 3
|
|
gateCfg := config.StreamEvidenceGateConf{
|
|
Enabled: true,
|
|
MaxRequestFaultRecovery: &fault,
|
|
Filters: []config.StreamGateFilterPolicyConf{{
|
|
Filter: config.StreamGateFilterProviderError,
|
|
}},
|
|
}
|
|
service := &providerFakeRunService{}
|
|
srv := NewServer(config.EdgeOpenAIConf{
|
|
Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15,
|
|
StreamEvidenceGate: gateCfg,
|
|
}, service, nil)
|
|
recorder := &recordingOpenAIObservationSink{}
|
|
srv.SetObservationSink(recorder)
|
|
|
|
rawRequest := []byte(fmt.Sprintf(`{"model":"client-model","stream":%t}`, tc.stream))
|
|
if tc.endpoint == openAIRebuildEndpointChat {
|
|
rawRequest = []byte(fmt.Sprintf(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":%t}`, tc.stream))
|
|
} else {
|
|
rawRequest = []byte(fmt.Sprintf(`{"model":"client-model","input":"hi","stream":%t}`, tc.stream))
|
|
}
|
|
route := routeDispatch{Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15}
|
|
requestCtx := newTestRequestContext(t, route, rawRequest)
|
|
req := openAITunnelStreamGateRequest{
|
|
route: route, ingress: requestCtx.ingress, endpoint: tc.endpoint,
|
|
method: http.MethodPost, path: "/v1/" + tc.endpoint, stream: tc.stream,
|
|
modelGroupKey: "client-model", requestModel: "",
|
|
authorize: func(context.Context) (map[string]string, error) { return nil, nil },
|
|
rewriteBody: func(body []byte, target string) ([]byte, error) { return body, nil },
|
|
}
|
|
|
|
frames := bufferedTunnelFrames(
|
|
&iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START,
|
|
StatusCode: http.StatusInternalServerError,
|
|
Headers: map[string]string{
|
|
"Content-Type": "application/json", "X-Provider-Error": "retained",
|
|
"Content-Length": "999", "Connection": "close",
|
|
},
|
|
},
|
|
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: tc.body},
|
|
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
|
|
)
|
|
handle := &fakeTunnelHandle{
|
|
dispatch: edgeservice.RunDispatch{
|
|
RunID: "error-" + strings.ReplaceAll(tc.name, " ", "-"), ModelGroupKey: "client-model",
|
|
Adapter: "openai-compat", Target: "served-model", ProviderID: "provider-a",
|
|
ExecutionPath: string(edgeservice.ProviderPoolPathTunnel),
|
|
},
|
|
frames: frames,
|
|
}
|
|
|
|
fctx, err := srv.openAITunnelOutputFilterContext(req)
|
|
if err != nil {
|
|
t.Fatalf("openAITunnelOutputFilterContext: %v", err)
|
|
}
|
|
registry, err := openAIStreamGateRegistrySnapshotFor(gateCfg, fctx)
|
|
if err != nil {
|
|
t.Fatalf("openAIStreamGateRegistrySnapshotFor: %v", err)
|
|
}
|
|
w := newRecordingResponseWriter()
|
|
var sink *openAITunnelReleaseSink
|
|
if tc.stream {
|
|
sink = newOpenAITunnelReleaseSink(w, w)
|
|
} else {
|
|
sink = newOpenAIBufferedTunnelReleaseSink(w, w, "")
|
|
}
|
|
runtime, _, err := srv.buildOpenAITunnelStreamGateRuntime(req, handle, sink, registry)
|
|
if err != nil {
|
|
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
|
|
}
|
|
runErr := runtime.Run(t.Context())
|
|
_ = runtime.CloseRequestResources(t.Context(), runErr == nil)
|
|
if runErr != nil {
|
|
t.Fatalf("runtime.Run: %v", runErr)
|
|
}
|
|
|
|
if w.code != http.StatusInternalServerError || w.headerCallCount() != 1 {
|
|
t.Fatalf("response start=(status=%d headers=%d), want one 500", w.code, w.headerCallCount())
|
|
}
|
|
if got := w.Header().Get("X-Provider-Error"); got != "retained" {
|
|
t.Fatalf("allowed provider header=%q, want retained", got)
|
|
}
|
|
if got := w.Header().Get("Content-Length"); got != "" {
|
|
t.Fatalf("Content-Length leaked after sanitization: %q", got)
|
|
}
|
|
if got := w.body.Bytes(); string(got) != string(tc.body) {
|
|
t.Fatalf("body=%q, want byte-identical %q", got, tc.body)
|
|
}
|
|
if got := len(service.tunnelReqsSnapshot()); got != 0 {
|
|
t.Fatalf("recovery dispatches=%d, want 0", got)
|
|
}
|
|
committed, success := sink.terminalStatus()
|
|
if !committed || success {
|
|
t.Fatalf("terminal=(committed=%v success=%v), want one committed provider error", committed, success)
|
|
}
|
|
|
|
assertUnmatchedProviderErrorPassObservations(t, recorder.Snapshot())
|
|
})
|
|
}
|
|
}
|
|
|
|
func assertUnmatchedProviderErrorPassObservations(t *testing.T, observations []streamgate.FilterObservation) {
|
|
t.Helper()
|
|
providerPass := false
|
|
terminalCommits := 0
|
|
for _, observation := range observations {
|
|
if observation.Kind() == streamgate.ObservationKindTerminalCommitted {
|
|
terminalCommits++
|
|
}
|
|
if observation.Recovery() != nil {
|
|
t.Fatalf("unexpected recovery observation: kind=%s", observation.Kind())
|
|
}
|
|
if evidence := observation.Evidence(); evidence != nil {
|
|
if evidence.EventKind() == streamgate.EventKindTextDelta || evidence.EventKind() == streamgate.EventKindToolCallFragment {
|
|
t.Fatalf("non-2xx body became semantic evidence: kind=%s", evidence.EventKind())
|
|
}
|
|
}
|
|
attribution := observation.Attribution()
|
|
decision := observation.DecisionPolicy()
|
|
if observation.Kind() == streamgate.ObservationKindFilterEvaluated &&
|
|
attribution != nil && attribution.FilterID() == openAIProviderErrorFilterID &&
|
|
decision != nil && decision.Outcome() == streamgate.FilterOutcomeKindEvaluated &&
|
|
decision.DecisionKind() == streamgate.FilterDecisionKindPass {
|
|
providerPass = true
|
|
}
|
|
}
|
|
if !providerPass {
|
|
t.Fatalf("provider-error evaluated-pass observation missing: kinds=%v", observationKinds(observations))
|
|
}
|
|
if terminalCommits != 1 {
|
|
t.Fatalf("terminal commit observations=%d, want 1", terminalCommits)
|
|
}
|
|
}
|
|
|
|
func mustFinishTunnelCodec(t *testing.T, codec *openAITunnelEndpointCodec) []streamgate.NormalizedEvent {
|
|
t.Helper()
|
|
events, err := codec.finishTransport(nil, false)
|
|
if err != nil {
|
|
t.Fatalf("finishTransport: %v", err)
|
|
}
|
|
return events
|
|
}
|
|
|
|
func byteFramesToStrings(frames [][]byte) []string {
|
|
out := make([]string, len(frames))
|
|
for i, frame := range frames {
|
|
out[i] = string(frame)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func releaseTunnelCodecEvents(t *testing.T, state *openAITunnelCodecState, events []streamgate.NormalizedEvent) string {
|
|
t.Helper()
|
|
w := newRecordingResponseWriter()
|
|
sink := newOpenAITunnelReleaseSink(w, w)
|
|
sink.codec = state
|
|
start, err := streamgate.NewResponseStartEvent(streamGateChannelDefault, http.StatusOK, map[string]string{"Content-Type": "text/event-stream"}, time.Now())
|
|
if err != nil {
|
|
t.Fatalf("NewResponseStartEvent: %v", err)
|
|
}
|
|
responseStart, err := start.AsResponseStart()
|
|
if err != nil {
|
|
t.Fatalf("AsResponseStart: %v", err)
|
|
}
|
|
if _, err := sink.CommitResponseStart(t.Context(), responseStart); err != nil {
|
|
t.Fatalf("CommitResponseStart: %v", err)
|
|
}
|
|
for _, event := range events {
|
|
switch event.Kind() {
|
|
case streamgate.EventKindTextDelta:
|
|
value, _ := event.AsTextDelta()
|
|
release, _ := streamgate.NewReleaseTextDeltaEvent(event.Channel(), value, event.Timestamp())
|
|
if _, err := sink.Release(t.Context(), release); err != nil {
|
|
t.Fatalf("Release text: %v", err)
|
|
}
|
|
case streamgate.EventKindReasoningDelta:
|
|
value, _ := event.AsReasoningDelta()
|
|
release, _ := streamgate.NewReleaseReasoningDeltaEvent(event.Channel(), value, event.Timestamp())
|
|
if _, err := sink.Release(t.Context(), release); err != nil {
|
|
t.Fatalf("Release reasoning: %v", err)
|
|
}
|
|
case streamgate.EventKindToolCallFragment:
|
|
value, _ := event.AsToolCallFragment()
|
|
release, _ := streamgate.NewReleaseToolCallFragmentEvent(event.Channel(), value.ID, value.Name, value.Arguments, event.Timestamp())
|
|
if _, err := sink.Release(t.Context(), release); err != nil {
|
|
t.Fatalf("Release tool: %v", err)
|
|
}
|
|
case streamgate.EventKindTerminal:
|
|
value, _ := event.AsTerminal()
|
|
if _, err := sink.CommitTerminal(t.Context(), value); err != nil {
|
|
t.Fatalf("CommitTerminal: %v", err)
|
|
}
|
|
}
|
|
}
|
|
return w.body.String()
|
|
}
|
|
|
|
func TestStreamGateConfiguredRepeatGuardLargeTunnelEvent(t *testing.T) {
|
|
payload := uniqueKoreanRunes(5000)
|
|
if got := utf8.RuneCountInString(payload); got != 5000 {
|
|
t.Fatalf("uniqueKoreanRunes count = %d, want 5000", got)
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
endpoint string
|
|
semanticFrame []byte
|
|
terminalFrame []byte
|
|
}{
|
|
{
|
|
name: "chat content",
|
|
endpoint: openAIRebuildEndpointChat,
|
|
semanticFrame: []byte(fmt.Sprintf("data: {\"id\":\"chatcmpl-1\",\"object\":\"chat.completion.chunk\",\"created\":1700000000,\"model\":\"served-model\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"%s\"}}]}\n\n", payload)),
|
|
terminalFrame: []byte("data: [DONE]\n\n"),
|
|
},
|
|
{
|
|
name: "chat reasoning",
|
|
endpoint: openAIRebuildEndpointChat,
|
|
semanticFrame: []byte(fmt.Sprintf("data: {\"id\":\"chatcmpl-1\",\"object\":\"chat.completion.chunk\",\"created\":1700000000,\"model\":\"served-model\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"%s\"}}]}\n\n", payload)),
|
|
terminalFrame: []byte("data: [DONE]\n\n"),
|
|
},
|
|
{
|
|
name: "responses output text",
|
|
endpoint: openAIRebuildEndpointResponses,
|
|
semanticFrame: []byte(fmt.Sprintf("data: {\"type\":\"response.output_text.delta\",\"delta\":\"%s\"}\n\n", payload)),
|
|
terminalFrame: []byte("data: [DONE]\n\n"),
|
|
},
|
|
{
|
|
name: "responses reasoning",
|
|
endpoint: openAIRebuildEndpointResponses,
|
|
semanticFrame: []byte(fmt.Sprintf("data: {\"type\":\"response.reasoning_text.delta\",\"delta\":\"%s\"}\n\n", payload)),
|
|
terminalFrame: []byte("data: [DONE]\n\n"),
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
fault := 3
|
|
gateCfg := config.StreamEvidenceGateConf{
|
|
Enabled: true,
|
|
MaxRequestFaultRecovery: &fault,
|
|
Filters: []config.StreamGateFilterPolicyConf{{
|
|
Filter: config.StreamGateFilterRepeatGuard,
|
|
Enforcement: config.StreamGateFilterEnforcementBlocking,
|
|
Priority: 10,
|
|
HoldEvidenceRunes: 500,
|
|
}},
|
|
}
|
|
service := &providerFakeRunService{}
|
|
srv := NewServer(config.EdgeOpenAIConf{
|
|
Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15,
|
|
StreamEvidenceGate: gateCfg,
|
|
}, service, nil)
|
|
observations := &recordingOpenAIObservationSink{}
|
|
srv.SetObservationSink(observations)
|
|
|
|
var rawReq []byte
|
|
if tt.endpoint == openAIRebuildEndpointChat {
|
|
rawReq = []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}]}`)
|
|
} else {
|
|
rawReq = []byte(`{"model":"client-model","input":"hi"}`)
|
|
}
|
|
route := routeDispatch{Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15}
|
|
requestCtx := newTestRequestContext(t, route, rawReq)
|
|
req := openAITunnelStreamGateRequest{
|
|
route: route, ingress: requestCtx.ingress, endpoint: tt.endpoint,
|
|
method: http.MethodPost, path: "/v1/" + tt.endpoint, stream: true,
|
|
modelGroupKey: "client-model",
|
|
authorize: func(context.Context) (map[string]string, error) { return nil, nil },
|
|
rewriteBody: func(body []byte, _ string) ([]byte, error) { return body, nil },
|
|
}
|
|
frames := make(chan *iop.ProviderTunnelFrame)
|
|
handle := &fakeTunnelHandle{
|
|
dispatch: edgeservice.RunDispatch{
|
|
RunID: "large-" + tt.name, ModelGroupKey: "client-model",
|
|
Adapter: "openai-compat", Target: "served-model",
|
|
ProviderID: "provider-a", ExecutionPath: string(edgeservice.ProviderPoolPathTunnel),
|
|
},
|
|
frames: frames,
|
|
}
|
|
fctx, err := srv.openAITunnelOutputFilterContext(req)
|
|
if err != nil {
|
|
t.Fatalf("openAITunnelOutputFilterContext: %v", err)
|
|
}
|
|
registry, err := openAIStreamGateRegistrySnapshotFor(gateCfg, fctx)
|
|
if err != nil {
|
|
t.Fatalf("openAIStreamGateRegistrySnapshotFor: %v", err)
|
|
}
|
|
writer := newSynchronizedTunnelLifecycleWriter()
|
|
sink := newOpenAITunnelReleaseSink(writer, writer)
|
|
runtime, _, err := srv.buildOpenAITunnelStreamGateRuntime(req, handle, sink, registry)
|
|
if err != nil {
|
|
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
|
|
}
|
|
runDone := make(chan error, 1)
|
|
go func() {
|
|
runDone <- runtime.Run(t.Context())
|
|
}()
|
|
|
|
frames <- &iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START,
|
|
StatusCode: http.StatusOK,
|
|
Headers: map[string]string{"Content-Type": "text/event-stream"},
|
|
}
|
|
frames <- &iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
|
|
Body: tt.semanticFrame,
|
|
}
|
|
frames <- &iop.ProviderTunnelFrame{
|
|
Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
|
|
Body: tt.terminalFrame,
|
|
}
|
|
close(frames)
|
|
|
|
select {
|
|
case runErr := <-runDone:
|
|
if runErr != nil {
|
|
t.Fatalf("runtime.Run: %v", runErr)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("runtime did not finish after large event evaluation")
|
|
}
|
|
if closeErr := runtime.CloseRequestResources(t.Context(), true); closeErr != nil {
|
|
t.Fatalf("CloseRequestResources: %v", closeErr)
|
|
}
|
|
|
|
status, headerCalls, body := writer.snapshot()
|
|
if headerCalls != 1 {
|
|
t.Fatalf("response-start commits = %d, want 1", headerCalls)
|
|
}
|
|
if status != http.StatusOK {
|
|
t.Fatalf("HTTP status = %d, want %d", status, http.StatusOK)
|
|
}
|
|
terminalCommitted, terminalSuccess := sink.terminalStatus()
|
|
if !terminalCommitted || !terminalSuccess {
|
|
t.Fatalf("terminal status = (committed=%v, success=%v), want (true, true)", terminalCommitted, terminalSuccess)
|
|
}
|
|
wantBody := string(tt.semanticFrame) + string(tt.terminalFrame)
|
|
if string(body) != wantBody {
|
|
t.Fatalf("released body length = %d, want exact wire length %d", len(body), len(wantBody))
|
|
}
|
|
|
|
for _, obs := range observations.Snapshot() {
|
|
if obs.Kind() == streamgate.ObservationKindRecoveryDispatched {
|
|
t.Fatalf("unexpected recovery dispatched observation: %+v", obs)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|