package openai

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/config"
	"iop/packages/go/streamgate"
	iop "iop/proto/gen/iop"
)

func confirmedStallFailure(health string) *iop.ExecutionFailure {
	return &iop.ExecutionFailure{
		Code:      openAIStallFailureCode,
		Retryable: true,
		Message:   "provider body, prompt, and credentials must not escape",
		Metadata: map[string]string{
			"failure_code":     openAIStallFailureCode,
			"attempt_fence":    "confirmed",
			"recovery_handoff": "confirmed",
			"provider_id":      "provider-a",
			"provider_health":  health,
			"untrusted":        "raw provider metadata",
		},
	}
}

func stallFilterContext(t *testing.T, commit streamgate.CommitState, sideEffect bool) streamgate.FilterContext {
	t.Helper()
	ctx, err := streamgate.NewFilterContextBuilder(streamGateConfigGeneration, "attempt.run-a").
		SetEndpoint(openAIRebuildEndpointChat).
		SetActualProvider("provider-a").
		SetCommitState(commit).
		SetHasToolSideEffect(sideEffect).
		Build()
	if err != nil {
		t.Fatalf("build filter context: %v", err)
	}
	return ctx
}

func stallBatch(t *testing.T, event streamgate.NormalizedEvent, commit streamgate.CommitState, pending ...streamgate.NormalizedEvent) streamgate.EvidenceBatch {
	t.Helper()
	batch, err := streamgate.NewEvidenceBatch([]streamgate.NormalizedEvent{event}, map[string][]streamgate.NormalizedEvent{streamGateChannelDefault: pending}, nil, nil, true, commit, time.Now())
	if err != nil {
		t.Fatalf("build stall batch: %v", err)
	}
	return batch
}

func TestOpenAIStallEventMapping(t *testing.T) {
	event, err := newOpenAIProviderErrorEventFromFailure(confirmedStallFailure("unknown"), streamGateErrorRunFailed)
	if err != nil {
		t.Fatalf("map confirmed stall: %v", err)
	}
	terminal, err := event.AsProviderError()
	if err != nil {
		t.Fatalf("AsProviderError: %v", err)
	}
	if desc := terminal.ExternalDesc(); desc == nil || desc.Code() != openAIStallFailureCode || desc.Message() != openAIStallFailureCode {
		t.Fatalf("descriptor = %#v", desc)
	}
	for _, cause := range terminal.FailureCauses().All() {
		if cause.Code() == "provider body, prompt, and credentials must not escape" || cause.Code() == "raw provider metadata" {
			t.Fatalf("raw failure data leaked into causes: %#v", cause)
		}
	}

	generic, err := newOpenAIProviderErrorEventFromFailure(&iop.ExecutionFailure{Code: "other", Message: "raw"}, streamGateErrorRunFailed)
	if err != nil {
		t.Fatalf("map generic failure: %v", err)
	}
	genericTerminal, _ := generic.AsProviderError()
	if got := genericTerminal.ExternalDesc().Code(); got != streamGateErrorRunFailed {
		t.Fatalf("generic descriptor code = %q", got)
	}
}

func TestOpenAIStallRecoveryFilter(t *testing.T) {
	for _, health := range []string{"available", "unavailable", "unknown"} {
		t.Run(health, func(t *testing.T) {
			state := &openAIStallRecoveryState{}
			filter, err := newOpenAIStallRecoveryFilter("openai.ingress.1", state)
			if err != nil {
				t.Fatalf("new filter: %v", err)
			}
			event, err := newOpenAIProviderErrorEventFromFailure(confirmedStallFailure(health), streamGateErrorRunFailed)
			if err != nil {
				t.Fatalf("map failure: %v", err)
			}
			decision, err := filter.Evaluate(context.Background(), stallFilterContext(t, streamgate.CommitStateTransportUncommitted, false), stallBatch(t, event, streamgate.CommitStateTransportUncommitted))
			if err != nil {
				t.Fatalf("evaluate: %v", err)
			}
			if decision.Kind() != streamgate.FilterDecisionKindViolation || decision.RecoveryIntent() == nil {
				t.Fatalf("decision = %#v", decision)
			}
			if !state.claimConfirmedClose("attempt.run-a") {
				t.Fatal("confirmed state was not armed")
			}
			provider, fallback, ok := state.consumeAdmission()
			if !ok || provider != "provider-a" || fallback {
				t.Fatalf("admission hint = %q/%t/%t", provider, fallback, ok)
			}
		})
	}
}

func TestOpenAIStallRecoveryIneligibleAfterCommitOrTool(t *testing.T) {
	event, err := newOpenAIProviderErrorEventFromFailure(confirmedStallFailure("available"), streamGateErrorRunFailed)
	if err != nil {
		t.Fatal(err)
	}
	for _, tc := range []struct {
		name       string
		commit     streamgate.CommitState
		sideEffect bool
	}{
		{"post_commit", streamgate.CommitStateStreamOpen, false},
		{"tool_side_effect", streamgate.CommitStateTransportUncommitted, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			state := &openAIStallRecoveryState{}
			filter, _ := newOpenAIStallRecoveryFilter("openai.ingress.1", state)
			decision, err := filter.Evaluate(context.Background(), stallFilterContext(t, tc.commit, tc.sideEffect), stallBatch(t, event, tc.commit))
			if err != nil {
				t.Fatal(err)
			}
			if decision.Kind() != streamgate.FilterDecisionKindPass || decision.RecoveryIntent() != nil {
				t.Fatalf("unsafe decision = %#v", decision)
			}
		})
	}
}

