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

package openai

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

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

func TestHotPathSelectorInstruction(t *testing.T) {
	prepareInstruction, err := buildHotPathSelectorProviderInstruction("req_wire_123", selectorInstructionPrepareOnly)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(prepareInstruction, "Operation: prepare-only") ||
		!strings.Contains(prepareInstruction, "JOB directory: .iop/job/req_wire_123") ||
		strings.Contains(prepareInstruction, "PLAN path:") || strings.Contains(prepareInstruction, "REVIEW path:") {
		t.Fatalf("prepare-only selector instruction is not operation-bounded: %q", prepareInstruction)
	}
	pairInstruction, err := buildHotPathSelectorProviderInstruction("req_wire_123", selectorInstructionPairWrite)
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(pairInstruction, "Operation: pair-write") ||
		!strings.Contains(pairInstruction, "PLAN path: .iop/job/req_wire_123/plan.md") ||
		!strings.Contains(pairInstruction, "REVIEW path: .iop/job/req_wire_123/review.md") {
		t.Fatalf("pair-write selector instruction is incomplete: %q", pairInstruction)
	}

	t.Run("chat_last_leading_system", func(t *testing.T) {
		original := []byte(`{"model":"selector","messages":[{"role":"system","content":"caller-system"},{"role":"user","content":"task"}],"caller_extension":{"kept":true}}`)
		body, err := injectHotPathChatSelectorInstruction(original, pairInstruction)
		if err != nil {
			t.Fatal(err)
		}
		var payload struct {
			Messages []struct {
				Role    string `json:"role"`
				Content string `json:"content"`
			} `json:"messages"`
			CallerExtension map[string]bool `json:"caller_extension"`
		}
		if err := json.Unmarshal(body, &payload); err != nil {
			t.Fatal(err)
		}
		if len(payload.Messages) != 3 || payload.Messages[0].Content != "caller-system" || payload.Messages[1].Content != pairInstruction || payload.Messages[2].Role != "user" {
			t.Fatalf("unexpected Chat selector authority ordering: %+v", payload.Messages)
		}
		if !payload.CallerExtension["kept"] || strings.Contains(string(original), "req_wire_123") {
			t.Fatalf("caller body was not preserved independently: original=%s provider=%s", original, body)
		}
	})

	t.Run("anthropic_appended_system_block", func(t *testing.T) {
		original := []byte(`{"model":"selector","system":"caller-system","messages":[{"role":"user","content":"task"}],"caller_extension":{"kept":true}}`)
		body, err := injectHotPathAnthropicSelectorInstruction(original, pairInstruction)
		if err != nil {
			t.Fatal(err)
		}
		var payload struct {
			System []struct {
				Type string `json:"type"`
				Text string `json:"text"`
			} `json:"system"`
			CallerExtension map[string]bool `json:"caller_extension"`
		}
		if err := json.Unmarshal(body, &payload); err != nil {
			t.Fatal(err)
		}
		if len(payload.System) != 2 || payload.System[0].Text != "caller-system" || payload.System[1].Type != "text" || payload.System[1].Text != pairInstruction {
			t.Fatalf("unexpected Messages selector authority ordering: %+v", payload.System)
		}
		if !payload.CallerExtension["kept"] || strings.Contains(string(original), "req_wire_123") {
			t.Fatalf("caller body was not preserved independently: original=%s provider=%s", original, body)
		}
	})

	t.Run("frontier_state_tracks_binding_and_phase", func(t *testing.T) {
		prepareFixture := newArtifactPairFixture(t, "openai", false)
		if got := prepareFixture.server.artifactFrontiers.selectorInstructionState(prepareFixture.requestID, prepareFixture.ownerEdgeID); got != selectorInstructionPrepareOnly {
			t.Fatalf("initial non-parent-creating state = %q", got)
		}
		prepareIDs := prepareFixture.issuePrepare()
		if _, _, _, err := prepareFixture.continueWithResult([]artifactTestResult{{id: prepareIDs[0], body: `{"written":true}`}}, nil); err != nil {
			t.Fatal(err)
		}
		if got := prepareFixture.server.artifactFrontiers.selectorInstructionState(prepareFixture.requestID, prepareFixture.ownerEdgeID); got != selectorInstructionPairWrite {
			t.Fatalf("resumed state = %q", got)
		}

		pairFixture := newArtifactPairFixture(t, "anthropic", true)
		if got := pairFixture.server.artifactFrontiers.selectorInstructionState(pairFixture.requestID, pairFixture.ownerEdgeID); got != selectorInstructionPairWrite {
			t.Fatalf("initial parent-creating state = %q", got)
		}
	})

	t.Run("marked_single_request_omitted", func(t *testing.T) {
		original := []byte(`{"model":"stage-model","messages":[{"role":"user","content":"private stage"}]}`)
		tunnel := edgeservice.SubmitProviderTunnelRequest{
			Operation: string(config.OperationChatCompletions),
			BuildBody: func(string) ([]byte, error) { return append([]byte(nil), original...), nil },
		}
		prepared, err := prepareHotPathSelectorProviderInstruction(tunnel, "")
		if err != nil {
			t.Fatal(err)
		}
		body, err := prepared.BuildBody("stage-model")
		if err != nil || string(body) != string(original) || strings.Contains(string(body), "IOP caller-workspace selector instruction") {
			t.Fatalf("marked single-request provider body changed: err=%v body=%s", err, body)
		}
	})
}

