iop/apps/edge/internal/openai/hot_path_anthropic_gate_test.go

package openai

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

	iop "iop/proto/gen/iop"
)

type hotPathAnthropicSSEEvent struct {
	name    string
	payload map[string]any
}

func TestHotPathAnthropicDirectStreamCodec(t *testing.T) {
	tests := []struct {
		name, profile, responseID, providerToolID, providerBody string
		wantSignature                                           string
	}{
		{
			name: "native provider", profile: "anthropic", responseID: "msg-anthropic-gate", providerToolID: "provider-native-tool",
			providerBody: strings.Join([]string{
				`data: {"type":"message_start","message":{"id":"msg-anthropic-gate","type":"message","role":"assistant","content":[],"usage":{"input_tokens":9,"output_tokens":0,"cache_read_input_tokens":2}}}`,
				`data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}}`,
				`data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"plan "}}`,
				`data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"now"}}`,
				`data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-native"}}`,
				`data: {"type":"content_block_stop","index":0}`,
				`data: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}`,
				`data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"alpha "}}`,
				`data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"omega"}}`,
				`data: {"type":"content_block_stop","index":1}`,
				`data: {"type":"content_block_start","index":2,"content_block":{"type":"tool_use","id":"provider-native-tool","name":"read_file","input":{}}}`,
				`data: {"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}}`,
				`data: {"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"\"README.md\"}"}}`,
				`data: {"type":"content_block_stop","index":2}`,
				`data: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":7}}`,
				`data: {"type":"message_stop"}`, "",
			}, "\n\n"),
			wantSignature: "sig-native",
		},
		{
			name: "OpenAI provider", profile: "openai", responseID: "chatcmpl-anthropic-gate", providerToolID: "provider-openai-tool",
			providerBody: strings.Join([]string{
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{"reasoning_content":"plan "},"finish_reason":null}]}`,
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{"reasoning_content":"now"},"finish_reason":null}]}`,
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{"content":"alpha "},"finish_reason":null}]}`,
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{"content":"omega"},"finish_reason":null}]}`,
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"provider-openai-tool","type":"function","function":{"name":"read_file","arguments":"{\"path\":"}}]},"finish_reason":null}]}`,
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"README.md\"}"}}]},"finish_reason":null}]}`,
				`data: {"id":"chatcmpl-anthropic-gate","created":1777002001,"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":9,"completion_tokens":7,"total_tokens":16}}`,
				`data: [DONE]`, "",
			}, "\n\n"),
		},
	}

	for _, test := range tests {
		test := test
		t.Run(test.name, func(t *testing.T) {
			var decoded normalizedStageOutput
			var decodeErr error
			if test.profile == "anthropic" {
				decoded, decodeErr = decodeAnthropicPresetSSE([]byte(test.providerBody))
			} else {
				decoded, decodeErr = decodeOpenAIPresetSSE([]byte(test.providerBody))
			}
			if decodeErr != nil || len(decoded.ToolCalls) != 1 || len(decoded.Deltas) != 6 {
				t.Fatalf("provider fixture decode: output=%+v err=%v", decoded, decodeErr)
			}
			candidate := anthropicTestCandidate(t, test.profile)
			fragments := splitAnthropicFixture([]byte(test.providerBody), 13, 79, 211, len(test.providerBody)-17)
			contentType := "text/event-stream"
			srv, fake := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, contentType, fragments...))
			response := serveHotPathAnthropicBody(t, srv, `{"model":"virtual-model","max_tokens":64,"messages":[{"role":"user","content":"hello"}],"tools":[{"name":"read_file","description":"read","input_schema":{"type":"object"}}],"stream":true}`)
			if response.Code != http.StatusOK || response.Header().Get("Content-Type") != "text/event-stream" {
				t.Fatalf("response mismatch: status=%d headers=%v body=%s", response.Code, response.Header(), response.Body.String())
			}

			events := decodeHotPathAnthropicSSE(t, response.Body.String())
			assertHotPathAnthropicDirectEvents(t, events, test.responseID, test.wantSignature)
			if fake.poolSubmitCountSnapshot() != 1 {
				t.Fatalf("selector submissions=%d, want 1", fake.poolSubmitCountSnapshot())
			}
			assertHotPathWaiting(t, srv, test.responseID+"-tool-1", test.providerToolID)
		})
	}
}