func TestOpenAIStallRecoveryIsBoundedToOneReplay(t *testing.T) {
	state := &openAIStallRecoveryState{}
	filter, err := newOpenAIStallRecoveryFilter("openai.ingress.1", state)
	if err != nil {
		t.Fatal(err)
	}
	event, err := newOpenAIProviderErrorEventFromFailure(confirmedStallFailure("available"), streamGateErrorRunFailed)
	if err != nil {
		t.Fatal(err)
	}
	ctx := stallFilterContext(t, streamgate.CommitStateTransportUncommitted, false)
	batch := stallBatch(t, event, streamgate.CommitStateTransportUncommitted)
	first, err := filter.Evaluate(context.Background(), ctx, batch)
	if err != nil {
		t.Fatal(err)
	}
	second, err := filter.Evaluate(context.Background(), ctx, batch)
	if err != nil {
		t.Fatal(err)
	}
	if first.Kind() != streamgate.FilterDecisionKindViolation || first.RecoveryIntent() == nil {
		t.Fatalf("first stall decision = %#v", first)
	}
	if second.Kind() != streamgate.FilterDecisionKindPass || second.RecoveryIntent() != nil {
		t.Fatalf("second stall decision = %#v", second)
	}
}

func stallMatrixSuccessAttempt(endpoint, path string, stream bool, runID, provider, marker string) scriptedPoolAttempt {
	attempt := scriptedPoolAttempt{path: path, runID: runID, provider: provider, target: "served-" + provider}
	if path == string(edgeservice.ProviderPoolPathNormalized) {
		attempt.runEvents = bufferedRunEvents(
			&iop.RunEvent{RunId: runID, Type: "delta", Delta: marker},
			&iop.RunEvent{RunId: runID, Type: "complete", Metadata: map[string]string{"finish_reason": "stop"}},
		)
		return attempt
	}
	body := []byte(fmt.Sprintf(`{"id":"chat-recovered","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":%q},"finish_reason":"stop"}]}`, marker))
	if endpoint == openAIRebuildEndpointResponses {
		if stream {
			body = []byte(fmt.Sprintf("data: {\"type\":\"response.output_text.delta\",\"delta\":%q}\n\ndata: {\"type\":\"response.completed\"}\n\ndata: [DONE]\n\n", marker))
		} else {
			body = []byte(fmt.Sprintf(`{"id":"resp-recovered","object":"response","status":"completed","output_text":%q,"output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":%q}]}]}`, marker, marker))
		}
	}
	contentType := "application/json"
	if stream {
		contentType = "text/event-stream"
	}
	attempt.frames = bufferedTunnelFrames(
		&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": contentType}},
		&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: body},
		&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
	)
	return attempt
}

func stallMatrixFailureAttempt(path, runID, provider, health string) scriptedPoolAttempt {
	attempt := scriptedPoolAttempt{path: path, runID: runID, provider: provider, target: "served-" + provider}
	if path == string(edgeservice.ProviderPoolPathNormalized) {
		attempt.runEvents = bufferedRunEvents(&iop.RunEvent{RunId: runID, Type: "error", Failure: confirmedStallFailure(health)})
	} else {
		attempt.frames = bufferedTunnelFrames(&iop.ProviderTunnelFrame{RunId: runID, Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: confirmedStallFailure(health)})
	}
	return attempt
}

func stallMatrixServer(service runService, semantic bool, budget int) *Server {
	srv := NewServer(config.EdgeOpenAIConf{
		TimeoutSec: 5,
		StreamEvidenceGate: config.StreamEvidenceGateConf{
			Enabled: semantic, MaxRequestFaultRecovery: &budget,
		},
	}, service, nil)
	srv.SetModelCatalog([]config.ModelCatalogEntry{{
		ID: "matrix-model", Providers: map[string]string{"provider-a": "served-a", "provider-b": "served-b"},
	}})
	return srv
}

func runStallMatrixHandler(t *testing.T, srv *Server, endpoint string, stream bool, ctx context.Context, bodyOverride ...string) *httptest.ResponseRecorder {
	t.Helper()
	path := "/v1/chat/completions"
	body := fmt.Sprintf(`{"model":"matrix-model","stream":%t,"messages":[{"role":"user","content":"hi"}]}`, stream)
	if endpoint == openAIRebuildEndpointResponses {
		path = "/v1/responses"
		body = fmt.Sprintf(`{"model":"matrix-model","stream":%t,"input":"hi"}`, stream)
	}
	if len(bodyOverride) > 0 && bodyOverride[0] != "" {
		body = bodyOverride[0]
	}
	r := httptest.NewRequest(http.MethodPost, path, strings.NewReader(body))
	if ctx != nil {
		r = r.WithContext(ctx)
	}
	w := httptest.NewRecorder()
	if endpoint == openAIRebuildEndpointChat {
		srv.handleChatCompletions(w, r)
	} else {
		srv.handleResponses(w, r)
	}
	return w
}

func stallPoolRequests(service *scriptedPoolRunService) []edgeservice.ProviderPoolDispatchRequest {
	service.mu.Lock()
	defer service.mu.Unlock()
	return append([]edgeservice.ProviderPoolDispatchRequest(nil), service.poolRequests...)
}

func countStallMatrixString(values []string, want string) int {
	count := 0
	for _, value := range values {
		if value == want {
			count++
		}
	}
	return count
}

func assertStallAttemptClosedOnce(t *testing.T, service *scriptedPoolRunService, path, runID string) {
	t.Helper()
	_, _, runCloses, tunnelCloses, _, _ := service.snapshot()
	closes := runCloses
	if path == string(edgeservice.ProviderPoolPathTunnel) {
		closes = tunnelCloses
	}
	if got := countStallMatrixString(closes, runID); got != 1 {
		t.Fatalf("transport close count for %s = %d, want 1 (run=%v tunnel=%v)", runID, got, runCloses, tunnelCloses)
	}
}

