package openai
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
edgeservice "iop/apps/edge/internal/service"
"iop/packages/go/config"
"iop/packages/go/streamgate"
iop "iop/proto/gen/iop"
)
// --- test-only InjectedViolationFilter --------------------------------------
//
// injectedViolationFilter is the test-only filter named by the vertical-slice
// plan (SDD S16). It is deliberately kept local to this test file: production
// wiring (openAIStreamGateRegistrations) never registers it. It triggers an
// exact_replay RecoveryIntent on the first `trigger` text_delta evaluations it
// sees, then passes for the rest of the request.
//
// It uses terminal_gate hold mode rather than none: none mode is reserved
// for provider-error-only observers (its trigger-ready rule only fires on
// provider_error/terminal events regardless of content subscription — see
// packages/go/streamgate/evidence_tail.go deriveTriggerReady), so a filter
// that wants to judge accumulated content must hold it until the terminal
// trigger like a real output-validation filter would.
// injectedViolationPriority is shared by every test registration of
// injectedViolationFilter and its emitted RecoveryIntent: the Arbiter
// requires an exact match between a violation's intent priority and the
// filter's effective registration priority.
const injectedViolationPriority = 10
type injectedViolationFilter struct {
streamgate.FilterBase
requestRef string
mu sync.Mutex
trigger int
}
func newInjectedViolationFilter(t *testing.T, id string, trigger int, requestRef string) *injectedViolationFilter {
t.Helper()
base, err := streamgate.NewFilterBase(id)
if err != nil {
t.Fatalf("NewFilterBase: %v", err)
}
return &injectedViolationFilter{FilterBase: base, requestRef: requestRef, trigger: trigger}
}
func (f *injectedViolationFilter) Applies(streamgate.FilterContext) bool { return true }
func (f *injectedViolationFilter) HoldRequirement(streamgate.FilterContext) streamgate.FilterHoldRequirement {
req, err := streamgate.NewFilterHoldRequirementTerminalGate(streamGateChannelDefault, []streamgate.EventKind{streamgate.EventKindTextDelta}, streamgate.EventKindTerminal)
if err != nil {
panic(err)
}
return req
}
func (f *injectedViolationFilter) Evaluate(ctx context.Context, fctx streamgate.FilterContext, batch streamgate.EvidenceBatch) (streamgate.FilterDecision, error) {
f.mu.Lock()
fire := f.trigger > 0
if fire {
f.trigger--
}
f.mu.Unlock()
events := batch.Events()
kind := streamgate.EventKindTextDelta
if len(events) > 0 {
kind = events[len(events)-1].Kind()
}
evidence, err := streamgate.NewSanitizedEvidence(kind, streamGateChannelDefault, "test.rule", "test_injected_violation", streamgate.FixedFingerprint{0x02}, 1, 0, streamgate.FilterOutcomeKindEvaluated, batch.CapturedAt())
if err != nil {
return streamgate.FilterDecision{}, err
}
if !fire {
return streamgate.NewFilterDecision(streamgate.FilterDecisionKindPass, "test.consumer", f.ID(), "test.rule", evidence, nil)
}
directive, err := streamgate.NewRecoveryDirectiveExact(f.requestRef)
if err != nil {
return streamgate.FilterDecision{}, err
}
intent, err := streamgate.NewRecoveryIntent(streamgate.RecoveryStrategyExactReplay, directive, "test_injected", injectedViolationPriority)
if err != nil {
return streamgate.FilterDecision{}, err
}
return streamgate.NewFilterDecision(streamgate.FilterDecisionKindViolation, "test.consumer", f.ID(), "test.rule", evidence, &intent)
}
var _ streamgate.Filter = (*injectedViolationFilter)(nil)
func streamGateTestRegistry(t *testing.T, extra ...streamgate.FilterRegistration) streamgate.FilterRegistrySnapshot {
t.Helper()
regs, err := openAIStreamGateNoopRegistrations()
if err != nil {
t.Fatalf("openAIStreamGateNoopRegistrations: %v", err)
}
regs = append(regs, extra...)
snap, err := streamgate.NewFilterRegistrySnapshot(streamGateConfigGeneration, regs, nil)
if err != nil {
t.Fatalf("NewFilterRegistrySnapshot: %v", err)
}
return snap
}
// --- recording ResponseWriter -------------------------------------------------
// recordingResponseWriter records the ordered sequence of Header/WriteHeader/
// Write calls so tests can assert that no status/header/role write happens
// before the Core's first safe release.
type recordingResponseWriter struct {
header http.Header
body bytes.Buffer
code int
calls []string
}
func newRecordingResponseWriter() *recordingResponseWriter {
return &recordingResponseWriter{header: make(http.Header)}
}
func (w *recordingResponseWriter) Header() http.Header { return w.header }
func (w *recordingResponseWriter) WriteHeader(code int) {
w.code = code
w.calls = append(w.calls, fmt.Sprintf("header:%d", code))
}
func (w *recordingResponseWriter) Write(p []byte) (int, error) {
w.calls = append(w.calls, fmt.Sprintf("write:%d", len(p)))
return w.body.Write(p)
}
func (w *recordingResponseWriter) Flush() {}
func (w *recordingResponseWriter) headerCallCount() int {
n := 0
for _, c := range w.calls {
if strings.HasPrefix(c, "header:") {
n++
}
}
return n
}
var (
_ http.ResponseWriter = (*recordingResponseWriter)(nil)
_ http.Flusher = (*recordingResponseWriter)(nil)
)
// --- SSE parsing helper -------------------------------------------------------
func parseSSEChatChunks(t *testing.T, body string) []chatCompletionChunk {
t.Helper()
var chunks []chatCompletionChunk
for _, frame := range strings.Split(body, "\n\n") {
frame = strings.TrimSpace(frame)
if frame == "" {
continue
}
payload := strings.TrimPrefix(frame, "data: ")
if payload == "[DONE]" {
continue
}
var chunk chatCompletionChunk
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
continue // error chunks decode into errorResponse, not chatCompletionChunk; skip
}
if len(chunk.Choices) == 0 {
continue
}
chunks = append(chunks, chunk)
}
return chunks
}
func joinedContent(chunks []chatCompletionChunk) string {
var b strings.Builder
for _, c := range chunks {
if len(c.Choices) > 0 {
b.WriteString(c.Choices[0].Delta.Content)
}
}
return b.String()
}
// --- chat handler-level: disabled vs enabled Noop parity ---------------------
func streamGateChatHandlerFixture(t *testing.T, enabled bool) (*Server, *fakeRunService) {
t.Helper()
fake := &fakeRunService{events: make(chan *iop.RunEvent, 3)}
fake.events <- &iop.RunEvent{Type: "delta", Delta: "hello"}
fake.events <- &iop.RunEvent{Type: "delta", Delta: " world"}
fake.events <- &iop.RunEvent{Type: "complete", Usage: &iop.Usage{InputTokens: 1, OutputTokens: 2}}
srv := NewServer(config.EdgeOpenAIConf{
Adapter: "ollama",
Target: "llama-fixed",
SessionID: "cli",
TimeoutSec: 15,
StreamEvidenceGate: config.StreamEvidenceGateConf{
Enabled: enabled,
},
}, fake, nil)
return srv, fake
}
func TestStreamGateChatHandlerDisabledVsEnabledNoopPass(t *testing.T) {
body := `{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`
for _, enabled := range []bool{false, true} {
enabled := enabled
name := "disabled"
if enabled {
name = "enabled"
}
t.Run(name, func(t *testing.T) {
srv, fake := streamGateChatHandlerFixture(t, enabled)
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(body))
w := httptest.NewRecorder()
srv.handleChatCompletions(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status: got %d body=%s", w.Code, w.Body.String())
}
chunks := parseSSEChatChunks(t, w.Body.String())
if len(chunks) == 0 {
t.Fatalf("no SSE chunks decoded from body=%s", w.Body.String())
}
if chunks[0].Choices[0].Delta.Role != "assistant" {
t.Fatalf("first chunk role: got %+v", chunks[0].Choices[0].Delta)
}
if got := joinedContent(chunks); got != "hello world" {
t.Fatalf("joined content: got %q", got)
}
last := chunks[len(chunks)-1]
if last.Choices[0].FinishReason != "stop" {
t.Fatalf("finish reason: got %q", last.Choices[0].FinishReason)
}
if !strings.Contains(w.Body.String(), "data: [DONE]") {
t.Fatalf("missing [DONE] sentinel: body=%s", w.Body.String())
}
if len(fake.reqsSnapshot()) != 1 {
t.Fatalf("dispatch attempts: got %d, want 1", len(fake.reqsSnapshot()))
}
})
}
}
// --- adapter-level: chat runtime + injected violation recovery ---------------
// buildStreamGateChatFixture builds a ready-to-run chatDispatchContext plus the
// initial fakeRunResult a handler would already have dispatched, mirroring
// production ingress without going through the HTTP decode path.
func buildStreamGateChatFixture(t *testing.T, enabled bool, maxRequestFaultRecovery int) (*Server, *chatDispatchContext, *fakeRunResult, *fakeRunService) {
t.Helper()
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
var req chatCompletionRequest
if err := json.Unmarshal(rawBody, &req); err != nil {
t.Fatalf("decode fixture body: %v", err)
}
fault := maxRequestFaultRecovery
fake := &fakeRunService{}
srv := NewServer(config.EdgeOpenAIConf{
Adapter: "ollama",
Target: "llama-fixed",
SessionID: "cli",
TimeoutSec: 15,
StreamEvidenceGate: config.StreamEvidenceGateConf{
Enabled: enabled,
MaxRequestFaultRecovery: &fault,
},
}, fake, nil)
route := routeDispatch{Adapter: "ollama", Target: "llama-fixed", SessionID: "cli", TimeoutSec: 15}
requestCtx := newTestRequestContext(t, route, rawBody)
basePrompt := promptFromMessages(req.Messages)
outputPolicy := srv.resolveOutputPolicy(basePrompt)
dc := srv.newChatDispatchContext(requestCtx, req, basePrompt, outputPolicy)
initial := &fakeRunResult{
dispatch: edgeservice.RunDispatch{RunID: "run-attempt-1", Adapter: "ollama", Target: "llama-fixed", SessionID: "cli", TimeoutSec: 15},
events: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "should-be-discarded"},
&iop.RunEvent{Type: "complete"},
),
}
return srv, dc, initial, fake
}
func TestStreamGateChatBuildRuntimeInjectedViolationSingleRetrySuccess(t *testing.T) {
srv, dc, initial, fake := buildStreamGateChatFixture(t, true, 3)
fake.eventRuns = []chan *iop.RunEvent{bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "final answer"},
&iop.RunEvent{Type: "complete"},
)}
fake.runIDs = []string{"run-attempt-2"}
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
violation := newInjectedViolationFilter(t, "test.injected_violation", 1, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
registry := streamGateTestRegistry(t, reg)
w := newRecordingResponseWriter()
rt, usage, 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)
}
_ = usage
if got := len(fake.reqsSnapshot()); got != 1 {
t.Fatalf("recovery dispatch count: got %d, want 1 (exact_replay resubmits once)", got)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1 (no eager header from the discarded first attempt)", got)
}
chunks := parseSSEChatChunks(t, w.body.String())
roleCount := 0
for _, c := range chunks {
if c.Choices[0].Delta.Role == "assistant" {
roleCount++
}
}
if roleCount != 1 {
t.Fatalf("assistant role chunk count: got %d, want 1", roleCount)
}
if got := joinedContent(chunks); got != "final answer" {
t.Fatalf("joined content: got %q, want only the retry attempt's content (no leak from the discarded first attempt)", got)
}
}
func TestStreamGateChatBuildRuntimeRecoveryBudgetExhaustionTerminatesWithError(t *testing.T) {
srv, dc, initial, fake := buildStreamGateChatFixture(t, true, 1)
fake.eventRuns = []chan *iop.RunEvent{bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "still-bad"},
&iop.RunEvent{Type: "complete"},
)}
fake.runIDs = []string{"run-attempt-2"}
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
// trigger=2: violates on both the initial attempt and the one allowed
// recovery attempt, so the request-total cap (1) is exhausted after the
// first recovery and the second violation must terminate, not dispatch a
// third attempt.
violation := newInjectedViolationFilter(t, "test.injected_violation", 2, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
registry := streamGateTestRegistry(t, reg)
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 != 1 {
t.Fatalf("recovery dispatch count: got %d, want 1 (cap=1 allows exactly one recovery dispatch)", got)
}
if !strings.Contains(w.body.String(), `"error"`) {
t.Fatalf("expected an SSE error terminal, got body=%s", w.body.String())
}
if strings.Contains(w.body.String(), "still-bad") {
t.Fatalf("exhausted recovery must not leak the still-violating attempt's content: body=%s", w.body.String())
}
}
func TestStreamGateChatBuildRuntimeObserveOnlyDoesNotBlock(t *testing.T) {
srv, dc, initial, fake := buildStreamGateChatFixture(t, true, 3)
initial.events = bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "observed content"},
&iop.RunEvent{Type: "complete"},
)
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
violation := newInjectedViolationFilter(t, "test.injected_violation_observe", 1, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementObserveOnly, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
registry := streamGateTestRegistry(t, reg)
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("observe-only violation must not trigger a recovery dispatch: got %d dispatches", got)
}
chunks := parseSSEChatChunks(t, w.body.String())
if got := joinedContent(chunks); got != "observed content" {
t.Fatalf("joined content: got %q, want the pass-through content unblocked by the observe-only violation", got)
}
}
// --- adapter-level: tunnel runtime Noop passthrough ---------------------------
func TestStreamGateTunnelBuildRuntimeNoopPassRelaysRawBytes(t *testing.T) {
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
fault := 3
fake := &providerFakeRunService{}
srv := NewServer(config.EdgeOpenAIConf{
Adapter: "openai-compat",
Target: "served-model",
TimeoutSec: 15,
StreamEvidenceGate: config.StreamEvidenceGateConf{
Enabled: true,
MaxRequestFaultRecovery: &fault,
},
}, fake, nil)
route := routeDispatch{Adapter: "openai-compat", Target: "served-model", TimeoutSec: 15}
requestCtx := newTestRequestContext(t, route, rawBody)
frames := make(chan *iop.ProviderTunnelFrame, 4)
frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200, Headers: map[string]string{"Content-Type": "text/event-stream"}}
frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte("data: {\"choices\":[{\"delta\":{\"content\":\"raw\"}}]}\n\n")}
frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END}
close(frames)
handle := &fakeTunnelHandle{
dispatch: edgeservice.RunDispatch{RunID: "tunnel-attempt-1", Adapter: "openai-compat", Target: "served-model"},
frames: frames,
}
req := openAITunnelStreamGateRequest{
route: route,
ingress: requestCtx.ingress,
endpoint: openAIRebuildEndpointChat,
method: http.MethodPost,
path: "/v1/chat/completions",
modelGroupKey: "client-model",
requestModel: "", // no model echo present in the fixture body; rewrite is a no-op
authorize: func(context.Context) (map[string]string, error) { return nil, nil },
rewriteBody: func(body []byte, target string) ([]byte, error) {
return body, nil
},
}
w := newRecordingResponseWriter()
registry := streamGateTestRegistry(t)
rt, _, err := srv.buildOpenAITunnelStreamGateRuntime(req, handle, newOpenAITunnelReleaseSink(w, w), registry)
if err != nil {
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
}
if err := rt.Run(context.Background()); err != nil {
t.Fatalf("rt.Run: %v", err)
}
if w.code != http.StatusOK {
t.Fatalf("status: got %d", w.code)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1", got)
}
if !strings.Contains(w.body.String(), `"content":"raw"`) {
t.Fatalf("raw provider bytes were not relayed byte-for-byte: body=%s", w.body.String())
}
}
// --- S16: strategy recovery cap enforcement (REVIEW_EDGE_VERTICAL_SLICE-4) ---
// newStrategyCapChatFixture mirrors buildStreamGateChatFixture but pins an
// explicit per-strategy recovery cap so the request-start RuntimeOptions
// snapshot carries a non-empty strategy-limit map into the Core policy.
func newStrategyCapChatFixture(t *testing.T, maxRequestFaultRecovery, maxStrategyFaultRecovery int) (*Server, *chatDispatchContext, *fakeRunResult, *fakeRunService) {
t.Helper()
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
var req chatCompletionRequest
if err := json.Unmarshal(rawBody, &req); err != nil {
t.Fatalf("decode fixture body: %v", err)
}
total := maxRequestFaultRecovery
strategy := maxStrategyFaultRecovery
fake := &fakeRunService{}
srv := NewServer(config.EdgeOpenAIConf{
Adapter: "ollama",
Target: "llama-fixed",
SessionID: "cli",
TimeoutSec: 15,
StreamEvidenceGate: config.StreamEvidenceGateConf{
Enabled: true,
MaxRequestFaultRecovery: &total,
MaxStrategyFaultRecovery: &strategy,
},
}, fake, nil)
route := routeDispatch{Adapter: "ollama", Target: "llama-fixed", SessionID: "cli", TimeoutSec: 15}
requestCtx := newTestRequestContext(t, route, rawBody)
basePrompt := promptFromMessages(req.Messages)
outputPolicy := srv.resolveOutputPolicy(basePrompt)
dc := srv.newChatDispatchContext(requestCtx, req, basePrompt, outputPolicy)
initial := &fakeRunResult{
dispatch: edgeservice.RunDispatch{RunID: "run-attempt-1", Adapter: "ollama", Target: "llama-fixed", SessionID: "cli", TimeoutSec: 15},
events: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "should-be-discarded"},
&iop.RunEvent{Type: "complete"},
),
}
return srv, dc, initial, fake
}
func runStrategyCapChatFixture(t *testing.T, srv *Server, dc *chatDispatchContext, initial *fakeRunResult, trigger int) (*recordingResponseWriter, error) {
t.Helper()
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
violation := newInjectedViolationFilter(t, "test.injected_violation", trigger, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
registry := streamGateTestRegistry(t, reg)
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)
}
return w, rt.Run(context.Background())
}
// TestStreamGateChatStrategyCapZeroBlocksExactReplayRecovery verifies that an
// explicit strategy cap of 0 forbids the exact_replay fault strategy even when
// the request-total cap allows recovery: the single violation must terminate
// with an error and dispatch zero recovery attempts.
func TestStreamGateChatStrategyCapZeroBlocksExactReplayRecovery(t *testing.T) {
srv, dc, initial, fake := newStrategyCapChatFixture(t, 3, 0)
fake.eventRuns = []chan *iop.RunEvent{bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "must-not-be-reached"},
&iop.RunEvent{Type: "complete"},
)}
fake.runIDs = []string{"run-attempt-2"}
w, runErr := runStrategyCapChatFixture(t, srv, dc, initial, 1)
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := len(fake.reqsSnapshot()); got != 0 {
t.Fatalf("strategy cap 0 must block all exact_replay dispatch: got %d dispatches", got)
}
if !strings.Contains(w.body.String(), `"error"`) {
t.Fatalf("expected an SSE error terminal for a strategy-capped violation, body=%s", w.body.String())
}
}
// TestStreamGateChatStrategyCapOneExhaustsOnSecondViolation verifies that a
// strategy cap of 1 (below the request-total cap of 3) allows exactly one
// exact_replay recovery dispatch and then exhausts, terminating on the second
// violation instead of dispatching a second recovery.
func TestStreamGateChatStrategyCapOneExhaustsOnSecondViolation(t *testing.T) {
srv, dc, initial, fake := newStrategyCapChatFixture(t, 3, 1)
fake.eventRuns = []chan *iop.RunEvent{bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "still-bad"},
&iop.RunEvent{Type: "complete"},
)}
fake.runIDs = []string{"run-attempt-2"}
w, runErr := runStrategyCapChatFixture(t, srv, dc, initial, 2)
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := len(fake.reqsSnapshot()); got != 1 {
t.Fatalf("strategy cap 1 allows exactly one exact_replay dispatch: got %d", got)
}
if !strings.Contains(w.body.String(), `"error"`) {
t.Fatalf("expected an SSE error terminal once the strategy cap is exhausted, body=%s", w.body.String())
}
}
// --- S16: current-binding lifecycle + terminal outcome (REVIEW_EDGE_VERTICAL_SLICE-3) ---
// TestStreamGateChatCloseRequestResourcesAbortsLatestRun verifies that after a
// recovery cycle the request runtime cleans up the *current* (latest) attempt
// binding — not just the initial handle — and that a non-graceful close cancels
// the latest provider run while a graceful close does not, and that repeated
// closes are idempotent.
func TestStreamGateChatCloseRequestResourcesAbortsLatestRun(t *testing.T) {
for _, graceful := range []bool{false, true} {
graceful := graceful
name := "abort"
if graceful {
name = "graceful"
}
t.Run(name, func(t *testing.T) {
srv, dc, initial, fake := buildStreamGateChatFixture(t, true, 3)
fake.eventRuns = []chan *iop.RunEvent{bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "final answer"},
&iop.RunEvent{Type: "complete"},
)}
fake.runIDs = []string{"run-attempt-2"}
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
violation := newInjectedViolationFilter(t, "test.injected_violation", 1, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
registry := streamGateTestRegistry(t, reg)
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)
}
// The recovery cycle already aborted the initial attempt (run-attempt-1).
cancelsAfterRun := fake.cancelCallsSnapshot()
if !cancelRunIDsContain(cancelsAfterRun, "run-attempt-1") {
t.Fatalf("recovery must abort the initial attempt run-attempt-1, cancels=%v", cancelRunIDs(cancelsAfterRun))
}
if cancelRunIDsContain(cancelsAfterRun, "run-attempt-2") {
t.Fatalf("the latest attempt must not be canceled by a successful terminal, cancels=%v", cancelRunIDs(cancelsAfterRun))
}
if err := rt.CloseRequestResources(context.Background(), graceful); err != nil {
t.Fatalf("CloseRequestResources: %v", err)
}
// Idempotent: a second close must be a no-op regardless of the flag.
if err := rt.CloseRequestResources(context.Background(), !graceful); err != nil {
t.Fatalf("second CloseRequestResources: %v", err)
}
cancelsAfterClose := fake.cancelCallsSnapshot()
latestCanceled := cancelRunIDsContain(cancelsAfterClose, "run-attempt-2")
if graceful && latestCanceled {
t.Fatalf("graceful close must not cancel the latest run, cancels=%v", cancelRunIDs(cancelsAfterClose))
}
if !graceful && !latestCanceled {
t.Fatalf("non-graceful close must cancel the latest run once, cancels=%v", cancelRunIDs(cancelsAfterClose))
}
// Exactly-once: at most one cancel per attempt id.
if n := cancelRunIDCount(cancelsAfterClose, "run-attempt-1"); n != 1 {
t.Fatalf("initial attempt canceled %d times, want exactly 1", n)
}
if n := cancelRunIDCount(cancelsAfterClose, "run-attempt-2"); n > 1 {
t.Fatalf("latest attempt canceled %d times, want at most 1", n)
}
})
}
}
func cancelRunIDs(cancels []edgeservice.CancelRunRequest) []string {
ids := make([]string, 0, len(cancels))
for _, c := range cancels {
ids = append(ids, c.RunID)
}
return ids
}
func cancelRunIDsContain(cancels []edgeservice.CancelRunRequest, runID string) bool {
return cancelRunIDCount(cancels, runID) > 0
}
func cancelRunIDCount(cancels []edgeservice.CancelRunRequest, runID string) int {
n := 0
for _, c := range cancels {
if c.RunID == runID {
n++
}
}
return n
}
// TestStreamGateUsageStatusFromTerminalOutcome verifies the host maps a
// Core-committed terminal into the correct usage-metric status: a committed
// error terminal is reported as error (never success), a committed success
// terminal as success, and a caller cancel/timeout as cancel.
func TestStreamGateUsageStatusFromTerminalOutcome(t *testing.T) {
cases := []struct {
name string
runErr error
committed bool
success bool
want string
}{
{"success terminal", nil, true, true, usageStatusSuccess},
{"error terminal not success metric", nil, true, false, usageStatusError},
{"caller cancel", context.Canceled, false, false, usageStatusCancel},
{"deadline exceeded", context.DeadlineExceeded, false, false, usageStatusCancel},
{"internal run error", fmt.Errorf("boom"), false, false, usageStatusError},
{"no terminal committed", nil, false, false, usageStatusError},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := streamGateUsageStatus(tc.runErr, tc.committed, tc.success); got != tc.want {
t.Fatalf("streamGateUsageStatus(%v,%v,%v) = %q, want %q", tc.runErr, tc.committed, tc.success, got, tc.want)
}
})
}
}
// TestOpenAIChatSSESinkCapturesTerminalOutcome verifies the release sink records
// the terminal disposition the Core commits so the host outcome is authoritative.
func TestOpenAIChatSSESinkCapturesTerminalOutcome(t *testing.T) {
successTerm, err := streamgate.NewSuccessTerminalResult(streamGateChannelDefault, time.Now())
if err != nil {
t.Fatalf("NewSuccessTerminalResult: %v", err)
}
w := newRecordingResponseWriter()
sink := newOpenAIChatSSEReleaseSink(w, w, "chatcmpl-x", time.Now().Unix(), "m")
if committed, _ := sink.terminalStatus(); committed {
t.Fatalf("sink reported a committed terminal before any commit")
}
if _, err := sink.CommitTerminal(context.Background(), successTerm); err != nil {
t.Fatalf("CommitTerminal: %v", err)
}
committed, success := sink.terminalStatus()
if !committed || !success {
t.Fatalf("sink terminalStatus after success = (%v,%v), want (true,true)", committed, success)
}
}
// Note: the separate fixed-run recovery and fixed-tunnel pass fixtures above
// do not prove the S16 provider/path-switch requirement. That evidence needs a
// provider-pool host fixture in which one RequestRuntime recovery cycle changes
// the actual AttemptBinding execution path before release.
// --- S16: provider-pool one cycle, path switch, and single re-admission ------
// buildStreamGatePoolChatFixture builds a provider-pool chat dispatch context
// bound to a scripted pool admission double, mirroring what
// handleChatCompletionsProviderPool freezes before its first SubmitProviderPool
// call. The returned pool request is the same template the recovery admission
// builder re-enters for every recovery attempt.
func buildStreamGatePoolChatFixture(t *testing.T, service runService, rawBody []byte, maxRequestFaultRecovery int) (*Server, *chatDispatchContext) {
t.Helper()
var req chatCompletionRequest
if err := json.Unmarshal(rawBody, &req); err != nil {
t.Fatalf("decode fixture body: %v", err)
}
fault := maxRequestFaultRecovery
srv := NewServer(config.EdgeOpenAIConf{
Adapter: "ollama",
Target: "llama-fixed",
TimeoutSec: 15,
StreamEvidenceGate: config.StreamEvidenceGateConf{
Enabled: true,
MaxRequestFaultRecovery: &fault,
},
}, service, nil)
route := routeDispatch{ProviderPool: true, SessionID: "cli", TimeoutSec: 15}
requestCtx := newTestRequestContext(t, route, rawBody)
basePrompt := promptFromMessages(req.Messages)
outputPolicy := srv.resolveOutputPolicy(basePrompt)
dc := srv.newChatDispatchContext(requestCtx, req, basePrompt, outputPolicy)
poolReq := edgeservice.ProviderPoolDispatchRequest{
Run: dc.submitReq,
Tunnel: edgeservice.SubmitProviderTunnelRequest{
ModelGroupKey: strings.TrimSpace(req.Model),
SessionID: route.SessionID,
Method: http.MethodPost,
Path: "/v1/chat/completions",
Stream: req.Stream,
TimeoutSec: route.TimeoutSec,
Metadata: dc.runMetadata,
EstimatedInputTokens: dc.estimate,
ContextClass: dc.contextClass,
ProviderPool: true,
},
PrepareTunnel: func(tunnelReq edgeservice.SubmitProviderTunnelRequest) (edgeservice.SubmitProviderTunnelRequest, error) {
return tunnelReq, nil
},
PrepareRun: func(runReq edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) {
return runReq, nil
},
}
poolReq.Tunnel.BuildBody = func(target string) ([]byte, error) {
body, err := dc.ingress.canonicalBody()
if err != nil {
return nil, err
}
return rewriteChatCompletionModel(body, target, req)
}
return srv, dc.withPoolDispatch(poolReq)
}
// runPoolChatStreamGate dispatches the initial pool admission exactly the way
// handleChatCompletionsProviderPool does, then drives the production
// provider-pool wiring (config, composite sink, dual event-source factory,
// recovery admission builder) through one request runtime.
func runPoolChatStreamGate(t *testing.T, srv *Server, dc *chatDispatchContext, extra ...streamgate.FilterRegistration) (*recordingResponseWriter, openAIStreamGateSink, error) {
t.Helper()
result, err := srv.service.SubmitProviderPool(context.Background(), *dc.poolDispatch)
if err != nil {
t.Fatalf("initial SubmitProviderPool: %v", err)
}
w := newRecordingResponseWriter()
cfg, sink, err := srv.newOpenAIChatPoolStreamGateConfig(w, w, dc, result, extra)
if err != nil {
t.Fatalf("newOpenAIChatPoolStreamGateConfig: %v", err)
}
rt, _, err := srv.buildOpenAIChatStreamGateRuntimeFor(dc, cfg)
if err != nil {
t.Fatalf("buildOpenAIChatStreamGateRuntimeFor: %v", err)
}
runErr := rt.Run(context.Background())
_ = rt.CloseRequestResources(context.Background(), false)
return w, sink, runErr
}
func newInjectedViolationRegistration(t *testing.T, dc *chatDispatchContext, id string, trigger int) streamgate.FilterRegistration {
t.Helper()
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
filter := newInjectedViolationFilter(t, id, trigger, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(filter, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
return reg
}
func poolChatSSEFrames(content string) chan *iop.ProviderTunnelFrame {
return bufferedTunnelFrames(
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200, Headers: map[string]string{"Content-Type": "text/event-stream"}},
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte("data: {\"choices\":[{\"delta\":{\"content\":\"" + content + "\"}}]}\n\n")},
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
)
}
type recordingOpenAIObservationSink struct {
mu sync.Mutex
got []streamgate.FilterObservation
delegate streamgate.ObservationSink
fail error
}
func (s *recordingOpenAIObservationSink) Emit(ctx context.Context, obs streamgate.FilterObservation) error {
s.mu.Lock()
s.got = append(s.got, obs)
s.mu.Unlock()
if s.delegate != nil {
if err := s.delegate.Emit(ctx, obs); err != nil {
return err
}
}
return s.fail
}
func (s *recordingOpenAIObservationSink) Snapshot() []streamgate.FilterObservation {
s.mu.Lock()
defer s.mu.Unlock()
return append([]streamgate.FilterObservation(nil), s.got...)
}
func observationKinds(observations []streamgate.FilterObservation) []streamgate.ObservationKind {
kinds := make([]streamgate.ObservationKind, 0, len(observations))
for _, obs := range observations {
kinds = append(kinds, obs.Kind())
}
return kinds
}
func assertProviderPoolObservationTimeline(t *testing.T, observations []streamgate.FilterObservation) {
t.Helper()
if len(observations) == 0 {
t.Fatal("observation timeline is empty")
}
initialTarget, err := streamgate.NewObservationAttemptTarget("client-model", "served-normalized", "prov-normalized", "normalized")
if err != nil {
t.Fatalf("initial target: %v", err)
}
reboundTarget, err := streamgate.NewObservationAttemptTarget("client-model", "served-tunnel", "prov-tunnel", "provider_tunnel")
if err != nil {
t.Fatalf("rebound target: %v", err)
}
// Per-observation invariants over the whole timeline: every row is raw-free
// valid, sequence is strictly monotonic from 1, correlation is stable, causes
// and evidence stay bounded, and the actual attempt id/target matches the
// segment the row belongs to. The segment switches to the rebound attempt at
// recovery_dispatched (inclusive), so any extra interleaved filter row is
// still attributed to the correct actual attempt.
seenDispatch := false
for i, obs := range observations {
if err := obs.Validate(); err != nil {
t.Fatalf("observation[%d] invalid: %v", i, err)
}
if obs.Sequence() != uint64(i+1) {
t.Fatalf("observation[%d] sequence=%d, want %d", i, obs.Sequence(), i+1)
}
if obs.StableCorrelation() != "req.pool-attempt-1" || obs.ConfigGeneration() != streamGateConfigGeneration {
t.Fatalf("observation[%d] correlation=(%q,%q)", i, obs.StableCorrelation(), obs.ConfigGeneration())
}
if obs.Kind() == streamgate.ObservationKindRecoveryDispatched {
seenDispatch = true
}
wantID, wantTarget := "attempt.pool-attempt-1", initialTarget
if seenDispatch {
wantID, wantTarget = "attempt.pool-attempt-2", reboundTarget
}
if obs.AttemptID() != wantID || obs.AttemptTarget() != wantTarget {
t.Fatalf("observation[%d] kind=%s attempt/target=(%q,%+v), want (%q,%+v)", i, obs.Kind(), obs.AttemptID(), obs.AttemptTarget(), wantID, wantTarget)
}
if obs.Causes().Len() > streamgate.MaxFailureCauses {
t.Fatalf("observation[%d] cause count=%d", i, obs.Causes().Len())
}
if evidence := obs.Evidence(); evidence != nil {
if err := evidence.Validate(); err != nil {
t.Fatalf("observation[%d] evidence invalid: %v", i, err)
}
}
}
// Exact chronological order: the actual normalized→tunnel recovery must
// appear as this ordered subsequence, consumed front-to-back. Extra filter
// rows (one evaluation_started/evaluated pair per registered filter) may
// interleave, but the relative order of these required rows must hold — a
// swapped recovery/terminal or a rebound segment that starts before dispatch
// can no longer pass. Each row pins the actual attempt segment, the commit
// state, and per-segment epoch consistency.
const (
segInitial = "initial"
segRebound = "rebound"
)
type wantRow struct {
kind streamgate.ObservationKind
segment string
commit streamgate.CommitState
// epochGroup buckets rows that must share the same non-zero epoch. The
// recovery rows carry the triggering (initial evaluation) epoch, so they
// share the initial group. An empty group means the row is epoch-unbound.
epochGroup string
}
uncommitted := streamgate.CommitStateTransportUncommitted
rows := []wantRow{
{streamgate.ObservationKindResponseStaged, segInitial, uncommitted, ""},
{streamgate.ObservationKindFilterEvaluationStarted, segInitial, uncommitted, "eval-initial"},
{streamgate.ObservationKindFilterEvaluated, segInitial, uncommitted, "eval-initial"},
{streamgate.ObservationKindArbitrationDecided, segInitial, uncommitted, "eval-initial"},
{streamgate.ObservationKindRecoveryPlanSelected, segInitial, uncommitted, "eval-initial"},
{streamgate.ObservationKindRecoveryAttemptAborted, segInitial, uncommitted, "eval-initial"},
{streamgate.ObservationKindRecoveryRebuilt, segInitial, uncommitted, "eval-initial"},
{streamgate.ObservationKindRecoveryDispatched, segRebound, uncommitted, "eval-initial"},
{streamgate.ObservationKindResponseStaged, segRebound, uncommitted, ""},
{streamgate.ObservationKindFilterEvaluationStarted, segRebound, uncommitted, "eval-rebound"},
{streamgate.ObservationKindFilterEvaluated, segRebound, uncommitted, "eval-rebound"},
{streamgate.ObservationKindArbitrationDecided, segRebound, uncommitted, "eval-rebound"},
{streamgate.ObservationKindReleaseCommitted, segRebound, streamgate.CommitStateStreamOpen, "eval-rebound"},
{streamgate.ObservationKindTerminalCommitted, segRebound, streamgate.CommitStateTerminalCommitted, "eval-rebound"},
}
epochByGroup := make(map[string]uint64)
idx := 0
for r, want := range rows {
match := -1
for j := idx; j < len(observations); j++ {
if observations[j].Kind() == want.kind {
match = j
break
}
}
if match < 0 {
t.Fatalf("required row %d kind=%s not found in order after index %d; timeline=%v", r, want.kind, idx, observationKinds(observations))
}
obs := observations[match]
wantID, wantTarget := "attempt.pool-attempt-1", initialTarget
if want.segment == segRebound {
wantID, wantTarget = "attempt.pool-attempt-2", reboundTarget
}
if obs.AttemptID() != wantID || obs.AttemptTarget() != wantTarget {
t.Fatalf("required row %d kind=%s attempt/target=(%q,%+v), want (%q,%+v)", r, want.kind, obs.AttemptID(), obs.AttemptTarget(), wantID, wantTarget)
}
if obs.CommitState() != want.commit {
t.Fatalf("required row %d kind=%s commit=%s, want %s", r, want.kind, obs.CommitState(), want.commit)
}
if want.epochGroup == "" {
if obs.EpochID() != 0 {
t.Fatalf("required row %d kind=%s epoch=%d, want 0 (unbound)", r, want.kind, obs.EpochID())
}
} else {
if obs.EpochID() == 0 {
t.Fatalf("required row %d kind=%s epoch=0, want non-zero epoch for group %s", r, want.kind, want.epochGroup)
}
if prev, ok := epochByGroup[want.epochGroup]; ok {
if prev != obs.EpochID() {
t.Fatalf("required row %d kind=%s epoch=%d, want %d consistent with group %s", r, want.kind, obs.EpochID(), prev, want.epochGroup)
}
} else {
epochByGroup[want.epochGroup] = obs.EpochID()
}
}
idx = match + 1
}
}
func TestStreamGateProviderPoolObservationCorrelationTimeline(t *testing.T) {
rawSentinels := []string{"SECRET_PROMPT_CONTENT", "SECRET_OUTPUT_CONTENT", "SECRET_TOOL_ARGS", "SECRET_AUTH_TOKEN", "SECRET_PREPARER_INPUT"}
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"SECRET_PROMPT_CONTENT SECRET_TOOL_ARGS SECRET_AUTH_TOKEN SECRET_PREPARER_INPUT"}],"stream":true}`)
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-1",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "SECRET_OUTPUT_CONTENT"},
&iop.RunEvent{Type: "complete"},
),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "pool-attempt-2",
provider: "prov-tunnel", target: "served-tunnel", frames: poolChatSSEFrames("switched"),
},
)
var logOutput bytes.Buffer
encoderConfig := zap.NewProductionEncoderConfig()
encoderConfig.TimeKey = ""
logger := zap.New(zapcore.NewCore(zapcore.NewJSONEncoder(encoderConfig), zapcore.AddSync(&logOutput), zapcore.InfoLevel))
recorder := &recordingOpenAIObservationSink{delegate: newZapFilterObservationSink(logger)}
srv, dc := buildStreamGatePoolChatFixture(t, service, rawBody, 3)
srv.SetObservationSink(recorder)
w, releaseSink, runErr := runPoolChatStreamGate(t, srv, dc, newInjectedViolationRegistration(t, dc, "test.injected_violation", 1))
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if !strings.Contains(w.body.String(), `"content":"switched"`) || strings.Contains(w.body.String(), "SECRET_OUTPUT_CONTENT") {
t.Fatalf("unexpected response body: %q", w.body.String())
}
composite := releaseSink.(*openAICompositeReleaseSink)
if committed, success := composite.terminalStatus(); !committed || !success {
t.Fatalf("terminal status=(%v,%v), want successful commit", committed, success)
}
observations := recorder.Snapshot()
assertProviderPoolObservationTimeline(t, observations)
eventDump, encoded := fmt.Sprintf("%+v", observations), logOutput.String()
for _, sentinel := range rawSentinels {
if strings.Contains(eventDump, sentinel) {
t.Fatalf("observation event contains raw sentinel %q", sentinel)
}
if strings.Contains(encoded, sentinel) {
t.Fatalf("observation log contains raw sentinel %q", sentinel)
}
}
for _, field := range []string{`"model_group":"client-model"`, `"actual_model":"served-normalized"`, `"actual_model":"served-tunnel"`, `"actual_provider":"prov-tunnel"`, `"execution_path":"provider_tunnel"`} {
if !strings.Contains(encoded, field) {
t.Errorf("observation log missing %s", field)
}
}
}
func TestStreamGateProviderPoolObservationSinkFailureIsolation(t *testing.T) {
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-1",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(&iop.RunEvent{Type: "delta", Delta: "discarded"}, &iop.RunEvent{Type: "complete"}),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "pool-attempt-2",
provider: "prov-tunnel", target: "served-tunnel", frames: poolChatSSEFrames("switched"),
},
)
recorder := &recordingOpenAIObservationSink{fail: errors.New("injected observation sink failure")}
srv, dc := buildStreamGatePoolChatFixture(t, service, rawBody, 3)
srv.SetObservationSink(recorder)
w, releaseSink, runErr := runPoolChatStreamGate(t, srv, dc, newInjectedViolationRegistration(t, dc, "test.injected_violation", 1))
if runErr != nil {
t.Fatalf("sink failure escaped runtime: %v", runErr)
}
if !strings.Contains(w.body.String(), `"content":"switched"`) {
t.Fatalf("successful response missing after sink failures: %q", w.body.String())
}
composite := releaseSink.(*openAICompositeReleaseSink)
if committed, success := composite.terminalStatus(); !committed || !success {
t.Fatalf("terminal status=(%v,%v), want successful commit", committed, success)
}
if service.poolSubmits() != 2 {
t.Fatalf("pool submits=%d, want 2", service.poolSubmits())
}
_, cancels, _, _, _, _ := service.snapshot()
if n := countString(cancels, "pool-attempt-1"); n != 1 {
t.Fatalf("replaced attempt canceled %d times, want exactly once", n)
}
if n := countString(cancels, "pool-attempt-2"); n > 1 {
t.Fatalf("terminal attempt canceled %d times, want at most once", n)
}
assertProviderPoolObservationTimeline(t, recorder.Snapshot())
}
func countString(values []string, want string) int {
n := 0
for _, value := range values {
if value == want {
n++
}
}
return n
}
// TestStreamGateProviderPoolRecoverySwitchesActualPathOnce is the S16
// vertical-slice evidence: one RequestRuntime admits the initial provider-pool
// attempt, a blocking violation triggers a single re-admission through
// SubmitProviderPool, the pool selects a candidate on the *other* execution
// path, and only that attempt's framing and bytes reach the caller.
func TestStreamGateProviderPoolRecoverySwitchesActualPathOnce(t *testing.T) {
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
t.Run("normalized to tunnel", func(t *testing.T) {
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-1",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "discarded-normalized"},
&iop.RunEvent{Type: "complete"},
),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "pool-attempt-2",
provider: "prov-tunnel", target: "served-tunnel",
frames: poolChatSSEFrames("switched"),
},
)
srv, dc := buildStreamGatePoolChatFixture(t, service, rawBody, 3)
w, sink, runErr := runPoolChatStreamGate(t, srv, dc, newInjectedViolationRegistration(t, dc, "test.injected_violation", 1))
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2 (initial admission + exactly one re-admission)", got)
}
composite, ok := sink.(*openAICompositeReleaseSink)
if !ok {
t.Fatalf("provider-pool request must use the composite sink, got %T", sink)
}
if got := composite.resolvedCodec(); got != openAIStreamGateCodecTunnel {
t.Fatalf("resolved codec: got %q, want tunnel framing from the recovered attempt", got)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1", got)
}
body := w.body.String()
if !strings.Contains(body, `"content":"switched"`) {
t.Fatalf("recovered tunnel attempt bytes missing: body=%q", body)
}
if strings.Contains(body, "discarded-normalized") {
t.Fatalf("discarded normalized attempt leaked into the response: body=%q", body)
}
if strings.Contains(body, "chat.completion.chunk") {
t.Fatalf("normalized SSE framing leaked into a tunnel-framed response: body=%q", body)
}
_, cancels, _, _, _, _ := service.snapshot()
if !containsString(cancels, "pool-attempt-1") {
t.Fatalf("the replaced attempt must be aborted, cancels=%v", cancels)
}
})
t.Run("tunnel to normalized", func(t *testing.T) {
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "pool-attempt-1",
provider: "prov-tunnel", target: "served-tunnel",
frames: poolChatSSEFrames("discarded-tunnel"),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-2",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "switched back"},
&iop.RunEvent{Type: "complete"},
),
},
)
srv, dc := buildStreamGatePoolChatFixture(t, service, rawBody, 3)
w, sink, runErr := runPoolChatStreamGate(t, srv, dc, newInjectedViolationRegistration(t, dc, "test.injected_violation", 1))
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2", got)
}
composite := sink.(*openAICompositeReleaseSink)
if got := composite.resolvedCodec(); got != openAIStreamGateCodecNormalized {
t.Fatalf("resolved codec: got %q, want normalized framing from the recovered attempt", got)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1", got)
}
chunks := parseSSEChatChunks(t, w.body.String())
if got := joinedContent(chunks); got != "switched back" {
t.Fatalf("joined content: got %q, want only the recovered normalized attempt's content", got)
}
if strings.Contains(w.body.String(), "discarded-tunnel") {
t.Fatalf("discarded tunnel attempt leaked into the response: body=%q", w.body.String())
}
})
}
// TestStreamGateProviderPoolExhaustionTerminatesWithoutPathLeak verifies the
// bounded budget still holds across a path switch: with a request-total cap of
// 1 the second violation terminates instead of re-admitting a third candidate,
// and no attempt's content is released.
func TestStreamGateProviderPoolExhaustionTerminatesWithoutPathLeak(t *testing.T) {
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-1",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(&iop.RunEvent{Type: "delta", Delta: "still-bad-1"}, &iop.RunEvent{Type: "complete"}),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "pool-attempt-2",
provider: "prov-tunnel", target: "served-tunnel",
frames: poolChatSSEFrames("still-bad-2"),
},
)
srv, dc := buildStreamGatePoolChatFixture(t, service, rawBody, 1)
w, _, runErr := runPoolChatStreamGate(t, srv, dc, newInjectedViolationRegistration(t, dc, "test.injected_violation", 2))
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2 (cap=1 allows exactly one re-admission)", got)
}
body := w.body.String()
if !strings.Contains(body, `"error"`) {
t.Fatalf("expected an error terminal once the recovery budget is exhausted, body=%q", body)
}
for _, leak := range []string{"still-bad-1", "still-bad-2"} {
if strings.Contains(body, leak) {
t.Fatalf("exhausted recovery leaked attempt content %q: body=%q", leak, body)
}
}
}
// TestStreamGateProviderPoolSimultaneousViolationsDispatchOnce verifies that two
// independent blocking filters violating on the same terminal produce exactly
// one recovery plan and one re-admission, not one per filter.
func TestStreamGateProviderPoolSimultaneousViolationsDispatchOnce(t *testing.T) {
rawBody := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hi"}],"stream":true}`)
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-1",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(&iop.RunEvent{Type: "delta", Delta: "discarded"}, &iop.RunEvent{Type: "complete"}),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-attempt-2",
provider: "prov-normalized", target: "served-normalized",
runEvents: bufferedRunEvents(&iop.RunEvent{Type: "delta", Delta: "final answer"}, &iop.RunEvent{Type: "complete"}),
},
)
srv, dc := buildStreamGatePoolChatFixture(t, service, rawBody, 3)
w, _, runErr := runPoolChatStreamGate(t, srv, dc,
newInjectedViolationRegistration(t, dc, "test.injected_violation_a", 1),
newInjectedViolationRegistration(t, dc, "test.injected_violation_b", 1),
)
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2 (two simultaneous violations must share one plan/dispatch)", got)
}
chunks := parseSSEChatChunks(t, w.body.String())
if got := joinedContent(chunks); got != "final answer" {
t.Fatalf("joined content: got %q, want the single recovered attempt's content", got)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1", got)
}
}
// --- S16: host recovery preparer lifecycle ----------------------------------
// recordingPreparer is a test-only host RecoveryPlanPreparer. Production OpenAI
// surfaces register no preparer; this fixture exercises the host wiring of the
// optional one-shot preparation seam (S23): it must run at most once per plan,
// after attempt ownership is closed and before the rebuild/dispatch.
type recordingPreparer struct {
mu sync.Mutex
calls int
failWith error
directive streamgate.RecoveryDirective
}
func (p *recordingPreparer) PrepareRecoveryPlan(ctx context.Context, plan streamgate.RecoveryPlan, snapshot streamgate.RecoveryPreparationSnapshot) (streamgate.RecoveryDirective, error) {
p.mu.Lock()
p.calls++
failWith := p.failWith
directive := p.directive
p.mu.Unlock()
if failWith != nil {
return streamgate.RecoveryDirective{}, failWith
}
return directive, nil
}
func (p *recordingPreparer) callCount() int {
p.mu.Lock()
defer p.mu.Unlock()
return p.calls
}
// recordingPreparationSnapshot is the minimal bounded snapshot seam; it only
// records that Core released it exactly once.
type recordingPreparationSnapshot struct {
ref string
releases *int32
}
func (s *recordingPreparationSnapshot) SnapshotRef() string { return s.ref }
func (s *recordingPreparationSnapshot) Release() error {
atomic.AddInt32(s.releases, 1)
return nil
}
type recordingPreparationSnapshotFactory struct {
ref string
creates *int32
releases *int32
}
func (f *recordingPreparationSnapshotFactory) CreatePreparationSnapshot(ctx context.Context, snapshot streamgate.RequestRuntimeSnapshot, binding streamgate.AttemptBinding) (streamgate.RecoveryPreparationSnapshot, error) {
atomic.AddInt32(f.creates, 1)
return &recordingPreparationSnapshot{ref: f.ref, releases: f.releases}, nil
}
// TestStreamGateChatRecoveryPreparerLifecycle pins the host preparer contract
// for one recovery cycle: on success the prepared directive is what gets
// rebuilt and dispatched exactly once, and on failure the cycle stops before
// any dispatch and converges on a single error terminal with no leaked content.
func TestStreamGateChatRecoveryPreparerLifecycle(t *testing.T) {
for _, tc := range []struct {
name string
preparerFail bool
wantDispatch int
}{
{"preparer success dispatches once", false, 1},
{"preparer failure dispatches nothing", true, 0},
} {
tc := tc
t.Run(tc.name, func(t *testing.T) {
srv, dc, initial, fake := buildStreamGateChatFixture(t, true, 3)
fake.eventRuns = []chan *iop.RunEvent{bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "final answer"},
&iop.RunEvent{Type: "complete"},
)}
fake.runIDs = []string{"run-attempt-2"}
snapRef, err := dc.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
directive, err := streamgate.NewRecoveryDirectiveExact(snapRef.SnapshotRef())
if err != nil {
t.Fatalf("NewRecoveryDirectiveExact: %v", err)
}
preparer := &recordingPreparer{directive: directive}
if tc.preparerFail {
preparer.failWith = fmt.Errorf("injected preparer failure")
}
var creates, releases int32
factory := &recordingPreparationSnapshotFactory{ref: "openai.prep.1", creates: &creates, releases: &releases}
registry := streamGateTestRegistry(t, newInjectedViolationRegistration(t, dc, "test.injected_violation", 1))
w := newRecordingResponseWriter()
sink := newOpenAIChatSSEReleaseSink(w, w, "chatcmpl-run-attempt-1", time.Now().Unix(), "client-model")
rt, _, err := srv.buildOpenAIChatStreamGateRuntimeFor(dc, openAIChatStreamGateConfig{
mode: openAIChatGateModeLive,
initial: openAIAttemptTransport{path: openAIAdmissionRun, run: initial},
dispatch: initial.Dispatch(),
closeAll: initial.Close,
sink: sink,
selector: newOpenAIStreamGateCodecSelector(openAIStreamGateCodecNormalized),
registry: registry,
preparer: preparer,
prepFactory: factory,
})
if err != nil {
t.Fatalf("buildOpenAIChatStreamGateRuntimeFor: %v", err)
}
if err := rt.Run(context.Background()); err != nil {
t.Fatalf("rt.Run: %v", err)
}
if got := preparer.callCount(); got != 1 {
t.Fatalf("preparer calls: got %d, want exactly 1 per recovery plan", got)
}
if got := atomic.LoadInt32(&creates); got != 1 {
t.Fatalf("preparation snapshot creations: got %d, want 1", got)
}
if got := atomic.LoadInt32(&releases); got != 1 {
t.Fatalf("preparation snapshot releases: got %d, want exactly 1", got)
}
if got := len(fake.reqsSnapshot()); got != tc.wantDispatch {
t.Fatalf("recovery dispatches: got %d, want %d", got, tc.wantDispatch)
}
body := w.body.String()
if strings.Contains(body, "should-be-discarded") {
t.Fatalf("the aborted attempt leaked into the response: %s", body)
}
if tc.preparerFail {
if !strings.Contains(body, `"error"`) {
t.Fatalf("a failed preparation must terminate with an error, body=%s", body)
}
return
}
if got := joinedContent(parseSSEChatChunks(t, body)); got != "final answer" {
t.Fatalf("joined content: got %q, want the prepared recovery attempt's content", got)
}
})
}
}
// --- S16: producer backpressure while an evaluator holds --------------------
// blockingEvaluationFilter blocks inside Evaluate until it is released, so a
// fixture can observe that the unbuffered provider producer cannot advance
// while the Core is evaluating an epoch.
type blockingEvaluationFilter struct {
streamgate.FilterBase
entered chan struct{}
release chan struct{}
once sync.Once
}
func newBlockingEvaluationFilter(t *testing.T, id string) *blockingEvaluationFilter {
t.Helper()
base, err := streamgate.NewFilterBase(id)
if err != nil {
t.Fatalf("NewFilterBase: %v", err)
}
return &blockingEvaluationFilter{
FilterBase: base,
entered: make(chan struct{}, 1),
release: make(chan struct{}),
}
}
func (f *blockingEvaluationFilter) Applies(streamgate.FilterContext) bool { return true }
func (f *blockingEvaluationFilter) HoldRequirement(streamgate.FilterContext) streamgate.FilterHoldRequirement {
req, err := streamgate.NewFilterHoldRequirementRolling(streamGateChannelDefault, []streamgate.EventKind{streamgate.EventKindTextDelta}, 1)
if err != nil {
panic(err)
}
return req
}
func (f *blockingEvaluationFilter) Evaluate(ctx context.Context, fctx streamgate.FilterContext, batch streamgate.EvidenceBatch) (streamgate.FilterDecision, error) {
f.once.Do(func() {
f.entered <- struct{}{}
select {
case <-f.release:
case <-ctx.Done():
}
})
kind := streamgate.EventKindTextDelta
if events := batch.Events(); len(events) > 0 {
kind = events[len(events)-1].Kind()
}
evidence, err := streamgate.NewSanitizedEvidence(kind, streamGateChannelDefault, "test.rule", "test_blocking_eval", streamgate.FixedFingerprint{0x03}, 1, 0, streamgate.FilterOutcomeKindEvaluated, batch.CapturedAt())
if err != nil {
return streamgate.FilterDecision{}, err
}
return streamgate.NewFilterDecision(streamgate.FilterDecisionKindPass, "test.consumer", f.ID(), "test.rule", evidence, nil)
}
var _ streamgate.Filter = (*blockingEvaluationFilter)(nil)
// TestStreamGateChatProducerBackpressureDuringEvaluation verifies the Core
// applies real backpressure: while a blocking filter is still evaluating the
// first epoch, the unbuffered provider producer cannot complete its next send,
// and it resumes only once the evaluation is released.
func TestStreamGateChatProducerBackpressureDuringEvaluation(t *testing.T) {
srv, dc, initial, _ := buildStreamGateChatFixture(t, true, 3)
events := make(chan *iop.RunEvent) // unbuffered: the producer blocks unless the source is reading
initial.events = events
sent := make(chan int, 3)
go func() {
for i, ev := range []*iop.RunEvent{
{Type: "delta", Delta: "one "},
{Type: "delta", Delta: "two"},
{Type: "complete"},
} {
events <- ev
sent <- i + 1
}
}()
blocking := newBlockingEvaluationFilter(t, "test.blocking_eval")
reg, err := streamgate.NewFilterRegistration(blocking, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, 5)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
registry := streamGateTestRegistry(t, reg)
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)
}
done := make(chan error, 1)
go func() { done <- rt.Run(context.Background()) }()
select {
case <-blocking.entered:
case <-time.After(5 * time.Second):
t.Fatal("the blocking evaluator was never entered")
}
// The first event was consumed by the event source; the second send must be
// parked because the runtime is inside the evaluator and not reading.
select {
case got := <-sent:
if got != 1 {
t.Fatalf("first completed send: got %d, want 1", got)
}
case <-time.After(5 * time.Second):
t.Fatal("the first send never completed")
}
select {
case got := <-sent:
t.Fatalf("send %d completed while the evaluator still held the epoch: producer backpressure is missing", got)
case <-time.After(150 * time.Millisecond):
}
close(blocking.release)
select {
case runErr := <-done:
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
case <-time.After(10 * time.Second):
t.Fatal("rt.Run did not finish after the evaluation was released")
}
for want := 2; want <= 3; want++ {
select {
case got := <-sent:
if got != want {
t.Fatalf("resumed send: got %d, want %d", got, want)
}
case <-time.After(5 * time.Second):
t.Fatalf("send %d never completed after the evaluation was released", want)
}
}
if got := joinedContent(parseSSEChatChunks(t, w.body.String())); got != "one two" {
t.Fatalf("joined content: got %q, want %q", got, "one two")
}
}
// --- S16: Responses provider-pool tunnel recovery ---------------------------
// buildStreamGateResponsesPoolFixture builds a streaming /v1/responses
// provider-pool passthrough request bound to a scripted pool double, mirroring
// what handleResponsesProviderPool freezes before its first SubmitProviderPool.
func buildStreamGateResponsesPoolFixture(t *testing.T, service runService, maxRequestFaultRecovery int) (*Server, *responsesRequestContext, edgeservice.ProviderPoolDispatchRequest) {
t.Helper()
rawBody := []byte(`{"model":"client-model","input":"hi","stream":true}`)
fault := maxRequestFaultRecovery
srv := NewServer(config.EdgeOpenAIConf{
Adapter: "openai-compat",
Target: "served-model",
TimeoutSec: 15,
StreamEvidenceGate: config.StreamEvidenceGateConf{
Enabled: true,
MaxRequestFaultRecovery: &fault,
},
}, service, nil)
route := routeDispatch{ProviderPool: true, TimeoutSec: 15}
base := newTestRequestContext(t, route, rawBody)
base.endpoint = usageEndpointResponses
requestCtx := &responsesRequestContext{
openAIRequestContext: base,
envelope: responsesEnvelope{Model: "client-model", Stream: true},
}
metadata := map[string]string{"openai_model": "client-model", "openai_stream": "true"}
poolReq := edgeservice.ProviderPoolDispatchRequest{
Run: edgeservice.SubmitRunRequest{
ModelGroupKey: "client-model",
Metadata: metadata,
ProviderPool: true,
},
Tunnel: edgeservice.SubmitProviderTunnelRequest{
ModelGroupKey: "client-model",
Method: http.MethodPost,
Path: "/v1/responses",
Stream: true,
TimeoutSec: 15,
Metadata: metadata,
ProviderPool: true,
},
PrepareTunnel: func(tunnelReq edgeservice.SubmitProviderTunnelRequest) (edgeservice.SubmitProviderTunnelRequest, error) {
return tunnelReq, nil
},
PrepareRun: func(runReq edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) {
return runReq, nil
},
}
poolReq.Tunnel.BuildBody = func(target string) ([]byte, error) {
body, err := requestCtx.ingress.canonicalBody()
if err != nil {
return nil, err
}
return rewriteResponsesModel(body, target)
}
return srv, requestCtx, poolReq
}
func responsesTunnelFrames(payload string) chan *iop.ProviderTunnelFrame {
return bufferedTunnelFrames(
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200, Headers: map[string]string{"Content-Type": "text/event-stream"}},
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte(payload)},
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
)
}
// TestStreamGateResponsesPoolRecoveryTerminal covers the Responses passthrough
// contract for a provider-pool recovery: the re-admission goes back through
// SubmitProviderPool, a tunnel candidate is relayed byte-for-byte, and a
// normalized candidate — which a pure passthrough surface cannot render —
// converges on a safe error terminal instead of re-framing the response.
func TestStreamGateResponsesPoolRecoveryTerminal(t *testing.T) {
t.Run("tunnel candidate is relayed after re-admission", func(t *testing.T) {
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "resp-attempt-1",
provider: "prov-a", target: "served-a",
frames: responsesTunnelFrames("data: {\"type\":\"response.output_text.delta\",\"delta\":\"discarded\"}\n\n"),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "resp-attempt-2",
provider: "prov-b", target: "served-b",
frames: responsesTunnelFrames("data: {\"type\":\"response.output_text.delta\",\"delta\":\"recovered\"}\n\n"),
},
)
srv, requestCtx, poolReq := buildStreamGateResponsesPoolFixture(t, service, 3)
result, err := service.SubmitProviderPool(context.Background(), poolReq)
if err != nil {
t.Fatalf("initial SubmitProviderPool: %v", err)
}
snapRef, err := requestCtx.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
violation := newInjectedViolationFilter(t, "test.injected_violation", 1, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
w := newRecordingResponseWriter()
sink := newOpenAITunnelReleaseSink(w, w)
rt, _, err := srv.buildOpenAITunnelStreamGateRuntime(
srv.openAIResponsesPoolTunnelStreamGateRequest(requestCtx, poolReq, poolReq.Tunnel.Metadata),
result.Tunnel, sink, streamGateTestRegistry(t, reg),
)
if err != nil {
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
}
if err := rt.Run(context.Background()); err != nil {
t.Fatalf("rt.Run: %v", err)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2 (initial + exactly one re-admission)", got)
}
body := w.body.String()
if !strings.Contains(body, "recovered") {
t.Fatalf("recovered tunnel bytes missing: %q", body)
}
if strings.Contains(body, "discarded") {
t.Fatalf("the discarded attempt leaked into the response: %q", body)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1", got)
}
})
t.Run("normalized candidate terminates safely", func(t *testing.T) {
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "resp-attempt-1",
provider: "prov-a", target: "served-a",
frames: responsesTunnelFrames("data: {\"type\":\"response.output_text.delta\",\"delta\":\"discarded\"}\n\n"),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "resp-attempt-2",
provider: "prov-b", target: "served-b",
runEvents: bufferedRunEvents(&iop.RunEvent{Type: "delta", Delta: "normalized"}, &iop.RunEvent{Type: "complete"}),
},
)
srv, requestCtx, poolReq := buildStreamGateResponsesPoolFixture(t, service, 3)
result, err := service.SubmitProviderPool(context.Background(), poolReq)
if err != nil {
t.Fatalf("initial SubmitProviderPool: %v", err)
}
snapRef, err := requestCtx.ingress.recoveryRef()
if err != nil {
t.Fatalf("recoveryRef: %v", err)
}
violation := newInjectedViolationFilter(t, "test.injected_violation", 1, snapRef.SnapshotRef())
reg, err := streamgate.NewFilterRegistration(violation, streamGateNoopCapability, true, streamgate.FilterEnforcementBlocking, streamGateFilterTimeout, injectedViolationPriority)
if err != nil {
t.Fatalf("NewFilterRegistration: %v", err)
}
w := newRecordingResponseWriter()
sink := newOpenAITunnelReleaseSink(w, w)
rt, _, err := srv.buildOpenAITunnelStreamGateRuntime(
srv.openAIResponsesPoolTunnelStreamGateRequest(requestCtx, poolReq, poolReq.Tunnel.Metadata),
result.Tunnel, sink, streamGateTestRegistry(t, reg),
)
if err != nil {
t.Fatalf("buildOpenAITunnelStreamGateRuntime: %v", err)
}
if err := rt.Run(context.Background()); err != nil {
t.Fatalf("rt.Run: %v", err)
}
committed, success := sink.terminalStatus()
if !committed || success {
t.Fatalf("terminal status: got (committed=%v, success=%v), want a committed error terminal", committed, success)
}
body := w.body.String()
if strings.Contains(body, "discarded") || strings.Contains(body, "normalized") {
t.Fatalf("no attempt content may reach the caller on a rejected path switch: %q", body)
}
// The rejected normalized attempt must not be left running.
_, cancels, _, _, _, _ := service.snapshot()
if !containsString(cancels, "resp-attempt-2") {
t.Fatalf("the rejected normalized attempt must be aborted, cancels=%v", cancels)
}
})
}
// --- S16: provider-pool + tool validation (one owner across a path switch) ---
const streamGatePoolToolBody = `{"model":"client-model","messages":[{"role":"user","content":"status"}],` +
`"tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object","properties":{"commands":{"type":"array","items":{"type":"string"}}}}}}]}`
// TestStreamGatePoolToolValidationReAdmitsThroughPool proves the tool-validation
// terminal gate and the provider-pool re-admission are the same single cycle:
// an invalid tool call on a normalized pool candidate re-enters
// SubmitProviderPool once, and the passing candidate's structured output is the
// only thing rendered.
func TestStreamGatePoolToolValidationReAdmitsThroughPool(t *testing.T) {
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-tool-1",
provider: "prov-a", target: "served-a",
runEvents: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "true"},
&iop.RunEvent{Type: "complete"},
),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-tool-2",
provider: "prov-b", target: "served-b",
runEvents: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "[\"ls\"]\n"},
&iop.RunEvent{Type: "complete"},
),
},
)
srv, dc := buildStreamGatePoolChatFixture(t, service, []byte(streamGatePoolToolBody), 1)
if !dc.validation.enabled {
t.Fatal("fixture must carry a tool-validation contract")
}
w, sink, runErr := runPoolChatStreamGate(t, srv, dc)
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2 (tool-validation recovery must re-enter the pool)", got)
}
if got := sink.(*openAICompositeReleaseSink).resolvedCodec(); got != openAIStreamGateCodecNormalized {
t.Fatalf("resolved codec: got %q, want normalized", got)
}
var resp chatCompletionResponse
if err := json.Unmarshal([]byte(w.body.String()), &resp); err != nil {
t.Fatalf("decode response: %v body=%s", err, w.body.String())
}
if len(resp.Choices) != 1 || resp.Choices[0].FinishReason != "tool_calls" {
t.Fatalf("expected a structured tool_calls completion, got %s", w.body.String())
}
if len(resp.Choices[0].Message.ToolCalls) != 1 {
t.Fatalf("tool calls: got %d, want 1: %s", len(resp.Choices[0].Message.ToolCalls), w.body.String())
}
if strings.Contains(w.body.String(), "delete_everything") {
t.Fatalf("the rejected attempt leaked into the response: %s", w.body.String())
}
}
// TestStreamGatePoolToolValidationDropsOnTunnelCandidate verifies the
// tool-validation contract is re-resolved per attempt: after a recovery selects
// a raw passthrough candidate the normalized contract no longer applies, and
// the response is relayed with the tunnel framing instead of being validated
// again.
func TestStreamGatePoolToolValidationDropsOnTunnelCandidate(t *testing.T) {
service := newScriptedPoolRunService(
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathNormalized), runID: "pool-tool-1",
provider: "prov-a", target: "served-a",
runEvents: bufferedRunEvents(
&iop.RunEvent{Type: "delta", Delta: "true"},
&iop.RunEvent{Type: "complete"},
),
},
scriptedPoolAttempt{
path: string(edgeservice.ProviderPoolPathTunnel), runID: "pool-tool-2",
provider: "prov-b", target: "served-b",
frames: bufferedTunnelFrames(
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200, Headers: map[string]string{"Content-Type": "application/json"}},
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte(`{"id":"provider-native","object":"chat.completion"}`)},
&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
),
},
)
srv, dc := buildStreamGatePoolChatFixture(t, service, []byte(streamGatePoolToolBody), 1)
w, sink, runErr := runPoolChatStreamGate(t, srv, dc)
if runErr != nil {
t.Fatalf("rt.Run: %v", runErr)
}
if got := service.poolSubmits(); got != 2 {
t.Fatalf("SubmitProviderPool calls: got %d, want 2", got)
}
if got := sink.(*openAICompositeReleaseSink).resolvedCodec(); got != openAIStreamGateCodecTunnel {
t.Fatalf("resolved codec: got %q, want tunnel", got)
}
body := w.body.String()
if !strings.Contains(body, `"id":"provider-native"`) {
t.Fatalf("raw passthrough bytes missing: %q", body)
}
if strings.Contains(body, "delete_everything") {
t.Fatalf("the rejected normalized attempt leaked into the response: %q", body)
}
if got := w.headerCallCount(); got != 1 {
t.Fatalf("WriteHeader call count: got %d, want 1", got)
}
}