func TestHotPathAnthropicDirectStreamPreservesEmptyToolInput(t *testing.T) {
	providerBody := strings.Join([]string{
		`data: {"type":"message_start","message":{"id":"msg-empty-tool","type":"message","role":"assistant","content":[],"usage":{"input_tokens":5,"output_tokens":0}}}`,
		`data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"provider-zero-arg-tool","name":"list_dir","input":{}}}`,
		`data: {"type":"content_block_stop","index":0}`,
		`data: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":4}}`,
		`data: {"type":"message_stop"}`, "",
	}, "\n\n")

	candidate := anthropicTestCandidate(t, "anthropic")
	contentType := "text/event-stream"
	srv, fake := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, contentType, []byte(providerBody)))
	response := serveHotPathAnthropicBody(t, srv, `{"model":"virtual-model","max_tokens":64,"messages":[{"role":"user","content":"list files"}],"tools":[{"name":"list_dir","description":"list","input_schema":{"type":"object"}}],"stream":true}`)
	if response.Code != http.StatusOK || response.Header().Get("Content-Type") != "text/event-stream" {
		t.Fatalf("response mismatch: status=%d headers=%v body=%s", response.Code, response.Header(), response.Body.String())
	}

	events := decodeHotPathAnthropicSSE(t, response.Body.String())
	wantNames := []string{
		"message_start",
		"content_block_start", "content_block_delta", "content_block_stop",
		"message_delta", "message_stop",
	}
	if got := hotPathAnthropicEventNames(events); strings.Join(got, ",") != strings.Join(wantNames, ",") {
		t.Fatalf("event order=%v, want %v; body=%s", got, wantNames, response.Body.String())
	}

	var toolID, toolName, partialJSON string
	var deltaCount int
	for _, event := range events {
		switch event.name {
		case "content_block_start":
			block := hotPathAnthropicMap(t, event.payload["content_block"])
			if block["type"] == "tool_use" {
				toolID, _ = block["id"].(string)
				toolName, _ = block["name"].(string)
			}
		case "content_block_delta":
			delta := hotPathAnthropicMap(t, event.payload["delta"])
			if delta["type"] == "input_json_delta" {
				deltaCount++
				partialJSON, _ = delta["partial_json"].(string)
			}
		}
	}

	if toolID != "msg-empty-tool-tool-1" || toolName != "list_dir" || deltaCount != 1 || partialJSON != "{}" {
		t.Fatalf("empty tool preservation mismatch: toolID=%q toolName=%q deltaCount=%d partialJSON=%q", toolID, toolName, deltaCount, partialJSON)
	}

	if fake.poolSubmitCountSnapshot() != 1 {
		t.Fatalf("selector submissions=%d, want 1", fake.poolSubmitCountSnapshot())
	}
	assertHotPathWaiting(t, srv, "msg-empty-tool-tool-1", "provider-zero-arg-tool")
}