func logicalFinishStallAttempt(runID, content string, committed bool) scriptedPoolAttempt {
	wire := []byte(fmt.Sprintf("data: {\"id\":\"logical-finish\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":%q},\"finish_reason\":\"stop\"}]}\n\n", content))
	if committed {
		// The first content frame opens the stream. The following finish-only
		// frame has no semantic event, so its identifiable wire remains pending
		// until [DONE]/END and must be discarded by the later stall terminal.
		wire = append(
			[]byte("data: {\"id\":\"logical-finish\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"committed-prefix\"},\"finish_reason\":null}]}\n\n"),
			[]byte(fmt.Sprintf("data: {\"id\":%q,\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n", content))...,
		)
	}
	return scriptedPoolAttempt{
		path: string(edgeservice.ProviderPoolPathTunnel), runID: runID, provider: "provider-a", target: "served-a",
		frames: bufferedTunnelFrames(
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "text/event-stream"}},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: wire},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: confirmedStallFailure("available")},
		),
	}
}

func logicalFinishMatrixServer(service runService, holdRunes int) *Server {
	budget := 1
	srv := NewServer(config.EdgeOpenAIConf{
		TimeoutSec: 5,
		StreamEvidenceGate: config.StreamEvidenceGateConf{
			Enabled: true, MaxRequestFaultRecovery: &budget,
			Filters: []config.StreamGateFilterPolicyConf{{
				Filter: config.StreamGateFilterRepeatGuard, Enforcement: config.StreamGateFilterEnforcementBlocking,
				Priority: 10, HoldEvidenceRunes: holdRunes,
			}},
		},
	}, service, nil)
	srv.SetModelCatalog([]config.ModelCatalogEntry{{
		ID: "matrix-model", ContextWindowTokens: 8192,
		Providers: map[string]string{"provider-a": "served-a", "provider-b": "served-b"},
	}})
	return srv
}

// TestOpenAIStallAfterLogicalFinishMatrix drives the observed shape through
// the production tunnel codec/Core/sink: finish_reason is pending protocol
// wire, while only the typed stall terminal decides replay or sanitized close.
func TestOpenAIStallAfterLogicalFinishMatrix(t *testing.T) {
	t.Run("pre-commit pending finish replays once", func(t *testing.T) {
		const pending = "pending-finish-must-stay-hidden"
		const recovered = "recovered-output-only"
		service := newScriptedPoolRunService(
			logicalFinishStallAttempt("logical-pre-a", pending, false),
			stallMatrixSuccessAttempt(openAIRebuildEndpointChat, string(edgeservice.ProviderPoolPathTunnel), true, "logical-pre-b", "provider-b", recovered),
		)
		w := runStallMatrixHandler(t, logicalFinishMatrixServer(service, 500), openAIRebuildEndpointChat, true, nil)
		body := w.Body.String()
		if w.Code != http.StatusOK || service.poolSubmits() != 2 || !strings.Contains(body, recovered) || strings.Contains(body, pending) {
			t.Fatalf("pre-commit recovery=(status=%d dispatches=%d body=%q)", w.Code, service.poolSubmits(), body)
		}
		if strings.Contains(body, "provider body") || strings.Contains(body, "raw provider metadata") {
			t.Fatalf("raw failure data leaked: %q", body)
		}
		requests := stallPoolRequests(service)
		if len(requests) != 2 || requests[1].AvoidProviderID != "provider-a" || requests[1].AllowAvoidedProviderFallback {
			t.Fatalf("pre-commit replay requests=%+v", requests)
		}
		if strings.Count(body, recovered) != 1 {
			t.Fatalf("recovered output count is not one: %q", body)
		}
		assertStallAttemptClosedOnce(t, service, string(edgeservice.ProviderPoolPathTunnel), "logical-pre-a")
		assertStallAttemptClosedOnce(t, service, string(edgeservice.ProviderPoolPathTunnel), "logical-pre-b")
	})

	t.Run("post-commit finish emits one sanitized terminal", func(t *testing.T) {
		const pending = "pending-finish-after-commit"
		service := newScriptedPoolRunService(
			logicalFinishStallAttempt("logical-post-a", pending, true),
			stallMatrixSuccessAttempt(openAIRebuildEndpointChat, string(edgeservice.ProviderPoolPathTunnel), true, "forbidden", "provider-b", "must-not-replay"),
		)
		w := runStallMatrixHandler(t, logicalFinishMatrixServer(service, 1), openAIRebuildEndpointChat, true, nil)
		body := w.Body.String()
		if w.Code != http.StatusOK || service.poolSubmits() != 1 || !strings.Contains(body, "committed-prefix") || strings.Contains(body, "must-not-replay") {
			t.Fatalf("post-commit recovery=(status=%d dispatches=%d body=%q)", w.Code, service.poolSubmits(), body)
		}
		if strings.Count(body, `"type":"run_error"`) != 1 || strings.Count(body, "data: [DONE]") != 1 {
			t.Fatalf("post-commit terminal count mismatch: %q", body)
		}
		if strings.Contains(body, "provider body") || strings.Contains(body, "raw provider metadata") || strings.Contains(body, pending) {
			t.Fatalf("pending/raw bytes leaked after commit: %q", body)
		}
		assertStallAttemptClosedOnce(t, service, string(edgeservice.ProviderPoolPathTunnel), "logical-post-a")
	})
}