func TestHotPathDirect(t *testing.T) {
	srv := NewServer(config.EdgeOpenAIConf{}, nil, nil)
	srv.SetEdgeID("edge-direct-test")
	snapshot, err := srv.requestCoordinator.create(logicalRequestAdmission{
		OwnerEdgeID: srv.edgeIDValue(), PrincipalRef: "principal-1",
		Lineage:          logicalRequestLineage{Endpoint: logicalRequestEndpointChat, HistoryDigest: "history", ToolsetDigest: "tools"},
		PresetGeneration: "preset-generation",
	})
	if err != nil {
		t.Fatal(err)
	}
	stageID, _ := srv.requestCoordinator.newStageID()
	if _, err := srv.requestCoordinator.activateStage(snapshot.ID, srv.edgeIDValue(), stageID); err != nil {
		t.Fatal(err)
	}
	recorder := httptest.NewRecorder()
	turn := &hotPathTurn{
		RequestID: snapshot.ID, StageID: stageID, OwnerEdgeID: srv.edgeIDValue(), Protocol: "openai",
		PublicModelID: "virtual-model", Writer: recorder,
	}
	output := normalizedStageOutput{
		ResponseID: "chatcmpl-provider-tool", Created: 1_777_000_001, TerminalReason: "tool_calls",
		ToolCalls: []normalizedToolCall{{
			ID: "call_public_1", ProviderCallID: "call_provider_1", Name: "read_file",
			Arguments: map[string]any{"path": "README.md"}, RawArgs: `{"path":"README.md"}`,
		}},
		Usage: json.RawMessage(`{"prompt_tokens":13,"completion_tokens":5,"total_tokens":18}`),
	}
	if err := srv.runDirectTurn(context.Background(), turn, output); err != nil {
		t.Fatalf("runDirectTurn: %v", err)
	}
	if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "chatcmpl-provider-tool") || strings.Contains(recorder.Body.String(), ".iop/job/") {
		t.Fatalf("unexpected direct response: status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	wantHash, err := directIssuedCallHash("openai", output)
	if err != nil {
		t.Fatal(err)
	}
	srv.requestCoordinator.mu.Lock()
	record := srv.requestCoordinator.requests[snapshot.ID]
	gotProvider := record.publicToProvider["call_public_1"]
	gotHash := record.expectedIssuedCallHash
	state := record.state
	srv.requestCoordinator.mu.Unlock()
	if state != logicalRequestStateWaiting || gotProvider != "call_provider_1" || gotHash != wantHash {
		t.Fatalf("frontier mismatch: state=%q provider=%q hash=%q wantHash=%q", state, gotProvider, gotHash, wantHash)
	}
}

func TestDirectTurnReleasesArtifactFrontier(t *testing.T) {
	srv := NewServer(config.EdgeOpenAIConf{}, nil, nil)
	srv.SetEdgeID("edge-direct-artifact-test")
	srv.artifactFrontiers = newArtifactFrontierStore(1)
	binding := mustBinding(t, workspaceAlternative("direct-artifact", "write_file", false, true), []any{openAIChatTool("write_file", structuredSchema())})

	newTurn := func(t *testing.T) *hotPathTurn {
		t.Helper()
		lineage := logicalRequestLineage{Endpoint: logicalRequestEndpointChat, HistoryDigest: "history", ToolsetDigest: "tools"}
		snapshot, err := srv.requestCoordinator.create(logicalRequestAdmission{
			OwnerEdgeID: srv.edgeIDValue(), PrincipalRef: "principal-direct-artifact",
			Lineage:          lineage,
			PresetGeneration: "preset-generation",
		})
		if err != nil {
			t.Fatal(err)
		}
		stageID, err := srv.requestCoordinator.newStageID()
		if err != nil {
			t.Fatal(err)
		}
		if _, err := srv.requestCoordinator.activateStage(snapshot.ID, srv.edgeIDValue(), stageID); err != nil {
			t.Fatal(err)
		}
		if err := srv.artifactFrontiers.pin(snapshot.ID, srv.edgeIDValue(), "principal-direct-artifact", "openai", stageID, lineage, binding); err != nil {
			t.Fatalf("pin artifact frontier: %v", err)
		}
		return &hotPathTurn{RequestID: snapshot.ID, StageID: stageID, OwnerEdgeID: srv.edgeIDValue(), Protocol: "openai", PublicModelID: "virtual-model", Writer: httptest.NewRecorder()}
	}

	for range 3 {
		turn := newTurn(t)
		if err := srv.runDirectTurn(context.Background(), turn, normalizedStageOutput{ResponseID: "chatcmpl-direct-terminal", Content: "done"}); err != nil {
			t.Fatalf("complete no-tool direct turn: %v", err)
		}
		if _, err := srv.requestCoordinator.snapshot(turn.RequestID); !errors.Is(err, errLogicalRequestNotFound) {
			t.Fatalf("direct terminal retained coordinator state: %v", err)
		}
		if srv.artifactFrontiers.pairRequired(turn.RequestID, turn.OwnerEdgeID) {
			t.Fatal("completed direct turn retained a pair-required artifact frontier")
		}
		srv.artifactFrontiers.mu.Lock()
		_, retained := srv.artifactFrontiers.records[turn.RequestID]
		srv.artifactFrontiers.mu.Unlock()
		if retained {
			t.Fatal("completed no-tool direct turn retained its artifact frontier")
		}
	}

	waiting := newTurn(t)
	waitingOutput := normalizedStageOutput{ResponseID: "chatcmpl-direct-tool", ToolCalls: []normalizedToolCall{{ID: "call_waiting", Name: "read_file", Arguments: map[string]any{"path": "README.md"}}}}
	if err := srv.runDirectTurn(context.Background(), waiting, waitingOutput); err != nil {
		t.Fatalf("issue ordinary direct tool: %v", err)
	}
	srv.artifactFrontiers.mu.Lock()
	_, retained := srv.artifactFrontiers.records[waiting.RequestID]
	srv.artifactFrontiers.mu.Unlock()
	if !retained {
		t.Fatal("ordinary direct tool turn unexpectedly released its artifact frontier")
	}
}

func TestArtifactPairHandlerDisposition(t *testing.T) {
	for _, endpoint := range []string{"openai", "anthropic"} {
		endpoint := endpoint
		t.Run(endpoint+" prepare resumes selector and pair reaches local handoff", func(t *testing.T) {
			candidate := anthropicTestCandidate(t, map[string]string{"openai": "openai", "anthropic": "anthropic"}[endpoint])
			service := &scriptedArtifactPoolService{endpoint: endpoint, candidate: candidate}
			service.selectorResponse = func(requestID string, state selectorInstructionState, call int) string {
				switch state {
				case selectorInstructionPrepareOnly:
					if call != 1 {
						t.Fatalf("prepare-only instruction arrived on call %d", call)
					}
					return scriptedArtifactPrepare(endpoint, requestID)
				case selectorInstructionPairWrite:
					if call != 2 {
						t.Fatalf("pair-write instruction arrived on call %d", call)
					}
					return scriptedArtifactPair(endpoint, requestID)
				case selectorInstructionNone:
					if call != 3 {
						t.Fatalf("instruction-free local handoff arrived on call %d", call)
					}
					return scriptedArtifactLocalRead(endpoint, requestID)
				default:
					t.Fatalf("unexpected selector instruction state %q", state)
					return ""
				}
			}
			srv := newScriptedArtifactHandlerServer(t, service)
			tools := scriptedArtifactTools(endpoint)
			history := []any{map[string]any{"role": "user", "content": "Working directory: /workspace\nwrite a plan"}}

			first := serveScriptedArtifactRequest(t, srv, endpoint, scriptedArtifactRequestBody(t, endpoint, tools, history))
			if first.Code != http.StatusOK || service.calls != 1 {
				t.Fatalf("prepare response: status=%d calls=%d body=%s", first.Code, service.calls, first.Body.String())
			}
			assistant, prepareIDs, err := artifactAssistantFromResponse(endpoint, first.Body.Bytes())
			if err != nil || len(prepareIDs) != 1 {
				t.Fatalf("decode prepare response: ids=%v err=%v", prepareIDs, err)
			}
			history = append(history, assistant)
			history = scriptedArtifactAppendResults(endpoint, history, prepareIDs, []string{`{"written":true}`})

			second := serveScriptedArtifactRequest(t, srv, endpoint, scriptedArtifactRequestBody(t, endpoint, tools, history))
			if second.Code != http.StatusOK || service.calls != 2 {
				t.Fatalf("pair response: status=%d calls=%d body=%s", second.Code, service.calls, second.Body.String())
			}
			assistant, pairIDs, err := artifactAssistantFromResponse(endpoint, second.Body.Bytes())
			if err != nil || len(pairIDs) != 2 {
				t.Fatalf("decode pair response: ids=%v err=%v", pairIDs, err)
			}
			history = append(history, assistant)
			history = scriptedArtifactAppendResults(endpoint, history, pairIDs, []string{`{"written":true}`, `{"written":true}`})

			third := serveScriptedArtifactRequest(t, srv, endpoint, scriptedArtifactRequestBody(t, endpoint, tools, history))
			if third.Code != http.StatusOK || service.calls != 3 {
				t.Fatalf("local handoff: status=%d calls=%d body=%s", third.Code, service.calls, third.Body.String())
			}
			if !strings.Contains(third.Body.String(), "read_file") {
				t.Fatalf("local handoff did not expose the caller tool: %s", third.Body.String())
			}
		})

		t.Run(endpoint+" rejects pair-write output for prepare-only instruction", func(t *testing.T) {
			candidate := anthropicTestCandidate(t, map[string]string{"openai": "openai", "anthropic": "anthropic"}[endpoint])
			service := &scriptedArtifactPoolService{endpoint: endpoint, candidate: candidate}
			service.selectorResponse = func(requestID string, state selectorInstructionState, call int) string {
				if state != selectorInstructionPrepareOnly || call != 1 {
					t.Fatalf("unexpected swapped-operation setup: state=%q call=%d", state, call)
				}
				return scriptedArtifactPair(endpoint, requestID)
			}
			srv := newScriptedArtifactHandlerServer(t, service)
			response := serveScriptedArtifactRequest(t, srv, endpoint, scriptedArtifactRequestBody(t, endpoint, scriptedArtifactTools(endpoint), []any{map[string]any{"role": "user", "content": "write a plan"}}))
			if response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), "prepare turn must contain exactly one call") {
				t.Fatalf("prepare-only accepted pair-write output: status=%d body=%s", response.Code, response.Body.String())
			}
		})

		t.Run(endpoint+" rejects prepare output for pair-write instruction", func(t *testing.T) {
			candidate := anthropicTestCandidate(t, map[string]string{"openai": "openai", "anthropic": "anthropic"}[endpoint])
			service := &scriptedArtifactPoolService{endpoint: endpoint, candidate: candidate}
			service.selectorResponse = func(requestID string, state selectorInstructionState, call int) string {
				switch state {
				case selectorInstructionPrepareOnly, selectorInstructionPairWrite:
					return scriptedArtifactPrepare(endpoint, requestID)
				default:
					t.Fatalf("unexpected instruction-free selector call %d", call)
					return ""
				}
			}
			srv := newScriptedArtifactHandlerServer(t, service)
			tools := scriptedArtifactTools(endpoint)
			history := []any{map[string]any{"role": "user", "content": "write a plan"}}
			first := serveScriptedArtifactRequest(t, srv, endpoint, scriptedArtifactRequestBody(t, endpoint, tools, history))
			assistant, prepareIDs, err := artifactAssistantFromResponse(endpoint, first.Body.Bytes())
			if first.Code != http.StatusOK || err != nil || len(prepareIDs) != 1 {
				t.Fatalf("prepare response: status=%d ids=%v err=%v body=%s", first.Code, prepareIDs, err, first.Body.String())
			}
			history = append(history, assistant)
			history = scriptedArtifactAppendResults(endpoint, history, prepareIDs, []string{`{"written":true}`})
			second := serveScriptedArtifactRequest(t, srv, endpoint, scriptedArtifactRequestBody(t, endpoint, tools, history))
			if second.Code != http.StatusBadRequest || service.calls != 2 || !strings.Contains(second.Body.String(), "pair turn must contain exactly two calls") {
				t.Fatalf("pair-write accepted prepare output: status=%d calls=%d body=%s", second.Code, service.calls, second.Body.String())
			}
		})
	}
}