func TestHotPathAnthropicLightStreamAggregatesStages(t *testing.T) {
	fixture := newScriptedLightFixture(t, "anthropic", false)
	fixture.service.responses[3] = func(string) string {
		return scriptedLightCompletionWithUsage("anthropic", "local-visible", "local-reason", 5, 3)
	}
	fixture.service.responses[4] = func(requestID string) string {
		return scriptedReviewWriteWithUsage("anthropic", requestID, 7, 4)
	}

	prepare := fixture.request()
	fixture.consumeToolResponse(prepare, []string{`{"written":true}`})
	pair := fixture.request()
	fixture.consumeToolResponse(pair, []string{`{"written":true}`, `{"written":true}`})
	localRead := fixture.request()
	fixture.consumeToolResponse(localRead, []string{`{"written":true}`})

	before := len(fixture.service.snapshots())
	response := fixture.requestWithOptions(64, true)
	if response.Code != http.StatusOK {
		t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
	}
	if got := len(fixture.service.snapshots()) - before; got != 2 {
		t.Fatalf("same-turn provider stages=%d, want 2", got)
	}
	requests := fixture.service.snapshots()
	assertCapturedHotPathBudget(t, requests[len(requests)-2], fixture.service.candidate, 64)
	assertCapturedHotPathBudget(t, requests[len(requests)-1], fixture.service.candidate, 61)
	events := decodeHotPathAnthropicSSE(t, response.Body.String())
	assertHotPathAnthropicBlockIndexes(t, events, 5)

	wantNames := []string{
		"message_start",
		"content_block_start", "content_block_delta", "content_block_delta", "content_block_stop",
		"content_block_start", "content_block_delta", "content_block_stop",
		"content_block_start", "content_block_delta", "content_block_delta", "content_block_stop",
		"content_block_start", "content_block_delta", "content_block_stop",
		"content_block_start", "content_block_delta", "content_block_stop",
		"message_delta", "message_stop",
	}
	if got := hotPathAnthropicEventNames(events); strings.Join(got, ",") != strings.Join(wantNames, ",") {
		t.Fatalf("event order=%v, want %v; body=%s", got, wantNames, response.Body.String())
	}
	startMessage := hotPathAnthropicMap(t, events[0].payload["message"])
	requestID, snapshot := soleHotPathSnapshot(t, fixture.server)
	if startMessage["id"] != "msg-light-complete" || startMessage["id"] == requestID || startMessage["model"] != "virtual-model" {
		t.Fatalf("outer identity mismatch: message=%+v logical_request=%s", startMessage, requestID)
	}

	wantKinds := []string{"thinking", "text", "thinking", "text", "tool_use"}
	var gotKinds, thinking, text []string
	var toolID, toolName, toolArgs, stopReason string
	for _, event := range events {
		switch event.name {
		case "content_block_start":
			block := hotPathAnthropicMap(t, event.payload["content_block"])
			gotKinds = append(gotKinds, fmt.Sprint(block["type"]))
			if block["type"] == "tool_use" {
				toolID, _ = block["id"].(string)
				toolName, _ = block["name"].(string)
			}
		case "content_block_delta":
			delta := hotPathAnthropicMap(t, event.payload["delta"])
			switch delta["type"] {
			case "thinking_delta":
				thinking = append(thinking, fmt.Sprint(delta["thinking"]))
			case "text_delta":
				text = append(text, fmt.Sprint(delta["text"]))
			case "input_json_delta":
				toolArgs += fmt.Sprint(delta["partial_json"])
			}
		case "message_delta":
			delta := hotPathAnthropicMap(t, event.payload["delta"])
			stopReason, _ = delta["stop_reason"].(string)
			usage := hotPathAnthropicMap(t, event.payload["usage"])
			if usage["input_tokens"] != float64(12) || usage["output_tokens"] != float64(7) {
				t.Fatalf("aggregate usage=%+v, want input=12 output=7", usage)
			}
		}
	}
	if strings.Join(gotKinds, ",") != strings.Join(wantKinds, ",") ||
		strings.Join(thinking, "") != "local-reasonreview-reason" || strings.Join(text, "") != "local-visiblereview-visible" ||
		toolName != "write_file" || !json.Valid([]byte(toolArgs)) || stopReason != "tool_use" {
		t.Fatalf("multi-stage output mismatch: kinds=%v thinking=%v text=%v tool=%q/%q/%q stop=%q body=%s",
			gotKinds, thinking, text, toolID, toolName, toolArgs, stopReason, response.Body.String())
	}
	if len(snapshot.ExpectedCallIDs) != 1 || snapshot.ExpectedCallIDs[0] != toolID {
		t.Fatalf("tool correlation mismatch: tool=%q snapshot=%+v", toolID, snapshot)
	}
	if toolID != "msg-light-complete-tool-1" || strings.Contains(response.Body.String(), "msg-review-write") {
		t.Fatalf("public identity/tool namespace leaked a later provider id: tool=%q body=%s", toolID, response.Body.String())
	}
}