func semanticDisabledCommittedStallAttempt(endpoint, runID, pending string) scriptedPoolAttempt {
	body := []byte(
		"data: {\"id\":\"disabled-chat\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"disabled-committed-prefix\"},\"finish_reason\":null}]}\n\n" +
			fmt.Sprintf("data: {\"id\":%q,\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n", pending),
	)
	if endpoint == openAIRebuildEndpointResponses {
		body = []byte(
			"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-disabled\",\"object\":\"response\",\"status\":\"in_progress\"},\"sequence_number\":1}\n\n" +
				"data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg-disabled\",\"output_index\":0,\"content_index\":0,\"delta\":\"disabled-committed-prefix\",\"sequence_number\":2}\n\n" +
				fmt.Sprintf("data: {\"type\":\"response.completed\",\"response\":{\"id\":%q,\"object\":\"response\",\"status\":\"completed\"},\"sequence_number\":3}\n\n", pending),
		)
	}
	return scriptedPoolAttempt{
		path: string(edgeservice.ProviderPoolPathTunnel), runID: runID, provider: "provider-a", target: "served-a",
		frames: bufferedTunnelFrames(
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "text/event-stream"}},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: body},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: confirmedStallFailure("available")},
		),
	}
}

// TestOpenAISemanticGateDisabledStallTerminalMatrix proves that endpoint
// framing and the private typed-stall owner remain active when semantic filters
// are disabled. Pre-commit stalls replay once; an opened Chat/Responses stream
// never replays and receives one sanitized endpoint-native terminal.
func TestOpenAISemanticGateDisabledStallTerminalMatrix(t *testing.T) {
	for _, endpoint := range []string{openAIRebuildEndpointChat, openAIRebuildEndpointResponses} {
		t.Run(endpoint+"/pre-commit", func(t *testing.T) {
			const recovered = "semantic-disabled-recovered"
			service := newScriptedPoolRunService(
				stallMatrixFailureAttempt(string(edgeservice.ProviderPoolPathTunnel), "disabled-pre-a", "provider-a", "available"),
				stallMatrixSuccessAttempt(endpoint, string(edgeservice.ProviderPoolPathTunnel), true, "disabled-pre-b", "provider-b", recovered),
			)
			w := runStallMatrixHandler(t, stallMatrixServer(service, false, 1), endpoint, true, nil)
			body := w.Body.String()
			if w.Code != http.StatusOK || service.poolSubmits() != 2 || strings.Count(body, recovered) != 1 {
				t.Fatalf("pre-commit response=(status=%d dispatches=%d body=%q)", w.Code, service.poolSubmits(), body)
			}
			if strings.Contains(body, "provider body") || strings.Contains(body, "raw provider metadata") {
				t.Fatalf("raw failure data leaked: %q", body)
			}
			assertStallAttemptClosedOnce(t, service, string(edgeservice.ProviderPoolPathTunnel), "disabled-pre-a")
			assertStallAttemptClosedOnce(t, service, string(edgeservice.ProviderPoolPathTunnel), "disabled-pre-b")
		})

		t.Run(endpoint+"/post-commit", func(t *testing.T) {
			const pending = "semantic-disabled-pending-terminal"
			service := newScriptedPoolRunService(
				semanticDisabledCommittedStallAttempt(endpoint, "disabled-post-a", pending),
				stallMatrixSuccessAttempt(endpoint, string(edgeservice.ProviderPoolPathTunnel), true, "forbidden", "provider-b", "must-not-replay"),
			)
			w := runStallMatrixHandler(t, stallMatrixServer(service, false, 1), endpoint, true, nil)
			body := w.Body.String()
			if w.Code != http.StatusOK || service.poolSubmits() != 1 || !strings.Contains(body, "disabled-committed-prefix") || strings.Contains(body, "must-not-replay") {
				t.Fatalf("post-commit response=(status=%d dispatches=%d body=%q)", w.Code, service.poolSubmits(), body)
			}
			if strings.Contains(body, pending) || strings.Contains(body, "provider body") || strings.Contains(body, "raw provider metadata") {
				t.Fatalf("pending/raw failure data leaked: %q", body)
			}
			if strings.Count(body, "data: [DONE]") != 1 {
				t.Fatalf("DONE terminal count mismatch: %q", body)
			}
			if endpoint == openAIRebuildEndpointChat {
				if strings.Count(body, `"type":"run_error"`) != 1 {
					t.Fatalf("Chat error terminal count mismatch: %q", body)
				}
			} else if strings.Count(body, `"type":"error"`) != 1 || strings.Contains(body, `"type":"response.completed"`) {
				t.Fatalf("Responses error terminal mismatch: %q", body)
			}
			assertStallAttemptClosedOnce(t, service, string(edgeservice.ProviderPoolPathTunnel), "disabled-post-a")
		})
	}
}