type scriptedArtifactPoolService struct {
	providerFakeRunService
	endpoint         string
	candidate        edgeservice.ProviderPoolCandidate
	calls            int
	response         func(requestID string, call int) string
	selectorResponse func(requestID string, state selectorInstructionState, call int) string
}

func (s *scriptedArtifactPoolService) SubmitProviderPool(_ context.Context, req edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
	s.calls++
	providerBody, operation, err := materializeScriptedProviderBody(req, s.candidate)
	if err != nil {
		return nil, err
	}
	var requestID string
	instructionState := selectorInstructionNone
	if strings.Contains(string(providerBody), "IOP caller-workspace selector instruction") {
		requestID, instructionState, err = scriptedSelectorDirective(providerBody, operation)
	} else {
		requestID, err = scriptedRequestIDFromProviderBody(providerBody)
	}
	if err != nil {
		return nil, err
	}
	var body string
	if s.selectorResponse != nil {
		body = s.selectorResponse(requestID, instructionState, s.calls)
	} else if s.response != nil {
		body = s.response(requestID, s.calls)
	} else {
		return nil, fmt.Errorf("scripted provider response is unavailable")
	}
	dispatch := edgeservice.RunDispatch{
		RunID: fmt.Sprintf("run-scripted-%d", s.calls), NodeID: "node-scripted", ModelGroupKey: req.Run.ModelGroupKey,
		ProviderID: s.candidate.ProviderID, ExecutionPath: string(edgeservice.ProviderPoolPathTunnel),
		ProfileID: s.candidate.ProfileID, ProfileDriver: s.candidate.ProfileDriver, ProfileCapabilities: append([]string(nil), s.candidate.ProfileCapabilities...),
	}
	frames := staticProviderTunnelFrames(body)
	if s.endpoint == "anthropic" {
		frames = anthropicTunnelFrames(http.StatusOK, "application/json", []byte(body))
	}
	return &edgeservice.ProviderPoolDispatchResult{
		Path:         edgeservice.ProviderPoolPathTunnel,
		Tunnel:       &fakeTunnelHandle{dispatch: dispatch, frames: frames},
		DispatchInfo: dispatch,
	}, nil
}

func materializeScriptedProviderBody(req edgeservice.ProviderPoolDispatchRequest, candidate edgeservice.ProviderPoolCandidate) ([]byte, string, error) {
	prepared := req.Tunnel
	var err error
	if req.PrepareProtocolTunnel != nil {
		prepared, err = req.PrepareProtocolTunnel(prepared, candidate)
		if err != nil {
			return nil, "", err
		}
	}
	if prepared.BuildBody != nil {
		target := candidate.ActualModel
		if strings.TrimSpace(target) == "" {
			target = "served-selector"
		}
		body, err := prepared.BuildBody(target)
		return body, prepared.Operation, err
	}
	if len(prepared.Body) == 0 {
		return nil, "", fmt.Errorf("scripted provider body is unavailable")
	}
	return append([]byte(nil), prepared.Body...), prepared.Operation, nil
}