func TestHotPathAnthropicToolIDsAreMonotonic(t *testing.T) {
	candidate := anthropicTestCandidate(t, "anthropic")
	providerBody := []byte(`{"id":"msg-anthropic-tools","type":"message","role":"assistant","content":[{"type":"tool_use","id":"provider-tool-a","name":"read_file","input":{"path":"a"}},{"type":"tool_use","id":"provider-tool-b","name":"read_file","input":{"path":"b"}}],"stop_reason":"tool_use","usage":{"input_tokens":4,"output_tokens":3}}`)
	srv, _ := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, "application/json", providerBody))
	response := serveHotPathAnthropic(t, srv, true)
	if response.Code != http.StatusOK {
		t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
	}
	events := decodeHotPathAnthropicSSE(t, response.Body.String())
	assertHotPathAnthropicBlockIndexes(t, events, 2)
	var toolIDs []string
	for _, event := range events {
		if event.name != "content_block_start" {
			continue
		}
		block := hotPathAnthropicMap(t, event.payload["content_block"])
		if block["type"] == "tool_use" {
			toolIDs = append(toolIDs, fmt.Sprint(block["id"]))
		}
	}
	wantIDs := []string{"msg-anthropic-tools-tool-1", "msg-anthropic-tools-tool-2"}
	if fmt.Sprint(toolIDs) != fmt.Sprint(wantIDs) {
		t.Fatalf("tool ids=%v, want %v; body=%s", toolIDs, wantIDs, response.Body.String())
	}
	requestID, snapshot := soleHotPathSnapshot(t, srv)
	expectedSet := make(map[string]bool, len(snapshot.ExpectedCallIDs))
	for _, id := range snapshot.ExpectedCallIDs {
		expectedSet[id] = true
	}
	if len(snapshot.ExpectedCallIDs) != len(wantIDs) || !expectedSet[wantIDs[0]] || !expectedSet[wantIDs[1]] {
		t.Fatalf("expected caller ids=%v, want %v", snapshot.ExpectedCallIDs, wantIDs)
	}
	srv.requestCoordinator.mu.Lock()
	record := srv.requestCoordinator.requests[requestID]
	mapping := map[string]string{}
	if record != nil {
		for _, id := range wantIDs {
			mapping[id] = record.publicToProvider[id]
		}
	}
	srv.requestCoordinator.mu.Unlock()
	if mapping[wantIDs[0]] != "provider-tool-a" || mapping[wantIDs[1]] != "provider-tool-b" {
		t.Fatalf("provider tool mapping=%v", mapping)
	}
}

func TestHotPathAnthropicCallerCapAndNonStream(t *testing.T) {
	candidate := anthropicTestCandidate(t, "anthropic")
	providerBody := []byte(`{"id":"msg-anthropic-cap","type":"message","role":"assistant","content":[{"type":"text","text":"abcdefghij"}],"stop_reason":"end_turn","usage":{"input_tokens":3,"output_tokens":2}}`)
	srv, fake := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, "application/json", providerBody[:31], providerBody[31:]))
	response := serveHotPathAnthropicBody(t, srv, `{"model":"virtual-model","max_tokens":2,"messages":[{"role":"user","content":"cap"}],"stream":false}`)
	if response.Code != http.StatusOK {
		t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
	}
	var decoded struct {
		ID         string            `json:"id"`
		Model      string            `json:"model"`
		Content    []json.RawMessage `json:"content"`
		StopReason string            `json:"stop_reason"`
		Usage      anthropicUsage    `json:"usage"`
	}
	if err := json.Unmarshal(response.Body.Bytes(), &decoded); err != nil {
		t.Fatal(err)
	}
	if decoded.ID != "msg-anthropic-cap" || decoded.Model != "virtual-model" || decoded.StopReason != "end_turn" ||
		decoded.Usage.InputTokens != 3 || decoded.Usage.OutputTokens != 2 || len(decoded.Content) != 1 {
		t.Fatalf("non-stream envelope mismatch: %+v body=%s", decoded, response.Body.String())
	}
	var textBlock struct {
		Type string `json:"type"`
		Text string `json:"text"`
	}
	if err := json.Unmarshal(decoded.Content[0], &textBlock); err != nil || textBlock.Type != "text" || textBlock.Text != "abcdefghij" {
		t.Fatalf("provider-token content=%+v err=%v", textBlock, err)
	}
	if fake.poolSubmitCountSnapshot() != 1 {
		t.Fatalf("selector submissions=%d, want 1", fake.poolSubmitCountSnapshot())
	}
	var upstream map[string]any
	if bodies := fake.tunnelBodiesSnapshot(); len(bodies) != 1 {
		t.Fatalf("upstream body count=%d, want 1", len(bodies))
	} else if err := json.Unmarshal(bodies[0], &upstream); err != nil || upstream["max_tokens"] != float64(2) {
		t.Fatalf("upstream max_tokens was not retained: body=%s decoded=%+v err=%v", bodies[0], upstream, err)
	}
	assertHotPathTerminal(t, srv)
}