// TestOpenAIStallRecoveryMatrix proves S05 through the supported production
// handlers and the production runtime adapter. It covers every endpoint/path/
// semantic-policy recovery product, then exercises the shared budget and every
// zero-recovery safety guard.
func TestOpenAIStallRecoveryMatrix(t *testing.T) {
	type recoveryCase struct {
		name            string
		endpoint        string
		initialPath     string
		replacementPath string
		stream          bool
		semantic        bool
	}
	tunnelPath := string(edgeservice.ProviderPoolPathTunnel)
	normPath := string(edgeservice.ProviderPoolPathNormalized)
	recoveryCases := []recoveryCase{
		{name: "chat", endpoint: openAIRebuildEndpointChat, initialPath: normPath, replacementPath: normPath, stream: false, semantic: false},
		{name: "chat", endpoint: openAIRebuildEndpointChat, initialPath: normPath, replacementPath: normPath, stream: false, semantic: true},
		{name: "chat", endpoint: openAIRebuildEndpointChat, initialPath: tunnelPath, replacementPath: tunnelPath, stream: false, semantic: false},
		{name: "chat", endpoint: openAIRebuildEndpointChat, initialPath: tunnelPath, replacementPath: tunnelPath, stream: false, semantic: true},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: normPath, replacementPath: normPath, stream: false, semantic: false},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: normPath, replacementPath: normPath, stream: false, semantic: true},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: tunnelPath, replacementPath: tunnelPath, stream: false, semantic: false},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: tunnelPath, replacementPath: tunnelPath, stream: false, semantic: true},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: tunnelPath, replacementPath: normPath, stream: false, semantic: false},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: tunnelPath, replacementPath: normPath, stream: false, semantic: true},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: normPath, replacementPath: tunnelPath, stream: false, semantic: false},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: normPath, replacementPath: tunnelPath, stream: false, semantic: true},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: tunnelPath, replacementPath: tunnelPath, stream: true, semantic: false},
		{name: "responses", endpoint: openAIRebuildEndpointResponses, initialPath: tunnelPath, replacementPath: tunnelPath, stream: true, semantic: true},
	}
	for _, tc := range recoveryCases {
		pathLabel := tc.initialPath
		if tc.initialPath != tc.replacementPath {
			pathLabel = fmt.Sprintf("%s_to_%s", tc.initialPath, tc.replacementPath)
		}
		name := fmt.Sprintf("recover/%s/%s/stream=%t/semantic=%t", tc.name, pathLabel, tc.stream, tc.semantic)
		t.Run(name, func(t *testing.T) {
			marker := fmt.Sprintf("recovered-%s-%s-stream-%t", tc.endpoint, pathLabel, tc.stream)
			service := newScriptedPoolRunService(
				stallMatrixFailureAttempt(tc.initialPath, "attempt-a", "provider-a", "unavailable"),
				stallMatrixSuccessAttempt(tc.endpoint, tc.replacementPath, tc.stream, "attempt-b", "provider-b", marker),
			)
			w := runStallMatrixHandler(t, stallMatrixServer(service, tc.semantic, 1), tc.endpoint, tc.stream, nil)
			if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), marker) {
				t.Fatalf("recovered response=(status=%d body=%q)", w.Code, w.Body.String())
			}
			if strings.Contains(w.Body.String(), "provider body") || strings.Contains(w.Body.String(), "raw provider metadata") {
				t.Fatalf("raw stall data leaked: %q", w.Body.String())
			}
			if tc.endpoint == openAIRebuildEndpointChat && strings.Count(w.Body.String(), `"object":"chat.completion"`) != 1 {
				t.Fatalf("chat terminal count is not one: %q", w.Body.String())
			}
			if tc.endpoint == openAIRebuildEndpointResponses && tc.stream {
				if strings.Count(w.Body.String(), `"type":"response.completed"`) != 1 || strings.Count(w.Body.String(), "data: [DONE]") != 1 {
					t.Fatalf("streaming Responses terminal count is not one: %q", w.Body.String())
				}
			} else if tc.endpoint == openAIRebuildEndpointResponses && strings.Count(w.Body.String(), `"object":"response"`) != 1 {
				t.Fatalf("responses terminal count is not one: %q", w.Body.String())
			}
			requests := stallPoolRequests(service)
			if len(requests) != 2 || requests[1].AvoidProviderID != "provider-a" || requests[1].AllowAvoidedProviderFallback {
				t.Fatalf("re-admission requests=%+v, want one provider-a avoidance without fallback", requests)
			}
			pools, cancels, _, _, runRequests, tunnelRequests := service.snapshot()
			if pools != 2 || len(cancels) != 0 {
				t.Fatalf("dispatch/cancel lifecycle=(%d,%v), want (2,none)", pools, cancels)
			}
			if tc.replacementPath == normPath {
				if len(runRequests) == 0 {
					t.Fatalf("expected at least one normalized run request, got 0")
				}
				replacementRun := runRequests[len(runRequests)-1]
				if replacementRun.TimeoutSec != 5 {
					t.Fatalf("normalized replacement TimeoutSec = %d, want ingress timeout 5", replacementRun.TimeoutSec)
				}
				wantPrompt := "user: hi"
				wantEstimate := 2
				wantEstimateStr := "2"
				if tc.endpoint == openAIRebuildEndpointResponses {
					wantPrompt = "hi"
					wantEstimate = 7
					wantEstimateStr = "7"
				}
				if replacementRun.Prompt != wantPrompt {
					t.Fatalf("normalized replacement Prompt = %q, want %q", replacementRun.Prompt, wantPrompt)
				}
				if tc.endpoint == openAIRebuildEndpointResponses {
					if prompt, ok := replacementRun.Input["prompt"].(string); !ok || prompt != "hi" {
						t.Fatalf("normalized replacement Input[\"prompt\"] = %v, want hi", replacementRun.Input["prompt"])
					}
				}
				if replacementRun.Metadata["openai_model"] != "matrix-model" || replacementRun.Metadata["openai_stream"] != fmt.Sprintf("%t", tc.stream) {
					t.Fatalf("normalized replacement metadata = %v, want model matrix-model and stream %t", replacementRun.Metadata, tc.stream)
				}
				if replacementRun.Metadata["strict_output"] != "false" ||
					replacementRun.Metadata["estimated_input_tokens"] != wantEstimateStr ||
					replacementRun.Metadata["context_class"] != "normal" {
					t.Fatalf("normalized replacement derived metadata = %v", replacementRun.Metadata)
				}
				if replacementRun.MaxQueue != 0 || replacementRun.QueueTimeoutMS != 0 {
					t.Fatalf("normalized replacement queue fields=(%d,%d), want (0,0)", replacementRun.MaxQueue, replacementRun.QueueTimeoutMS)
				}
				if replacementRun.EstimatedInputTokens != wantEstimate || replacementRun.ContextClass != "normal" {
					t.Fatalf("normalized replacement estimate/class=(%d,%q), want (%d,normal)", replacementRun.EstimatedInputTokens, replacementRun.ContextClass, wantEstimate)
				}
			} else {
				if len(tunnelRequests) == 0 {
					t.Fatalf("expected at least one provider tunnel request, got 0")
				}
				replacementTunnel := tunnelRequests[len(tunnelRequests)-1]
				if replacementTunnel.TimeoutSec != 5 {
					t.Fatalf("tunnel replacement TimeoutSec = %d, want 5", replacementTunnel.TimeoutSec)
				}
				if replacementTunnel.Stream != tc.stream {
					t.Fatalf("tunnel replacement Stream = %t, want %t", replacementTunnel.Stream, tc.stream)
				}
				if replacementTunnel.Metadata["openai_model"] != "matrix-model" || replacementTunnel.Metadata["openai_stream"] != fmt.Sprintf("%t", tc.stream) {
					t.Fatalf("tunnel replacement metadata = %v, want model matrix-model and stream %t", replacementTunnel.Metadata, tc.stream)
				}
				if replacementTunnel.EstimatedInputTokens <= 0 || replacementTunnel.ContextClass == "" {
					t.Fatalf("tunnel replacement estimate/class invalid: estimate=%d class=%q", replacementTunnel.EstimatedInputTokens, replacementTunnel.ContextClass)
				}
				if replacementTunnel.BuildBody == nil {
					t.Fatalf("tunnel replacement BuildBody is nil")
				}
				rebuilt, err := replacementTunnel.BuildBody("served-b")
				if err != nil {
					t.Fatalf("tunnel replacement BuildBody failed: %v", err)
				}
				if !strings.Contains(string(rebuilt), `"model":"served-b"`) {
					t.Fatalf("tunnel replacement body missing target model served-b: %q", string(rebuilt))
				}
				if tc.endpoint == openAIRebuildEndpointResponses {
					if !strings.Contains(string(rebuilt), `"input":"hi"`) || !strings.Contains(string(rebuilt), fmt.Sprintf(`"stream":%t`, tc.stream)) {
						t.Fatalf("tunnel replacement body lost input or stream: %q", string(rebuilt))
					}
				}
			}
			if tc.endpoint == openAIRebuildEndpointResponses && tc.stream && strings.Contains(w.Body.String(), `"type":"error"`) {
				t.Fatalf("streaming Responses rendered an error terminal: %q", w.Body.String())
			}
			assertStallAttemptClosedOnce(t, service, tc.initialPath, "attempt-a")
			assertStallAttemptClosedOnce(t, service, tc.replacementPath, "attempt-b")
		})
	}

	t.Run("stall replay never grants same-provider fallback", func(t *testing.T) {
		path := string(edgeservice.ProviderPoolPathNormalized)
		service := newScriptedPoolRunService(
			stallMatrixFailureAttempt(path, "available-a", "provider-a", "available"),
			stallMatrixSuccessAttempt(openAIRebuildEndpointChat, path, false, "available-b", "provider-a", "same-provider-recovered"),
		)
		w := runStallMatrixHandler(t, stallMatrixServer(service, false, 1), openAIRebuildEndpointChat, false, nil)
		requests := stallPoolRequests(service)
		if w.Code != http.StatusOK || len(requests) != 2 || requests[1].AvoidProviderID != "provider-a" || requests[1].AllowAvoidedProviderFallback {
			t.Fatalf("available fallback response=%d/%q requests=%+v", w.Code, w.Body.String(), requests)
		}
	})

	t.Run("shared budget emits one sanitized terminal", func(t *testing.T) {
		path := string(edgeservice.ProviderPoolPathNormalized)
		service := newScriptedPoolRunService(
			stallMatrixFailureAttempt(path, "budget-a", "provider-a", "unavailable"),
			stallMatrixFailureAttempt(path, "budget-b", "provider-b", "unavailable"),
			stallMatrixSuccessAttempt(openAIRebuildEndpointChat, path, false, "budget-c", "provider-a", "must-not-dispatch"),
		)
		w := runStallMatrixHandler(t, stallMatrixServer(service, true, 1), openAIRebuildEndpointChat, false, nil)
		if service.poolSubmits() != 2 || w.Code != http.StatusBadGateway || strings.Count(w.Body.String(), `"type":"run_error"`) != 1 || strings.Contains(w.Body.String(), "provider body") || strings.Contains(w.Body.String(), "must-not-dispatch") {
			t.Fatalf("budget terminal=(dispatches=%d status=%d body=%q)", service.poolSubmits(), w.Code, w.Body.String())
		}
		assertStallAttemptClosedOnce(t, service, path, "budget-a")
		assertStallAttemptClosedOnce(t, service, path, "budget-b")
	})

	for _, guard := range []struct {
		name    string
		budget  int
		failure *iop.ExecutionFailure
	}{
		{name: "generic-unconfirmed", budget: 1, failure: &iop.ExecutionFailure{Code: openAIStallFailureCode, Retryable: true, Message: "secret generic"}},
		{name: "exhausted", budget: 0, failure: confirmedStallFailure("unknown")},
		{name: "unsupported-health", budget: 1, failure: func() *iop.ExecutionFailure {
			f := confirmedStallFailure("unknown")
			f.Metadata[openAIStallProviderHealthKey] = "unsupported"
			return f
		}()},
	} {
		t.Run("guard/"+guard.name, func(t *testing.T) {
			path := string(edgeservice.ProviderPoolPathNormalized)
			attempt := scriptedPoolAttempt{path: path, runID: "guard-" + guard.name, provider: "provider-a", target: "served-a", runEvents: bufferedRunEvents(&iop.RunEvent{Type: "error", Failure: guard.failure})}
			service := newScriptedPoolRunService(attempt, stallMatrixSuccessAttempt(openAIRebuildEndpointChat, path, false, "forbidden", "provider-b", "must-not-render"))
			w := runStallMatrixHandler(t, stallMatrixServer(service, true, guard.budget), openAIRebuildEndpointChat, false, nil)
			if service.poolSubmits() != 1 || w.Code != http.StatusBadGateway || strings.Contains(w.Body.String(), "secret") || strings.Contains(w.Body.String(), "must-not-render") {
				t.Fatalf("guard result=(dispatches=%d status=%d body=%q)", service.poolSubmits(), w.Code, w.Body.String())
			}
		})
	}

	t.Run("guard/committed", func(t *testing.T) {
		path := string(edgeservice.ProviderPoolPathTunnel)
		wire := []byte("data: {\"id\":\"partial\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"committed\"},\"finish_reason\":null}]}\n\n")
		attempt := scriptedPoolAttempt{path: path, runID: "committed-a", provider: "provider-a", target: "served-a", frames: bufferedTunnelFrames(
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "text/event-stream"}},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: wire},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: confirmedStallFailure("available")},
		)}
		service := newScriptedPoolRunService(attempt, stallMatrixSuccessAttempt(openAIRebuildEndpointChat, path, false, "forbidden", "provider-b", "must-not-render"))
		w := runStallMatrixHandler(t, stallMatrixServer(service, true, 1), openAIRebuildEndpointChat, true, nil)
		if service.poolSubmits() != 1 || !strings.Contains(w.Body.String(), "committed") || strings.Contains(w.Body.String(), "must-not-render") {
			t.Fatalf("committed guard=(dispatches=%d status=%d body=%q)", service.poolSubmits(), w.Code, w.Body.String())
		}
	})

	t.Run("guard/caller-cancelled", func(t *testing.T) {
		path := string(edgeservice.ProviderPoolPathNormalized)
		openEvents := make(chan *iop.RunEvent)
		service := newScriptedPoolRunService(scriptedPoolAttempt{path: path, runID: "cancelled-a", provider: "provider-a", target: "served-a", runEvents: openEvents})
		ctx, cancel := context.WithCancel(context.Background())
		cancel()
		w := runStallMatrixHandler(t, stallMatrixServer(service, true, 1), openAIRebuildEndpointChat, false, ctx)
		if service.poolSubmits() != 1 || w.Code != http.StatusRequestTimeout {
			t.Fatalf("cancel guard=(dispatches=%d status=%d body=%q)", service.poolSubmits(), w.Code, w.Body.String())
		}
	})

	t.Run("guard/tool-side-effect", func(t *testing.T) {
		path := string(edgeservice.ProviderPoolPathTunnel)
		toolWire := []byte("data: {\"id\":\"tool\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-1\",\"type\":\"function\",\"function\":{\"name\":\"act\",\"arguments\":\"{}\"}}]},\"finish_reason\":null}]}\n\n")
		attempt := scriptedPoolAttempt{path: path, runID: "tool-a", provider: "provider-a", target: "served-a", frames: bufferedTunnelFrames(
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "text/event-stream"}},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: toolWire},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Failure: confirmedStallFailure("available")},
		)}
		service := newScriptedPoolRunService(attempt, stallMatrixSuccessAttempt(openAIRebuildEndpointChat, path, false, "forbidden", "provider-b", "must-not-render"))
		w := runStallMatrixHandler(t, stallMatrixServer(service, true, 1), openAIRebuildEndpointChat, true, nil)
		if service.poolSubmits() != 1 || !strings.Contains(w.Body.String(), "call-1") || strings.Contains(w.Body.String(), "must-not-render") {
			t.Fatalf("tool guard=(dispatches=%d status=%d body=%q)", service.poolSubmits(), w.Code, w.Body.String())
		}
	})

	for _, owner := range []struct {
		name       string
		requestRef string
		register   bool
	}{
		{name: "missing-snapshot", requestRef: "", register: true},
		{name: "no-owner", register: false},
	} {
		t.Run("guard/"+owner.name, func(t *testing.T) {
			raw := []byte(`{"model":"matrix-model","stream":true,"messages":[{"role":"user","content":"hi"}]}`)
			service := &fakeRunService{}
			srv := NewServer(config.EdgeOpenAIConf{Adapter: "ollama"}, service, nil)
			base := newTestRequestContext(t, routeDispatch{Adapter: "ollama", Target: "served-a", TimeoutSec: 5}, raw)
			var req chatCompletionRequest
			if err := json.Unmarshal(raw, &req); err != nil {
				t.Fatal(err)
			}
			dc := srv.newChatDispatchContext(base, req, "hi", strictOutputPolicy{})
			handle := &fakeRunResult{dispatch: edgeservice.RunDispatch{RunID: owner.name + "-a", NodeID: "node-a", ProviderID: "provider-a", ModelGroupKey: "matrix-model", Target: "served-a"}, events: bufferedRunEvents(&iop.RunEvent{Type: "error", Failure: confirmedStallFailure("unknown")})}
			var registry streamgate.FilterRegistrySnapshot
			var err error
			if owner.register {
				_, registration, regErr := openAIStallRecoveryRegistration(openAIOutputFilterContext{requestRef: owner.requestRef})
				if regErr != nil {
					t.Fatal(regErr)
				}
				registry, err = openAIStreamGateRegistrySnapshotWith(registration)
			} else {
				registry, err = openAIStreamGateRegistrySnapshot()
			}
			if err != nil {
				t.Fatal(err)
			}
			w := httptest.NewRecorder()
			sink := newOpenAIChatSSEReleaseSink(w, nil, "chatcmpl-guard", time.Now().Unix(), "matrix-model")
			runtime, _, err := srv.buildOpenAIChatStreamGateRuntime(dc, handle, sink, registry)
			if err != nil {
				t.Fatal(err)
			}
			runErr := runtime.Run(t.Context())
			_ = runtime.CloseRequestResources(t.Context(), runErr == nil)
			if len(service.reqsSnapshot()) != 0 || !strings.Contains(w.Body.String(), openAIStallFailureCode) || strings.Contains(w.Body.String(), "provider body") {
				t.Fatalf("owner guard=(dispatches=%d runErr=%v body=%q)", len(service.reqsSnapshot()), runErr, w.Body.String())
			}
		})
	}
}