func scriptedSelectorDirective(providerBody []byte, operation string) (string, selectorInstructionState, error) {
	instruction, err := scriptedSelectorInstruction(providerBody, operation)
	if err != nil {
		return "", selectorInstructionNone, err
	}
	state := selectorInstructionNone
	pathPrefix := ""
	switch {
	case strings.Contains(instruction, "Operation: prepare-only"):
		state = selectorInstructionPrepareOnly
		pathPrefix = "JOB directory: .iop/job/"
	case strings.Contains(instruction, "Operation: pair-write"):
		state = selectorInstructionPairWrite
		pathPrefix = "PLAN path: .iop/job/"
	default:
		return "", selectorInstructionNone, fmt.Errorf("provider selector instruction is missing a closed operation")
	}
	start := strings.Index(instruction, pathPrefix)
	if start < 0 {
		return "", selectorInstructionNone, fmt.Errorf("provider selector instruction is missing operation path")
	}
	start += len(pathPrefix)
	remainder := instruction[start:]
	end := strings.IndexAny(remainder, "/\n\r\t ")
	if end < 0 {
		end = len(remainder)
	}
	if end == 0 {
		return "", selectorInstructionNone, fmt.Errorf("provider selector instruction has an invalid operation path")
	}
	requestID := remainder[:end]
	paths := newReservedPaths(requestID)
	if state == selectorInstructionPrepareOnly {
		required := []string{
			"Return exactly one admitted prepare tool call",
			"JOB directory: " + paths.JobDir,
			"Do not write PLAN or REVIEW in this turn.",
		}
		for _, fragment := range required {
			if !strings.Contains(instruction, fragment) {
				return "", selectorInstructionNone, fmt.Errorf("provider prepare-only instruction is missing fragment %q", fragment)
			}
		}
		if strings.Contains(instruction, "PLAN path:") || strings.Contains(instruction, "REVIEW path:") {
			return "", selectorInstructionNone, fmt.Errorf("provider prepare-only instruction contains pair paths")
		}
		return requestID, state, nil
	}
	required := []string{
		"Return exactly one iop_write_artifact_pair tool call",
		"PLAN path: " + paths.PlanPath,
		"REVIEW path: " + paths.ReviewPath,
		"You are the Planner.",
		"Analyze the immutable user task first.",
		"Preserve every explicit requirement, constraint, deliverable, and acceptance condition.",
		"# Plan\n\n## Goal\n" + hotPathCallerWorkspacePlanBoundary + " <one non-empty task goal on the same line>\n\n## Steps\n- [P1] <non-empty one-line step>\n- [P2] <non-empty one-line step>",
		"The pending REVIEW must use exactly the Plan's P1..Pn inventory in order",
		"# Review\n\n## Worker Item Status\n- P1: pending\n- P2: pending",
		"## Worker Changes\nPending worker execution.",
		"## Worker Verification\nPending worker verification.",
		"## Deviations\nNone recorded.",
		"Do not copy the explanatory optional-status text into REVIEW.",
	}
	for _, fragment := range required {
		if !strings.Contains(instruction, fragment) {
			return "", selectorInstructionNone, fmt.Errorf("provider pair-write instruction is missing grammar fragment %q", fragment)
		}
	}
	return requestID, state, nil
}

func scriptedSelectorRequestID(providerBody []byte, operation string) (string, error) {
	requestID, _, err := scriptedSelectorDirective(providerBody, operation)
	return requestID, err
}

func scriptedSelectorInstruction(providerBody []byte, operation string) (string, error) {
	switch config.ProtocolOperation(operation) {
	case config.OperationChatCompletions:
		var payload struct {
			Messages []struct {
				Role    string `json:"role"`
				Content any    `json:"content"`
			} `json:"messages"`
		}
		if err := json.Unmarshal(providerBody, &payload); err != nil {
			return "", err
		}
		for _, message := range payload.Messages {
			if message.Role != "system" {
				break
			}
			text := extractMessageContentString(message.Content)
			if strings.Contains(text, "IOP caller-workspace selector instruction") {
				return text, nil
			}
		}
	case config.OperationMessages:
		var payload struct {
			System any `json:"system"`
		}
		if err := json.Unmarshal(providerBody, &payload); err != nil {
			return "", err
		}
		text := extractMessageContentString(payload.System)
		if strings.Contains(text, "IOP caller-workspace selector instruction") {
			return text, nil
		}
	default:
		return "", fmt.Errorf("unexpected scripted selector operation %q", operation)
	}
	return "", fmt.Errorf("actual provider body is missing the selector instruction")
}

func scriptedRequestIDFromProviderBody(providerBody []byte) (string, error) {
	const prefix = ".iop/job/"
	text := string(providerBody)
	start := strings.Index(text, prefix)
	if start < 0 {
		return "", fmt.Errorf("actual provider body is missing a request-local artifact path")
	}
	start += len(prefix)
	end := strings.IndexByte(text[start:], '/')
	if end <= 0 {
		return "", fmt.Errorf("actual provider body has an invalid request-local artifact path")
	}
	return text[start : start+end], nil
}

func newScriptedArtifactHandlerServer(t *testing.T, service *scriptedArtifactPoolService) *Server {
	t.Helper()
	preset := hotPathSelectorPreset([]string{config.ModeDirect, config.ModeLight})
	preset.WorkspaceTools = []config.ExecutionWorkspaceToolAlternative{{
		Name: "scripted-fs",
		Operations: map[string]config.ExecutionWorkspaceOperation{
			"prepare": {ToolName: "mkdir_p", SchemaMatcher: map[string]any{"type": "object"}, ArgumentMap: map[string]any{"path": "path"}, ResultMatcher: successMatcher(), CreatesParents: true},
			"read":    {ToolName: "read_file", SchemaMatcher: map[string]any{"type": "object"}, ArgumentMap: map[string]any{"path": "path"}, ResultMatcher: successMatcher()},
			"write":   {ToolName: "write_file", SchemaMatcher: map[string]any{"type": "object"}, ArgumentMap: map[string]any{"path": "path", "content": "content"}, ResultMatcher: successMatcher(), CreatesParents: false},
			"delete":  {ToolName: "delete_file", SchemaMatcher: map[string]any{"type": "object"}, ArgumentMap: map[string]any{"path": "path"}, ResultMatcher: successMatcher()},
		},
	}}
	srv := NewServer(config.EdgeOpenAIConf{}, service, nil)
	srv.SetEdgeID("edge-scripted-artifact")
	srv.SetExecutionPresets([]config.ExecutionPreset{preset})
	srv.SetModelCatalog([]config.ModelCatalogEntry{
		{ID: "virtual-model", ExecutionPreset: preset.ID},
		{ID: "selector-model", Providers: map[string]string{service.candidate.ProviderID: "served-selector"}},
		{ID: "local-model", Providers: map[string]string{service.candidate.ProviderID: "served-local"}},
		{ID: "review-model", Providers: map[string]string{service.candidate.ProviderID: "served-review"}},
	})
	return srv
}