func TestHotPathAnthropicErrorBoundaries(t *testing.T) {
	t.Run("required max tokens fails before dispatch", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "anthropic")
		srv, fake := newHotPathHandlerServer(t, candidate, nil)
		response := serveHotPathAnthropicBody(t, srv, `{"model":"virtual-model","messages":[{"role":"user","content":"missing cap"}],"stream":true}`)
		if response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), `"type":"invalid_request_error"`) ||
			strings.Contains(response.Body.String(), "message_start") {
			t.Fatalf("pre-dispatch validation mismatch: status=%d body=%s", response.Code, response.Body.String())
		}
		if fake.poolSubmitCountSnapshot() != 0 {
			t.Fatalf("selector submissions=%d, want 0", fake.poolSubmitCountSnapshot())
		}
	})

	t.Run("provider error before commit is JSON", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "anthropic")
		frames := make(chan *iop.ProviderTunnelFrame, 2)
		frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusBadGateway}
		frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true}
		close(frames)
		srv, fake := newHotPathHandlerServer(t, candidate, frames)
		response := serveHotPathAnthropic(t, srv, true)
		if response.Code != http.StatusBadGateway || !strings.Contains(response.Body.String(), `"type":"api_error"`) ||
			strings.Contains(response.Body.String(), "message_start") || strings.Contains(response.Body.String(), "message_stop") {
			t.Fatalf("pre-commit error mismatch: status=%d body=%s", response.Code, response.Body.String())
		}
		if fake.poolSubmitCountSnapshot() != 1 {
			t.Fatalf("selector submissions=%d, want 1", fake.poolSubmitCountSnapshot())
		}
		assertHotPathTerminal(t, srv)
	})

	t.Run("missing provider identity fails before commit", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "anthropic")
		body := []byte("data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"unsafe\"}}\n\n")
		srv, fake := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, "text/event-stream", body))
		response := serveHotPathAnthropic(t, srv, true)
		if response.Code != http.StatusBadGateway || !strings.Contains(response.Body.String(), `"type":"api_error"`) ||
			strings.Contains(response.Body.String(), "message_start") {
			t.Fatalf("missing-identity failure mismatch: status=%d body=%s", response.Code, response.Body.String())
		}
		if fake.poolSubmitCountSnapshot() != 1 {
			t.Fatalf("selector submissions=%d, want 1", fake.poolSubmitCountSnapshot())
		}
	})

	t.Run("conflicting provider identity fails after commit", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "anthropic")
		frames := make(chan *iop.ProviderTunnelFrame, 4)
		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: []byte(strings.Join([]string{
			`data: {"type":"message_start","message":{"id":"msg-first","usage":{"input_tokens":1}}}`,
			`data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":"visible"}}`, "",
		}, "\n\n"))}
		frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY,
			Body: []byte("data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg-conflict\",\"usage\":{\"input_tokens\":1}}}\n\n")}
		frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true}
		close(frames)
		srv, _ := newHotPathHandlerServer(t, candidate, frames)
		response := serveHotPathAnthropic(t, srv, true)
		events := decodeHotPathAnthropicSSE(t, response.Body.String())
		if got := hotPathAnthropicEventNames(events); strings.Join(got, ",") != "message_start,content_block_start,content_block_delta,error" ||
			strings.Contains(response.Body.String(), "msg-conflict") || strings.Contains(response.Body.String(), "message_stop") {
			t.Fatalf("conflicting-identity terminal mismatch: events=%v body=%s", got, response.Body.String())
		}
	})
}