// TestOpenAISemanticGateDisabledCompatibility proves that the always-owned
// runtime preserves endpoint-native behavior while semantic filters are off.
func TestOpenAISemanticGateDisabledCompatibility(t *testing.T) {
	t.Run("chat/normalized/sse", func(t *testing.T) {
		service := newScriptedPoolRunService(scriptedPoolAttempt{
			path: string(edgeservice.ProviderPoolPathNormalized), runID: "compat-chat-run", provider: "provider-a", target: "served-a",
			runEvents: bufferedRunEvents(
				&iop.RunEvent{Type: "reasoning_delta", Delta: "private compatibility reasoning"},
				&iop.RunEvent{Type: "delta", Delta: "compatibility chat"},
				&iop.RunEvent{Type: "complete", Metadata: map[string]string{"finish_reason": "length"}, Usage: &iop.Usage{InputTokens: 2, OutputTokens: 3}},
			),
		})
		srv := stallMatrixServer(service, false, 1)
		w := runStallMatrixHandler(t, srv, openAIRebuildEndpointChat, true, nil)
		body := w.Body.String()
		if srv.streamGateSemanticEnabled() || w.Code != http.StatusOK || w.Header().Get("Content-Type") != "text/event-stream" || !strings.Contains(body, "compatibility chat") || !strings.Contains(body, `"reasoning_content":"private compatibility reasoning"`) || !strings.Contains(body, `"finish_reason":"length"`) || strings.Count(body, "data: [DONE]") != 1 || service.poolSubmits() != 1 {
			t.Fatalf("normalized Chat compatibility=(status=%d headers=%v dispatches=%d body=%q)", w.Code, w.Header(), service.poolSubmits(), body)
		}
	})

	t.Run("chat/tunnel/sse-byte-order", func(t *testing.T) {
		wire := []byte("data: {\"id\":\"compat-chat\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"compat tunnel\"},\"finish_reason\":null}]}\n\ndata: {\"id\":\"compat-chat\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"length\"}]}\n\ndata: [DONE]\n\n")
		service := newScriptedPoolRunService(scriptedPoolAttempt{path: string(edgeservice.ProviderPoolPathTunnel), runID: "compat-chat-tunnel", provider: "provider-a", target: "served-a", frames: bufferedTunnelFrames(
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "text/event-stream", "X-Compat": "chat"}},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: wire[:len(wire)/2]},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: wire[len(wire)/2:]},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
		)})
		w := runStallMatrixHandler(t, stallMatrixServer(service, false, 1), openAIRebuildEndpointChat, true, nil)
		if w.Code != http.StatusOK || w.Header().Get("X-Compat") != "chat" || w.Body.String() != string(wire) || strings.Count(w.Body.String(), "data: [DONE]") != 1 || service.poolSubmits() != 1 {
			t.Fatalf("tunnel Chat compatibility=(status=%d headers=%v dispatches=%d body=%q want=%q)", w.Code, w.Header(), service.poolSubmits(), w.Body.String(), wire)
		}
	})

	t.Run("responses/normalized/json", func(t *testing.T) {
		service := newScriptedPoolRunService(scriptedPoolAttempt{
			path: string(edgeservice.ProviderPoolPathNormalized), runID: "compat-responses-run", provider: "provider-a", target: "served-a",
			runEvents: bufferedRunEvents(
				&iop.RunEvent{Type: "reasoning_delta", Delta: "compatibility reasoning"},
				&iop.RunEvent{Type: "delta", Delta: "compatibility responses"},
				&iop.RunEvent{Type: "complete", Usage: &iop.Usage{InputTokens: 4, OutputTokens: 5, ReasoningTokens: 2}},
			),
		})
		w := runStallMatrixHandler(t, stallMatrixServer(service, false, 1), openAIRebuildEndpointResponses, false, nil)
		var response responsesResponse
		if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
			t.Fatalf("decode Responses compatibility: %v body=%q", err, w.Body.String())
		}
		if w.Code != http.StatusOK || response.OutputText != "compatibility responses" || response.Usage.TotalTokens != 9 || strings.Count(w.Body.String(), `"object":"response"`) != 1 || service.poolSubmits() != 1 {
			t.Fatalf("normalized Responses compatibility=(status=%d response=%+v dispatches=%d body=%q)", w.Code, response, service.poolSubmits(), w.Body.String())
		}
	})

	t.Run("responses/tunnel/json-byte-order", func(t *testing.T) {
		wire := []byte(`{"id":"compat-responses","object":"response","status":"completed","output_text":"compat tunnel responses","output":[]}`)
		service := newScriptedPoolRunService(scriptedPoolAttempt{path: string(edgeservice.ProviderPoolPathTunnel), runID: "compat-responses-tunnel", provider: "provider-a", target: "served-a", frames: bufferedTunnelFrames(
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": "application/json", "X-Compat": "responses"}},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: wire[:31]},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: wire[31:]},
			&iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true},
		)})
		w := runStallMatrixHandler(t, stallMatrixServer(service, false, 1), openAIRebuildEndpointResponses, false, nil)
		if w.Code != http.StatusOK || w.Header().Get("X-Compat") != "responses" || w.Body.String() != string(wire) || strings.Count(w.Body.String(), `"object":"response"`) != 1 || service.poolSubmits() != 1 {
			t.Fatalf("tunnel Responses compatibility=(status=%d headers=%v dispatches=%d body=%q want=%q)", w.Code, w.Header(), service.poolSubmits(), w.Body.String(), wire)
		}
	})
}