func scriptedArtifactTools(endpoint string) []any {
	schema := map[string]any{"type": "object", "properties": map[string]any{"path": map[string]any{"type": "string"}, "content": map[string]any{}}, "required": []any{"path"}}
	if endpoint == "anthropic" {
		return []any{anthropicWorkspaceTool("mkdir_p", schema), anthropicWorkspaceTool("read_file", schema), anthropicWorkspaceTool("write_file", schema), anthropicWorkspaceTool("delete_file", schema)}
	}
	return []any{openAIChatTool("mkdir_p", schema), openAIChatTool("read_file", schema), openAIChatTool("write_file", schema), openAIChatTool("delete_file", schema)}
}

func scriptedArtifactRequestBody(t *testing.T, endpoint string, tools, history []any) []byte {
	return scriptedArtifactRequestBodyWithOptions(t, endpoint, tools, history, 0, false)
}

func scriptedArtifactRequestBodyWithOptions(t *testing.T, endpoint string, tools, history []any, outputCap int, stream bool) []byte {
	t.Helper()
	envelope := map[string]any{"model": "virtual-model", "messages": history, "tools": tools, "stream": stream}
	if endpoint == "anthropic" {
		if outputCap <= 0 {
			outputCap = 64
		}
		envelope["max_tokens"] = outputCap
	} else if outputCap > 0 {
		envelope["max_tokens"] = outputCap
	}
	body, err := json.Marshal(envelope)
	if err != nil {
		t.Fatal(err)
	}
	return body
}

func serveScriptedArtifactRequest(t *testing.T, srv *Server, endpoint string, body []byte) *httptest.ResponseRecorder {
	return serveScriptedArtifactRequestContext(t, srv, endpoint, body, context.Background())
}

func serveScriptedArtifactRequestContext(t *testing.T, srv *Server, endpoint string, body []byte, ctx context.Context) *httptest.ResponseRecorder {
	t.Helper()
	path := "/v1/chat/completions"
	if endpoint == "anthropic" {
		path = "/v1/messages"
	}
	request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(string(body))).WithContext(ctx)
	if endpoint == "anthropic" {
		request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
	}
	recorder := httptest.NewRecorder()
	srv.routes().ServeHTTP(recorder, request)
	return recorder
}

func scriptedArtifactAppendResults(endpoint string, history []any, ids, bodies []string) []any {
	if endpoint == "anthropic" {
		blocks := make([]any, 0, len(ids))
		for index, id := range ids {
			blocks = append(blocks, map[string]any{"type": "tool_result", "tool_use_id": id, "content": bodies[index]})
		}
		return append(history, map[string]any{"role": "user", "content": blocks})
	}
	for index, id := range ids {
		history = append(history, map[string]any{"role": "tool", "tool_call_id": id, "content": bodies[index]})
	}
	return history
}

func scriptedArtifactPrepare(endpoint, requestID string) string {
	path := newReservedPaths(requestID).JobDir
	if endpoint == "anthropic" {
		return fmt.Sprintf(`{"id":"msg-scripted","type":"message","role":"assistant","content":[{"type":"tool_use","id":"provider-prepare","name":"mkdir_p","input":{"path":%q}}],"stop_reason":"tool_use"}`, path)
	}
	arguments, _ := json.Marshal(map[string]string{"path": path})
	return fmt.Sprintf(`{"id":"chatcmpl-scripted","created":1,"choices":[{"message":{"role":"assistant","tool_calls":[{"id":"provider-prepare","type":"function","function":{"name":"mkdir_p","arguments":%q}}]},"finish_reason":"tool_calls"}]}`, string(arguments))
}

func scriptedArtifactPair(endpoint, requestID string) string {
	paths := newReservedPaths(requestID)
	plan, _ := singlerequesttemplate.RenderPlan(singlerequesttemplate.DefaultPlanTemplate, singlerequesttemplate.PlanFields{
		Goal: "Complete the caller workspace task", Steps: []string{"Inspect the requested result", "Implement and verify the result"},
		Verification: []string{"Confirm the caller workspace result"},
	}, singlerequesttemplate.MaxTemplateBytes)
	review, _ := singlerequesttemplate.RenderReview(singlerequesttemplate.DefaultReviewTemplate, singlerequesttemplate.ReviewFields{
		ItemStatus: "- P1: pending\n- P2: pending", Changes: "Pending worker execution.",
		Verification: "Pending worker verification.", Deviations: "None recorded.",
	}, singlerequesttemplate.MaxTemplateBytes)
	if endpoint == "anthropic" {
		return fmt.Sprintf(`{"id":"msg-scripted-pair","type":"message","role":"assistant","content":[{"type":"tool_use","id":"provider-plan","name":"write_file","input":{"path":%q,"content":%q}},{"type":"tool_use","id":"provider-review","name":"write_file","input":{"path":%q,"content":%q}}],"stop_reason":"tool_use"}`, paths.PlanPath, string(plan), paths.ReviewPath, string(review))
	}
	planArgs, _ := json.Marshal(map[string]string{"path": paths.PlanPath, "content": string(plan)})
	reviewArgs, _ := json.Marshal(map[string]string{"path": paths.ReviewPath, "content": string(review)})
	return fmt.Sprintf(`{"id":"chatcmpl-scripted-pair","created":2,"choices":[{"message":{"role":"assistant","tool_calls":[{"id":"provider-plan","type":"function","function":{"name":"write_file","arguments":%q}},{"id":"provider-review","type":"function","function":{"name":"write_file","arguments":%q}}]},"finish_reason":"tool_calls"}]}`, string(planArgs), string(reviewArgs))
}

func scriptedArtifactLocalRead(endpoint, requestID string) string {
	path := newReservedPaths(requestID).PlanPath
	if endpoint == "anthropic" {
		return fmt.Sprintf(`{"id":"msg-scripted-local","type":"message","role":"assistant","content":[{"type":"text","text":"local-visible"},{"type":"tool_use","id":"provider-local-read","name":"read_file","input":{"path":%q}}],"stop_reason":"tool_use"}`, path)
	}
	arguments, _ := json.Marshal(map[string]string{"path": path})
	return fmt.Sprintf(`{"id":"chatcmpl-scripted-local","created":3,"choices":[{"message":{"role":"assistant","content":"local-visible","tool_calls":[{"id":"provider-local-read","type":"function","function":{"name":"read_file","arguments":%q}}]},"finish_reason":"tool_calls"}]}`, string(arguments))
}

func scriptedArtifactDirect(endpoint string) string {
	if endpoint == "anthropic" {
		return `{"id":"msg-scripted-direct","type":"message","role":"assistant","content":[{"type":"text","text":"must not escape pair frontier"}],"stop_reason":"end_turn"}`
	}
	return `{"id":"chatcmpl-scripted-direct","created":3,"choices":[{"message":{"role":"assistant","content":"must not escape pair frontier"},"finish_reason":"stop"}]}`
}

