1020 lines
51 KiB
Go
1020 lines
51 KiB
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,
|
|
"Keep every explicit requirement, exact literal, filename, command, output string, and acceptance condition.",
|
|
"Do not write markdown; IOP renders the fixed Plan and pending Review templates.",
|
|
"IOP will append the final Review handoff step and matching pending status itself.",
|
|
}
|
|
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)
|
|
}
|
|
}
|