func TestHotPathAnthropicFlushesBeforeEndAndErrorsAfterCommit(t *testing.T) {
	candidate := anthropicTestCandidate(t, "anthropic")
	frames := make(chan *iop.ProviderTunnelFrame, 4)
	frames <- &iop.ProviderTunnelFrame{
		Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK,
		Headers: map[string]string{"Content-Type": "text/event-stream"}, RunId: "run-anthropic-live",
	}
	frames <- &iop.ProviderTunnelFrame{
		Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, RunId: "run-anthropic-live",
		Body: []byte(strings.Join([]string{
			`data: {"type":"message_start","message":{"id":"msg-anthropic-live","usage":{"input_tokens":3,"output_tokens":0}}}`,
			`data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
			`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"early-visible"}}`, "",
		}, "\n\n")),
	}
	srv, fake := newHotPathHandlerServer(t, candidate, frames)
	httpServer := httptest.NewServer(srv.routes())
	defer httpServer.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	body := `{"model":"virtual-model","max_tokens":64,"messages":[{"role":"user","content":"flush"}],"stream":true}`
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, httpServer.URL+"/v1/messages", strings.NewReader(body))
	if err != nil {
		t.Fatal(err)
	}
	request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		t.Fatalf("stream request did not flush before provider END: %v", err)
	}
	defer response.Body.Close()
	reader := bufio.NewReader(response.Body)
	var early strings.Builder
	for range 3 {
		frame, err := readHotPathSSEFrame(reader)
		if err != nil {
			t.Fatalf("read pre-END Anthropic frame: %v", err)
		}
		early.WriteString(frame)
	}
	if response.StatusCode != http.StatusOK || !strings.Contains(early.String(), `"id":"msg-anthropic-live"`) ||
		!strings.Contains(early.String(), `"text":"early-visible"`) || strings.Contains(early.String(), "message_stop") {
		t.Fatalf("pre-END flush mismatch: status=%d body=%s", response.StatusCode, early.String())
	}
	frames <- &iop.ProviderTunnelFrame{
		Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, Error: "provider failed", RunId: "run-anthropic-live",
	}
	close(frames)
	rest, err := io.ReadAll(reader)
	if err != nil {
		t.Fatalf("read post-commit error: %v", err)
	}
	wire := early.String() + string(rest)
	events := decodeHotPathAnthropicSSE(t, wire)
	if got := hotPathAnthropicEventNames(events); strings.Join(got, ",") != "message_start,content_block_start,content_block_delta,error" ||
		strings.Count(wire, "event: error") != 1 || strings.Contains(wire, "message_delta") || strings.Contains(wire, "message_stop") {
		t.Fatalf("post-commit provider error mismatch: events=%v body=%s", got, wire)
	}
	if fake.poolSubmitCountSnapshot() != 1 {
		t.Fatalf("selector submissions=%d, want 1", fake.poolSubmitCountSnapshot())
	}
}

func serveHotPathAnthropicBody(t *testing.T, srv *Server, body string) *httptest.ResponseRecorder {
	t.Helper()
	request := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body))
	request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
	response := httptest.NewRecorder()
	srv.routes().ServeHTTP(response, request)
	return response
}

func decodeHotPathAnthropicSSE(t *testing.T, body string) []hotPathAnthropicSSEEvent {
	t.Helper()
	body = strings.ReplaceAll(body, "\r\n", "\n")
	var events []hotPathAnthropicSSEEvent
	for _, frame := range strings.Split(body, "\n\n") {
		frame = strings.TrimSpace(frame)
		if frame == "" {
			continue
		}
		var name string
		var data []string
		for _, line := range strings.Split(frame, "\n") {
			switch {
			case strings.HasPrefix(line, "event:"):
				name = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
			case strings.HasPrefix(line, "data:"):
				data = append(data, strings.TrimSpace(strings.TrimPrefix(line, "data:")))
			}
		}
		if name == "" || len(data) == 0 {
			t.Fatalf("malformed Anthropic SSE frame %q", frame)
		}
		var payload map[string]any
		if err := json.Unmarshal([]byte(strings.Join(data, "\n")), &payload); err != nil {
			t.Fatalf("decode Anthropic SSE %q: %v", frame, err)
		}
		if payload["type"] != name {
			t.Fatalf("event/type mismatch: event=%q payload=%+v", name, payload)
		}
		events = append(events, hotPathAnthropicSSEEvent{name: name, payload: payload})
	}
	return events
}

func hotPathAnthropicEventNames(events []hotPathAnthropicSSEEvent) []string {
	names := make([]string, 0, len(events))
	for _, event := range events {
		names = append(names, event.name)
	}
	return names
}

func hotPathAnthropicMap(t *testing.T, value any) map[string]any {
	t.Helper()
	mapped, ok := value.(map[string]any)
	if !ok {
		t.Fatalf("value is not an object: %#v", value)
	}
	return mapped
}

func assertHotPathAnthropicDirectEvents(t *testing.T, events []hotPathAnthropicSSEEvent, responseID, signature string) {
	t.Helper()
	assertHotPathAnthropicBlockIndexes(t, events, 3)
	wantNames := []string{
		"message_start",
		"content_block_start", "content_block_delta", "content_block_delta",
	}
	if signature != "" {
		wantNames = append(wantNames, "content_block_delta")
	}
	wantNames = append(wantNames,
		"content_block_stop",
		"content_block_start", "content_block_delta", "content_block_delta", "content_block_stop",
		"content_block_start", "content_block_delta", "content_block_delta", "content_block_stop",
		"message_delta", "message_stop",
	)
	if got := hotPathAnthropicEventNames(events); strings.Join(got, ",") != strings.Join(wantNames, ",") {
		t.Fatalf("event order=%v, want %v", got, wantNames)
	}
	message := hotPathAnthropicMap(t, events[0].payload["message"])
	if message["id"] != responseID || message["model"] != "virtual-model" {
		t.Fatalf("message_start mismatch: %+v", message)
	}
	if signature != "" {
		startUsage := hotPathAnthropicMap(t, message["usage"])
		if startUsage["input_tokens"] != float64(9) {
			t.Fatalf("message_start usage mismatch: %+v", message)
		}
	}

	wantKinds := []string{"thinking", "text", "tool_use"}
	var kinds, thinking, text, toolFragments []string
	var toolID, toolName, stopReason, gotSignature string
	for _, event := range events {
		switch event.name {
		case "content_block_start":
			block := hotPathAnthropicMap(t, event.payload["content_block"])
			kinds = append(kinds, fmt.Sprint(block["type"]))
			if block["type"] == "tool_use" {
				toolID, _ = block["id"].(string)
				toolName, _ = block["name"].(string)
			}
		case "content_block_delta":
			delta := hotPathAnthropicMap(t, event.payload["delta"])
			switch delta["type"] {
			case "thinking_delta":
				thinking = append(thinking, fmt.Sprint(delta["thinking"]))
			case "text_delta":
				text = append(text, fmt.Sprint(delta["text"]))
			case "input_json_delta":
				toolFragments = append(toolFragments, fmt.Sprint(delta["partial_json"]))
			case "signature_delta":
				gotSignature, _ = delta["signature"].(string)
			}
		case "message_delta":
			delta := hotPathAnthropicMap(t, event.payload["delta"])
			stopReason, _ = delta["stop_reason"].(string)
			usage := hotPathAnthropicMap(t, event.payload["usage"])
			if usage["input_tokens"] != float64(9) || usage["output_tokens"] != float64(7) {
				t.Fatalf("terminal usage=%+v, want input=9 output=7", usage)
			}
		}
	}
	if strings.Join(kinds, ",") != strings.Join(wantKinds, ",") || strings.Join(thinking, "") != "plan now" ||
		strings.Join(text, "") != "alpha omega" || strings.Join(toolFragments, "") != `{"path":"README.md"}` ||
		len(toolFragments) != 2 || toolID != responseID+"-tool-1" || toolName != "read_file" ||
		stopReason != "tool_use" || gotSignature != signature {
		t.Fatalf("stream aggregate mismatch: kinds=%v thinking=%v text=%v tool=%q/%q/%v stop=%q signature=%q",
			kinds, thinking, text, toolID, toolName, toolFragments, stopReason, gotSignature)
	}
}

func assertHotPathAnthropicBlockIndexes(t *testing.T, events []hotPathAnthropicSSEEvent, wantBlocks int) {
	t.Helper()
	nextStart := 0
	active := -1
	for _, event := range events {
		switch event.name {
		case "content_block_start":
			index := int(event.payload["index"].(float64))
			if active != -1 || index != nextStart {
				t.Fatalf("non-monotonic block start: active=%d index=%d next=%d", active, index, nextStart)
			}
			active = index
			nextStart++
		case "content_block_delta":
			index := int(event.payload["index"].(float64))
			if index != active {
				t.Fatalf("block delta index=%d, active=%d", index, active)
			}
		case "content_block_stop":
			index := int(event.payload["index"].(float64))
			if index != active {
				t.Fatalf("block stop index=%d, active=%d", index, active)
			}
			active = -1
		}
	}
	if active != -1 || nextStart != wantBlocks {
		t.Fatalf("block boundary mismatch: active=%d starts=%d want=%d", active, nextStart, wantBlocks)
	}
}