func TestHotPathPresetHandlersDirect(t *testing.T) {
	t.Run("DirectOnlyPresetUsesDirectTerminalForChatAndMessages", func(t *testing.T) {
		preset := hotPathSelectorPreset([]string{config.ModeDirect})
		preset.WorkspaceTools = nil

		chatCandidate := anthropicTestCandidate(t, "openai")
		chatBody := `{"id":"chatcmpl-direct-only","created":1777000001,"choices":[{"message":{"role":"assistant","content":"chat direct"},"finish_reason":"stop"}]}`
		chatServer, chatFake := newHotPathHandlerServerWithPreset(t, preset, chatCandidate, staticProviderTunnelFrames(chatBody))
		chatResponse := serveHotPathChat(t, chatServer, false)
		if chatResponse.Code != http.StatusOK || !strings.Contains(chatResponse.Body.String(), "chatcmpl-direct-only") {
			t.Fatalf("direct-only Chat response: status=%d body=%s", chatResponse.Code, chatResponse.Body.String())
		}
		if chatFake.poolLastRunSnapshot().ModelGroupKey != "selector-model" || chatFake.poolSubmitCountSnapshot() != 1 {
			t.Fatalf("direct-only Chat selector admission mismatch: %+v", chatFake.poolLastRunSnapshot())
		}
		assertHotPathTerminal(t, chatServer)

		messagesCandidate := anthropicTestCandidate(t, "anthropic")
		messagesBody := []byte(`{"id":"msg_direct_only","type":"message","role":"assistant","content":[{"type":"text","text":"messages direct"}],"stop_reason":"end_turn"}`)
		messagesServer, messagesFake := newHotPathHandlerServerWithPreset(t, preset, messagesCandidate, anthropicTunnelFrames(http.StatusOK, "application/json", messagesBody))
		messagesResponse := serveHotPathAnthropic(t, messagesServer, false)
		if messagesResponse.Code != http.StatusOK || !strings.Contains(messagesResponse.Body.String(), "msg_direct_only") {
			t.Fatalf("direct-only Messages response: status=%d body=%s", messagesResponse.Code, messagesResponse.Body.String())
		}
		if messagesFake.poolLastRunSnapshot().ModelGroupKey != "selector-model" || messagesFake.poolSubmitCountSnapshot() != 1 {
			t.Fatalf("direct-only Messages selector admission mismatch: %+v", messagesFake.poolLastRunSnapshot())
		}
		assertHotPathTerminal(t, messagesServer)
	})

	t.Run("ChatNonStreamReasoningMetadataAndTerminal", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "openai")
		providerBody := `{"id":"chatcmpl-provider-101","object":"chat.completion","created":1777000101,"model":"served-selector","choices":[{"index":0,"message":{"role":"assistant","content":"final text","reasoning_content":"actual reasoning"},"finish_reason":"stop"}],"usage":{"prompt_tokens":17,"completion_tokens":29,"total_tokens":46,"provider_extra":7}}`
		srv, fake := newHotPathHandlerServer(t, candidate, staticProviderTunnelFrames(providerBody))
		response := serveHotPathChat(t, srv, false)
		if response.Code != http.StatusOK {
			t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
		}
		var body map[string]any
		if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
			t.Fatal(err)
		}
		usage := body["usage"].(map[string]any)
		if body["id"] != "chatcmpl-provider-101" || body["created"] != float64(1_777_000_101) || body["model"] != "virtual-model" || usage["provider_extra"] != float64(7) {
			t.Fatalf("provider metadata was not preserved: %+v", body)
		}
		assertHotPathTerminal(t, srv)
		if fake.poolLastRunSnapshot().ModelGroupKey != "selector-model" || fake.poolSubmitCountSnapshot() != 1 {
			t.Fatalf("selector admission mismatch: %+v", fake.poolLastRunSnapshot())
		}
		assertNoReservedPath(t, response.Body.String())
	})

	t.Run("TunnelTransportMetadataDoesNotBecomePublic", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "openai")
		providerBody := `{"id":"chatcmpl-provider-public","created":1777000111,"choices":[{"message":{"role":"assistant","content":"final text"},"finish_reason":"stop"}]}`
		srv, _ := newHotPathHandlerServer(t, candidate, hotPathTunnelFrames(providerBody, "application/json", "run-internal-only", 1_555_000_000_000_000_000))
		response := serveHotPathChat(t, srv, false)
		if response.Code != http.StatusOK || strings.Contains(response.Body.String(), "run-internal-only") {
			t.Fatalf("transport metadata leaked: status=%d body=%s", response.Code, response.Body.String())
		}
		var body map[string]any
		if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
			t.Fatal(err)
		}
		if body["id"] != "chatcmpl-provider-public" || body["created"] != float64(1_777_000_111) {
			t.Fatalf("provider metadata was replaced: %+v", body)
		}
		assertHotPathTerminal(t, srv)
	})

	t.Run("MissingProviderMetadataReturnsEndpointErrors", func(t *testing.T) {
		const (
			missingRunID               = "run-should-not-leak"
			missingFrameTimestampNano  = int64(1_555_000_000_000_000_000)
			missingFrameTimestampSecs  = "1555000000"
			missingFrameTimestampNanos = "1555000000000000000"
		)
		tests := []struct {
			name      string
			candidate edgeservice.ProviderPoolCandidate
			frames    chan *iop.ProviderTunnelFrame
			serve     func(*testing.T, *Server, bool) *httptest.ResponseRecorder
			stream    bool
			errorTyp  string
		}{
			{
				name: "ChatJSONMissingID", candidate: anthropicTestCandidate(t, "openai"),
				frames: hotPathTunnelFrames(`{"created":1777000121,"choices":[{"message":{"role":"assistant","content":"bad"},"finish_reason":"stop"}]}`, "application/json", missingRunID, missingFrameTimestampNano),
				serve:  serveHotPathChat, errorTyp: "run_error",
			},
			{
				name: "ChatSSEMissingID", candidate: anthropicTestCandidate(t, "openai"),
				frames: hotPathTunnelFrames("data: {\"created\":1777000122,\"choices\":[{\"delta\":{\"content\":\"bad\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", "text/event-stream", missingRunID, missingFrameTimestampNano),
				serve:  serveHotPathChat, errorTyp: "run_error",
			},
			{
				name: "MessagesJSONMissingID", candidate: anthropicTestCandidate(t, "anthropic"),
				frames: hotPathTunnelFrames(`{"type":"message","role":"assistant","content":[{"type":"text","text":"bad"}],"stop_reason":"end_turn"}`, "application/json", missingRunID, missingFrameTimestampNano),
				serve:  serveHotPathAnthropic, errorTyp: "api_error",
			},
			{
				name: "MessagesSSEMissingID", candidate: anthropicTestCandidate(t, "anthropic"),
				frames: hotPathTunnelFrames("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"type\":\"message\",\"role\":\"assistant\",\"model\":\"served-selector\",\"content\":[]}}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", "text/event-stream", missingRunID, missingFrameTimestampNano),
				serve:  serveHotPathAnthropic, stream: true, errorTyp: "api_error",
			},
		}
		for _, test := range tests {
			t.Run(test.name, func(t *testing.T) {
				srv, _ := newHotPathHandlerServer(t, test.candidate, test.frames)
				response := test.serve(t, srv, test.stream)
				body := response.Body.String()
				if response.Code != http.StatusBadGateway || !strings.Contains(body, `"type":"`+test.errorTyp+`"`) || strings.Contains(body, missingRunID) || strings.Contains(body, missingFrameTimestampNanos) || strings.Contains(body, missingFrameTimestampSecs) {
					t.Fatalf("missing provider metadata response: status=%d body=%s", response.Code, body)
				}
				assertHotPathTerminal(t, srv)
			})
		}
	})

	t.Run("ChatNormalizedRunEventMetadataAndTerminal", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "openai")
		candidate.ExecutionPath = string(edgeservice.ProviderPoolPathNormalized)
		srv, fake := newHotPathHandlerServer(t, candidate, nil)
		dispatch := edgeservice.RunDispatch{
			RunID: "run-normalized-provider-151", NodeID: "node-normalized", ModelGroupKey: "selector-model",
			ProviderID: candidate.ProviderID, ExecutionPath: string(edgeservice.ProviderPoolPathNormalized),
			ProfileID: candidate.ProfileID, ProfileDriver: candidate.ProfileDriver,
			ProfileCapabilities: append([]string(nil), candidate.ProfileCapabilities...),
		}
		events := bufferedRunEvents(
			&iop.RunEvent{RunId: dispatch.RunID, Type: "reasoning_delta", Delta: "normalized reasoning", Timestamp: 1_777_000_151_000_000_000, Metadata: map[string]string{hotPathOpenAIResponseIDMetadata: "chatcmpl-normalized-provider-151"}},
			&iop.RunEvent{RunId: dispatch.RunID, Type: "delta", Delta: "normalized final", Timestamp: 1_777_000_151_000_000_000, Metadata: map[string]string{hotPathOpenAIResponseIDMetadata: "chatcmpl-normalized-provider-151"}},
			&iop.RunEvent{RunId: dispatch.RunID, Type: "complete", Timestamp: 1_777_000_151_000_000_000, Metadata: map[string]string{"finish_reason": "stop", hotPathOpenAIResponseIDMetadata: "chatcmpl-normalized-provider-151"}, Usage: &iop.Usage{InputTokens: 43, OutputTokens: 17}},
		)
		fake.poolSubmitResults = []edgeservice.ProviderPoolDispatchResult{{
			Path: edgeservice.ProviderPoolPathNormalized, DispatchInfo: dispatch,
			Run: &fakeRunResult{dispatch: dispatch, events: events},
		}}
		response := serveHotPathChat(t, srv, false)
		if response.Code != http.StatusOK {
			t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
		}
		var body map[string]any
		if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
			t.Fatal(err)
		}
		usage := body["usage"].(map[string]any)
		if body["id"] != "chatcmpl-normalized-provider-151" || body["created"] != float64(1_777_000_151) || body["model"] != "virtual-model" || usage["prompt_tokens"] != float64(43) {
			t.Fatalf("normalized metadata mismatch: %+v", body)
		}
		if strings.Contains(response.Body.String(), dispatch.RunID) {
			t.Fatalf("normalized run identity leaked: %s", response.Body.String())
		}
		assertHotPathTerminal(t, srv)
	})

	t.Run("ChatStreamToolFrontierAndUsage", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "openai")
		stream := strings.Join([]string{
			`data: {"id":"chatcmpl-provider-202","object":"chat.completion.chunk","created":1777000202,"model":"served-selector","choices":[{"index":0,"delta":{"reasoning_content":"inspect"},"finish_reason":null}]}`,
			`data: {"id":"chatcmpl-provider-202","object":"chat.completion.chunk","created":1777000202,"model":"served-selector","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_provider_202","function":{"name":"read_file","arguments":"{\"path\":\"README.md\"}"}}]},"finish_reason":null}]}`,
			`data: {"id":"chatcmpl-provider-202","object":"chat.completion.chunk","created":1777000202,"model":"served-selector","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":23,"completion_tokens":11,"total_tokens":34}}`,
			`data: [DONE]`, "",
		}, "\n\n")
		srv, _ := newHotPathHandlerServer(t, candidate, staticProviderTunnelFrames(stream))
		response := serveHotPathChat(t, srv, true)
		if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "chatcmpl-provider-202") || !strings.Contains(response.Body.String(), `"prompt_tokens":23`) {
			t.Fatalf("stream metadata mismatch: status=%d body=%s", response.Code, response.Body.String())
		}
		assertHotPathWaiting(t, srv, "chatcmpl-provider-202-tool-1", "call_provider_202")
		assertNoReservedPath(t, response.Body.String())
	})

	t.Run("AnthropicNativeNonStreamMetadataAndTerminal", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "anthropic")
		providerBody := []byte(`{"id":"msg_provider_303","type":"message","role":"assistant","model":"served-selector","content":[{"type":"thinking","thinking":"native thought","signature":"sig"},{"type":"text","text":"native final"}],"stop_reason":"end_turn","stop_sequence":null,"usage":{"input_tokens":31,"output_tokens":19,"cache_read_input_tokens":5}}`)
		srv, _ := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, "application/json", providerBody))
		response := serveHotPathAnthropic(t, srv, false)
		if response.Code != http.StatusOK {
			t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
		}
		var body map[string]any
		if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
			t.Fatal(err)
		}
		usage := body["usage"].(map[string]any)
		if body["id"] != "msg_provider_303" || body["model"] != "virtual-model" || usage["input_tokens"] != float64(31) || usage["cache_read_input_tokens"] != float64(5) {
			t.Fatalf("native metadata mismatch: %+v", body)
		}
		content := body["content"].([]any)
		if content[0].(map[string]any)["signature"] != "sig" {
			t.Fatalf("thinking signature was not preserved: %+v", content)
		}
		assertHotPathTerminal(t, srv)
		assertNoReservedPath(t, response.Body.String())
	})

	t.Run("AnthropicNativeStreamToolFrontier", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "anthropic")
		stream := strings.Join([]string{
			`event: message_start\ndata: {"type":"message_start","message":{"id":"msg_provider_404","type":"message","role":"assistant","model":"served-selector","content":[],"stop_reason":null,"usage":{"input_tokens":41,"output_tokens":0}}}`,
			`event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_provider_404","name":"read_file","input":{}}}`,
			`event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"path\":\"README.md\"}"}}`,
			`event: content_block_stop\ndata: {"type":"content_block_stop","index":0}`,
			`event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":7}}`,
			`event: message_stop\ndata: {"type":"message_stop"}`, "",
		}, "\n\n")
		stream = strings.ReplaceAll(stream, `\n`, "\n")
		srv, _ := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, "text/event-stream", []byte(stream)))
		response := serveHotPathAnthropic(t, srv, true)
		if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "msg_provider_404") || !strings.Contains(response.Body.String(), `"output_tokens":7`) {
			t.Fatalf("native stream mismatch: status=%d body=%s", response.Code, response.Body.String())
		}
		assertHotPathWaiting(t, srv, "msg_provider_404-tool-1", "toolu_provider_404")
		assertNoReservedPath(t, response.Body.String())
	})

	t.Run("AnthropicChatBridgePreservesProviderIdentity", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "openai")
		providerBody := []byte(`{"id":"chatcmpl_bridge_505","model":"served-selector","choices":[{"message":{"role":"assistant","content":"bridge final"},"finish_reason":"stop"}],"usage":{"prompt_tokens":37,"completion_tokens":13,"prompt_tokens_details":{"cached_tokens":9}}}`)
		srv, _ := newHotPathHandlerServer(t, candidate, anthropicTunnelFrames(http.StatusOK, "application/json", providerBody))
		response := serveHotPathAnthropic(t, srv, false)
		if response.Code != http.StatusOK {
			t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
		}
		var body map[string]any
		if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
			t.Fatal(err)
		}
		usage := body["usage"].(map[string]any)
		if body["id"] != "chatcmpl_bridge_505" || body["model"] != "virtual-model" || usage["input_tokens"] != float64(37) || usage["cache_read_input_tokens"] != float64(9) {
			t.Fatalf("bridge metadata mismatch: %+v", body)
		}
		assertHotPathTerminal(t, srv)
	})

	t.Run("MalformedReservedControlRejectedBeforeDirect", func(t *testing.T) {
		candidate := anthropicTestCandidate(t, "openai")
		providerBody := `{"id":"chatcmpl-provider-bad","created":1777000606,"choices":[{"message":{"role":"assistant","content":"","tool_calls":[{"id":"call_bad_control","type":"function","function":{"name":"shell","arguments":"{\"path\":\".iop/job/not-issued/plan.md\"}"}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`
		srv, _ := newHotPathHandlerServer(t, candidate, staticProviderTunnelFrames(providerBody))
		response := serveHotPathChat(t, srv, false)
		if response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), reasonMalformedControlRole) {
			t.Fatalf("malformed selector response was not rejected: status=%d body=%s", response.Code, response.Body.String())
		}
		assertHotPathTerminal(t, srv)
	})
}

