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