func newHotPathHandlerServer(t *testing.T, candidate edgeservice.ProviderPoolCandidate, frames chan *iop.ProviderTunnelFrame) (*Server, *providerFakeRunService) {
	return newHotPathHandlerServerWithPreset(t, hotPathSelectorPreset([]string{config.ModeDirect}), candidate, frames)
}

func newHotPathHandlerServerWithPreset(t *testing.T, preset config.ExecutionPreset, candidate edgeservice.ProviderPoolCandidate, frames chan *iop.ProviderTunnelFrame) (*Server, *providerFakeRunService) {
	t.Helper()
	fake := &providerFakeRunService{
		poolDispatchPath: string(edgeservice.ProviderPoolPathTunnel), poolSelectedCandidate: candidate,
		tunnelServedTarget: "served-selector", tunnelFrames: frames,
	}
	srv := NewServer(config.EdgeOpenAIConf{}, fake, nil)
	srv.SetEdgeID("edge-hot-path-test")
	srv.SetExecutionPresets([]config.ExecutionPreset{preset})
	srv.SetModelCatalog([]config.ModelCatalogEntry{
		{ID: "virtual-model", ExecutionPreset: preset.ID},
		{ID: "selector-model", Providers: map[string]string{candidate.ProviderID: "served-selector"}},
	})
	return srv, fake
}

func hotPathTunnelFrames(body, contentType, runID string, timestamp int64) chan *iop.ProviderTunnelFrame {
	frames := make(chan *iop.ProviderTunnelFrame, 3)
	frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: http.StatusOK, Headers: map[string]string{"Content-Type": contentType}, RunId: runID, Timestamp: timestamp}
	frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte(body), RunId: runID, Timestamp: timestamp}
	frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true, RunId: runID, Timestamp: timestamp}
	close(frames)
	return frames
}

func serveHotPathChat(t *testing.T, srv *Server, stream bool) *httptest.ResponseRecorder {
	t.Helper()
	body := `{"model":"virtual-model","messages":[{"role":"user","content":"hello"}],"tools":[{"type":"function","function":{"name":"read_file","parameters":{"type":"object"}}}],"stream":` + fmt.Sprintf("%t", stream) + `}`
	request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(body))
	recorder := httptest.NewRecorder()
	srv.routes().ServeHTTP(recorder, request)
	return recorder
}

func serveHotPathAnthropic(t *testing.T, srv *Server, stream bool) *httptest.ResponseRecorder {
	t.Helper()
	body := `{"model":"virtual-model","max_tokens":64,"messages":[{"role":"user","content":"hello"}],"tools":[{"name":"read_file","description":"read","input_schema":{"type":"object"}}],"stream":` + fmt.Sprintf("%t", stream) + `}`
	request := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body))
	request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
	recorder := httptest.NewRecorder()
	srv.routes().ServeHTTP(recorder, request)
	return recorder
}

func soleHotPathSnapshot(t *testing.T, srv *Server) (string, logicalRequestSnapshot) {
	t.Helper()
	coordinator := srv.requestCoordinator
	coordinator.mu.Lock()
	if len(coordinator.requests) != 1 {
		count := len(coordinator.requests)
		coordinator.mu.Unlock()
		t.Fatalf("logical request count=%d, want 1", count)
	}
	var requestID string
	for id := range coordinator.requests {
		requestID = id
	}
	coordinator.mu.Unlock()
	snapshot, err := coordinator.snapshot(requestID)
	if err != nil {
		t.Fatal(err)
	}
	return requestID, snapshot
}

func assertHotPathTerminal(t *testing.T, srv *Server) {
	t.Helper()
	srv.requestCoordinator.mu.Lock()
	remaining := len(srv.requestCoordinator.requests)
	srv.requestCoordinator.mu.Unlock()
	if remaining != 0 {
		t.Fatalf("logical terminal retained %d coordinator records", remaining)
	}
}

func assertHotPathWaiting(t *testing.T, srv *Server, callID string, providerID ...string) {
	t.Helper()
	requestID, snapshot := soleHotPathSnapshot(t, srv)
	if snapshot.State != logicalRequestStateWaiting || len(snapshot.ExpectedCallIDs) != 1 || snapshot.ExpectedCallIDs[0] != callID {
		t.Fatalf("logical frontier mismatch: %+v", snapshot)
	}
	if len(providerID) > 0 {
		srv.requestCoordinator.mu.Lock()
		record := srv.requestCoordinator.requests[requestID]
		got := ""
		if record != nil {
			got = record.publicToProvider[callID]
		}
		srv.requestCoordinator.mu.Unlock()
		if got != providerID[0] {
			t.Fatalf("logical provider mapping for %q = %q, want %q", callID, got, providerID[0])
		}
	}
}

func assertNoReservedPath(t *testing.T, body string) {
	t.Helper()
	if strings.Contains(body, ".iop/job/") {
		t.Fatalf("direct response contains reserved path: %s", body)
	}
}