667 lines
25 KiB
Go
667 lines
25 KiB
Go
package openai
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"iop/packages/go/config"
|
|
)
|
|
|
|
// TestPresetRequestIdentityAcrossChatTurns tests full multi-turn Chat completions
|
|
// ingress through the coordinator: begin turn, stage activation, tool result continuation,
|
|
// and rejection cases.
|
|
func TestPresetRequestIdentityAcrossChatTurns(t *testing.T) {
|
|
fake := &providerFakeRunService{
|
|
poolDispatchPath: string(edgeservice.ProviderPoolPathNormalized),
|
|
}
|
|
|
|
preset := config.ExecutionPreset{
|
|
ID: "preset-chat-test",
|
|
AllowedModes: []string{"direct"},
|
|
}
|
|
|
|
rawToken1 := "token-user-1"
|
|
sum1 := sha256.Sum256([]byte(rawToken1))
|
|
rawToken2 := "token-user-2"
|
|
sum2 := sha256.Sum256([]byte(rawToken2))
|
|
|
|
cfg := config.EdgeOpenAIConf{
|
|
PrincipalTokens: []config.OpenAIPrincipalTokenConf{
|
|
{TokenRef: "tok-1", TokenHashSHA256: hex.EncodeToString(sum1[:]), PrincipalRef: "user-1"},
|
|
{TokenRef: "tok-2", TokenHashSHA256: hex.EncodeToString(sum2[:]), PrincipalRef: "user-2"},
|
|
},
|
|
}
|
|
|
|
srv := NewServer(cfg, fake, nil)
|
|
srv.SetEdgeID("edge-identity-test")
|
|
srv.SetExecutionPresets([]config.ExecutionPreset{preset})
|
|
srv.SetModelCatalog([]config.ModelCatalogEntry{
|
|
{
|
|
ID: "virtual-preset-chat",
|
|
ExecutionPreset: "preset-chat-test",
|
|
},
|
|
})
|
|
|
|
// 1. Turn 1 (Begin): User 1 sends initial prompt
|
|
bodyTurn1 := `{
|
|
"model": "virtual-preset-chat",
|
|
"messages": [{"role": "user", "content": "hello"}]
|
|
}`
|
|
req1 := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyTurn1))
|
|
req1.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
w1 := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(w1, req1)
|
|
if w1.Code != http.StatusOK {
|
|
t.Fatalf("Turn 1 status: got %d, body: %s", w1.Code, w1.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != 1 {
|
|
t.Fatalf("Turn 1 pool submit count: got %d, want 1", got)
|
|
}
|
|
|
|
// Retrieve logical request state from coordinator
|
|
coord := srv.logicalRequests()
|
|
coord.mu.Lock()
|
|
if len(coord.requests) != 1 {
|
|
coord.mu.Unlock()
|
|
t.Fatalf("coordinator requests count = %d, want 1", len(coord.requests))
|
|
}
|
|
var reqID string
|
|
var rec *logicalRequestRecord
|
|
for id, r := range coord.requests {
|
|
reqID = id
|
|
rec = r
|
|
break
|
|
}
|
|
stageID := rec.activeStageID
|
|
coord.mu.Unlock()
|
|
|
|
if rec.principalRef != "user-1" {
|
|
t.Fatalf("principalRef = %q, want user-1", rec.principalRef)
|
|
}
|
|
if rec.ownerEdgeID != "edge-identity-test" {
|
|
t.Fatalf("ownerEdgeID = %q, want edge-identity-test", rec.ownerEdgeID)
|
|
}
|
|
|
|
// Trusted per-turn identity must be attached to the dispatched run metadata,
|
|
// server-issued and never chosen by the caller.
|
|
meta1 := fake.poolLastRunSnapshot().Metadata
|
|
turn1ReqID := meta1["iop_logical_request_id"]
|
|
turn1CallID := meta1["iop_call_id"]
|
|
turn1StageID := meta1["iop_stage_id"]
|
|
if turn1ReqID != reqID {
|
|
t.Fatalf("Turn 1 dispatch logical request id = %q, want coordinator id %q", turn1ReqID, reqID)
|
|
}
|
|
if turn1StageID != stageID {
|
|
t.Fatalf("Turn 1 dispatch stage id = %q, want %q", turn1StageID, stageID)
|
|
}
|
|
if turn1CallID == "" {
|
|
t.Fatalf("Turn 1 dispatch call id is empty: %+v", meta1)
|
|
}
|
|
|
|
// Simulate stage 1 assistant issuing tool call "call_c1"
|
|
assistantMsg := json.RawMessage(`{"role":"assistant","tool_calls":[{"id":"call_c1","type":"function","function":{"name":"search"}}]}`)
|
|
issuedHash, err := fingerprintCanonicalJSON(logicalRequestEndpointChat, assistantMsg)
|
|
if err != nil {
|
|
t.Fatalf("fingerprintCanonicalJSON: %v", err)
|
|
}
|
|
|
|
if _, err := coord.awaitToolResults(reqID, "edge-identity-test", stageID, []logicalRequestExpectedTool{
|
|
{PublicCallID: "call_c1", ProviderCallID: "prov_c1"},
|
|
}, issuedHash); err != nil {
|
|
t.Fatalf("awaitToolResults: %v", err)
|
|
}
|
|
|
|
// 2. Turn 2 Continuation (Valid Resume by User 1)
|
|
bodyTurn2 := `{
|
|
"model": "virtual-preset-chat",
|
|
"messages": [
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "tool_calls": [{"id": "call_c1", "type": "function", "function": {"name": "search"}}]},
|
|
{"role": "tool", "tool_call_id": "call_c1", "content": "search result"}
|
|
]
|
|
}`
|
|
req2 := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyTurn2))
|
|
req2.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
w2 := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(w2, req2)
|
|
if w2.Code != http.StatusOK {
|
|
t.Fatalf("Turn 2 status: got %d, body: %s", w2.Code, w2.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != 2 {
|
|
t.Fatalf("Turn 2 pool submit count: got %d, want 2", got)
|
|
}
|
|
|
|
// The logical request id is stable across continuation, while each inbound
|
|
// HTTP turn receives a distinct, non-empty call id and a fresh stage id.
|
|
meta2 := fake.poolLastRunSnapshot().Metadata
|
|
if got := meta2["iop_logical_request_id"]; got != reqID {
|
|
t.Fatalf("Turn 2 dispatch logical request id = %q, want stable %q", got, reqID)
|
|
}
|
|
if got := meta2["iop_stage_id"]; got == "" || got == turn1StageID {
|
|
t.Fatalf("Turn 2 stage id not fresh: turn1=%q turn2=%q", turn1StageID, got)
|
|
}
|
|
if got := meta2["iop_call_id"]; got == "" || got == turn1CallID {
|
|
t.Fatalf("Turn 2 call id not distinct: turn1=%q turn2=%q", turn1CallID, got)
|
|
}
|
|
|
|
// Verify state after Turn 2 resume
|
|
snap2, err := coord.snapshot(reqID)
|
|
if err != nil {
|
|
t.Fatalf("snapshot reqID: %v", err)
|
|
}
|
|
if snap2.State != logicalRequestStateActive || snap2.ActiveStageID == "" {
|
|
t.Fatalf("Turn 2 snapshot state: %+v", snap2)
|
|
}
|
|
}
|
|
|
|
// TestPresetRequestIdentityAcrossAnthropicTurns tests full multi-turn Anthropic Messages
|
|
// ingress through the coordinator: begin turn, stage activation, tool result continuation,
|
|
// and rejection cases.
|
|
func TestPresetRequestIdentityAcrossAnthropicTurns(t *testing.T) {
|
|
candidate := anthropicTestCandidate(t, "anthropic")
|
|
fake := &providerFakeRunService{
|
|
poolDispatchPath: string(edgeservice.ProviderPoolPathTunnel),
|
|
poolSelectedCandidate: candidate,
|
|
tunnelServedTarget: "upstream-claude",
|
|
}
|
|
|
|
preset := config.ExecutionPreset{
|
|
ID: "preset-anthropic-test",
|
|
AllowedModes: []string{"direct"},
|
|
}
|
|
|
|
rawToken1 := "token-user-1"
|
|
sum1 := sha256.Sum256([]byte(rawToken1))
|
|
|
|
cfg := config.EdgeOpenAIConf{
|
|
PrincipalTokens: []config.OpenAIPrincipalTokenConf{
|
|
{TokenRef: "tok-1", TokenHashSHA256: hex.EncodeToString(sum1[:]), PrincipalRef: "user-1"},
|
|
},
|
|
}
|
|
|
|
srv := NewServer(cfg, fake, nil)
|
|
srv.SetEdgeID("edge-identity-test")
|
|
srv.SetExecutionPresets([]config.ExecutionPreset{preset})
|
|
srv.SetModelCatalog([]config.ModelCatalogEntry{
|
|
{
|
|
ID: "virtual-preset-anthropic",
|
|
ExecutionPreset: "preset-anthropic-test",
|
|
},
|
|
})
|
|
|
|
fixture := mustReadAnthropicFixture(t, "native_message.json")
|
|
fake.tunnelFrames = anthropicTunnelFrames(http.StatusOK, "application/json", fixture)
|
|
|
|
// 1. Turn 1 Begin
|
|
bodyTurn1 := `{
|
|
"model": "virtual-preset-anthropic",
|
|
"max_tokens": 64,
|
|
"messages": [{"role": "user", "content": "hello"}]
|
|
}`
|
|
req1 := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(bodyTurn1))
|
|
req1.Header.Set("X-Api-Key", rawToken1)
|
|
req1.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
|
|
w1 := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(w1, req1)
|
|
if w1.Code != http.StatusOK {
|
|
t.Fatalf("Anthropic Turn 1 status: got %d, body: %s", w1.Code, w1.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != 1 {
|
|
t.Fatalf("Anthropic Turn 1 submit count: got %d, want 1", got)
|
|
}
|
|
|
|
coord := srv.logicalRequests()
|
|
coord.mu.Lock()
|
|
if len(coord.requests) != 1 {
|
|
coord.mu.Unlock()
|
|
t.Fatalf("coordinator requests count = %d, want 1", len(coord.requests))
|
|
}
|
|
var reqID string
|
|
var rec *logicalRequestRecord
|
|
for id, r := range coord.requests {
|
|
reqID = id
|
|
rec = r
|
|
break
|
|
}
|
|
stageID := rec.activeStageID
|
|
coord.mu.Unlock()
|
|
|
|
// Trusted per-turn identity must be attached to the dispatched run metadata.
|
|
meta1 := fake.poolLastRunSnapshot().Metadata
|
|
turn1ReqID := meta1["iop_logical_request_id"]
|
|
turn1CallID := meta1["iop_call_id"]
|
|
turn1StageID := meta1["iop_stage_id"]
|
|
if turn1ReqID != reqID {
|
|
t.Fatalf("Anthropic Turn 1 dispatch logical request id = %q, want %q", turn1ReqID, reqID)
|
|
}
|
|
if turn1StageID != stageID {
|
|
t.Fatalf("Anthropic Turn 1 dispatch stage id = %q, want %q", turn1StageID, stageID)
|
|
}
|
|
if turn1CallID == "" {
|
|
t.Fatalf("Anthropic Turn 1 dispatch call id is empty: %+v", meta1)
|
|
}
|
|
|
|
// Simulate assistant issuing tool_use block tu_a1
|
|
assistantMsg := json.RawMessage(`{"role":"assistant","content":[{"type":"tool_use","id":"tu_a1","name":"search","input":{}}]}`)
|
|
issuedHash, err := fingerprintCanonicalJSON(logicalRequestEndpointAnthropic, assistantMsg)
|
|
if err != nil {
|
|
t.Fatalf("fingerprintCanonicalJSON: %v", err)
|
|
}
|
|
|
|
if _, err := coord.awaitToolResults(reqID, "edge-identity-test", stageID, []logicalRequestExpectedTool{
|
|
{PublicCallID: "tu_a1", ProviderCallID: "prov_tu_a1"},
|
|
}, issuedHash); err != nil {
|
|
t.Fatalf("awaitToolResults: %v", err)
|
|
}
|
|
|
|
// 2. Turn 2 Continuation (Valid Resume)
|
|
fake.tunnelFrames = anthropicTunnelFrames(http.StatusOK, "application/json", fixture)
|
|
bodyTurn2 := `{
|
|
"model": "virtual-preset-anthropic",
|
|
"max_tokens": 64,
|
|
"messages": [
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": [{"type": "tool_use", "id": "tu_a1", "name": "search", "input": {}}]},
|
|
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "tu_a1", "content": "ok"}]}
|
|
]
|
|
}`
|
|
req2 := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(bodyTurn2))
|
|
req2.Header.Set("X-Api-Key", rawToken1)
|
|
req2.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
|
|
w2 := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(w2, req2)
|
|
if w2.Code != http.StatusOK {
|
|
t.Fatalf("Anthropic Turn 2 status: got %d, body: %s", w2.Code, w2.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != 2 {
|
|
t.Fatalf("Anthropic Turn 2 submit count: got %d, want 2", got)
|
|
}
|
|
|
|
// Stable logical request id across the continuation; distinct call id and a
|
|
// fresh stage id per HTTP turn.
|
|
meta2 := fake.poolLastRunSnapshot().Metadata
|
|
if got := meta2["iop_logical_request_id"]; got != reqID {
|
|
t.Fatalf("Anthropic Turn 2 logical request id = %q, want stable %q", got, reqID)
|
|
}
|
|
if got := meta2["iop_stage_id"]; got == "" || got == turn1StageID {
|
|
t.Fatalf("Anthropic Turn 2 stage id not fresh: turn1=%q turn2=%q", turn1StageID, got)
|
|
}
|
|
if got := meta2["iop_call_id"]; got == "" || got == turn1CallID {
|
|
t.Fatalf("Anthropic Turn 2 call id not distinct: turn1=%q turn2=%q", turn1CallID, got)
|
|
}
|
|
}
|
|
|
|
// TestPresetRequestIdentityRejectionCases verifies that cross-principal, missing-store,
|
|
// and history-mutation rejections write endpoint-standard errors and dispatch zero
|
|
// providers, while caller identity metadata is neutralized by trusted overwrite.
|
|
func TestPresetRequestIdentityRejectionCases(t *testing.T) {
|
|
fake := &providerFakeRunService{
|
|
poolDispatchPath: string(edgeservice.ProviderPoolPathNormalized),
|
|
}
|
|
|
|
preset := config.ExecutionPreset{
|
|
ID: "preset-rejection-test",
|
|
AllowedModes: []string{"direct"},
|
|
}
|
|
|
|
rawToken1 := "token-user-1"
|
|
sum1 := sha256.Sum256([]byte(rawToken1))
|
|
rawToken2 := "token-user-2"
|
|
sum2 := sha256.Sum256([]byte(rawToken2))
|
|
|
|
cfg := config.EdgeOpenAIConf{
|
|
PrincipalTokens: []config.OpenAIPrincipalTokenConf{
|
|
{TokenRef: "tok-1", TokenHashSHA256: hex.EncodeToString(sum1[:]), PrincipalRef: "user-1"},
|
|
{TokenRef: "tok-2", TokenHashSHA256: hex.EncodeToString(sum2[:]), PrincipalRef: "user-2"},
|
|
},
|
|
}
|
|
|
|
srv := NewServer(cfg, fake, nil)
|
|
srv.SetEdgeID("edge-identity-test")
|
|
srv.SetExecutionPresets([]config.ExecutionPreset{preset})
|
|
srv.SetModelCatalog([]config.ModelCatalogEntry{
|
|
{
|
|
ID: "virtual-preset-rej",
|
|
ExecutionPreset: "preset-rejection-test",
|
|
},
|
|
{
|
|
ID: "legacy-route",
|
|
Providers: map[string]string{"dummy": "model-legacy"},
|
|
},
|
|
})
|
|
|
|
// 1. Begin request by User 1
|
|
bodyTurn1 := `{
|
|
"model": "virtual-preset-rej",
|
|
"messages": [{"role": "user", "content": "initial"}]
|
|
}`
|
|
req1 := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyTurn1))
|
|
req1.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
w1 := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(w1, req1)
|
|
if w1.Code != http.StatusOK {
|
|
t.Fatalf("Turn 1 status: got %d", w1.Code)
|
|
}
|
|
|
|
coord := srv.logicalRequests()
|
|
coord.mu.Lock()
|
|
var reqID string
|
|
var rec *logicalRequestRecord
|
|
for id, r := range coord.requests {
|
|
reqID = id
|
|
rec = r
|
|
break
|
|
}
|
|
stageID := rec.activeStageID
|
|
coord.mu.Unlock()
|
|
|
|
assistantMsg := json.RawMessage(`{"role":"assistant","tool_calls":[{"id":"call_r1","type":"function","function":{"name":"search"}}]}`)
|
|
issuedHash, err := fingerprintCanonicalJSON(logicalRequestEndpointChat, assistantMsg)
|
|
if err != nil {
|
|
t.Fatalf("fingerprintCanonicalJSON: %v", err)
|
|
}
|
|
if _, err := coord.awaitToolResults(reqID, "edge-identity-test", stageID, []logicalRequestExpectedTool{
|
|
{PublicCallID: "call_r1", ProviderCallID: "prov_r1"},
|
|
}, issuedHash); err != nil {
|
|
t.Fatalf("awaitToolResults: %v", err)
|
|
}
|
|
|
|
initialSubmits := fake.poolSubmitCountSnapshot()
|
|
|
|
// Rejection Case 1: Cross-Principal Resume (User 2 attempts to send tool results for call_r1)
|
|
bodyCrossPrincipal := `{
|
|
"model": "virtual-preset-rej",
|
|
"messages": [
|
|
{"role": "user", "content": "initial"},
|
|
{"role": "assistant", "tool_calls": [{"id": "call_r1", "type": "function", "function": {"name": "search"}}]},
|
|
{"role": "tool", "tool_call_id": "call_r1", "content": "result"}
|
|
]
|
|
}`
|
|
reqCross := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyCrossPrincipal))
|
|
reqCross.Header.Set("Authorization", "Bearer "+rawToken2)
|
|
wCross := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wCross, reqCross)
|
|
|
|
if wCross.Code != http.StatusBadRequest {
|
|
t.Fatalf("Cross-principal status: got %d, want 400. body: %s", wCross.Code, wCross.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != initialSubmits {
|
|
t.Fatalf("Provider dispatched on cross-principal rejection: got %d, want %d", got, initialSubmits)
|
|
}
|
|
|
|
// Rejection Case 2: Missing / Unknown Store State (tool_call_id "call_unknown")
|
|
bodyMissingState := `{
|
|
"model": "virtual-preset-rej",
|
|
"messages": [
|
|
{"role": "user", "content": "initial"},
|
|
{"role": "assistant", "tool_calls": [{"id": "call_unknown", "type": "function", "function": {"name": "search"}}]},
|
|
{"role": "tool", "tool_call_id": "call_unknown", "content": "result"}
|
|
]
|
|
}`
|
|
reqMissing := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyMissingState))
|
|
reqMissing.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wMissing := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wMissing, reqMissing)
|
|
|
|
if wMissing.Code != http.StatusBadRequest {
|
|
t.Fatalf("Missing state status: got %d, want 400. body: %s", wMissing.Code, wMissing.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != initialSubmits {
|
|
t.Fatalf("Provider dispatched on missing-state rejection: got %d, want %d", got, initialSubmits)
|
|
}
|
|
|
|
// Rejection Case 3: History Mutation (User 1 alters previous user message "initial" -> "mutated")
|
|
bodyMutatedHistory := `{
|
|
"model": "virtual-preset-rej",
|
|
"messages": [
|
|
{"role": "user", "content": "mutated"},
|
|
{"role": "assistant", "tool_calls": [{"id": "call_r1", "type": "function", "function": {"name": "search"}}]},
|
|
{"role": "tool", "tool_call_id": "call_r1", "content": "result"}
|
|
]
|
|
}`
|
|
reqMutated := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyMutatedHistory))
|
|
reqMutated.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wMutated := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wMutated, reqMutated)
|
|
|
|
if wMutated.Code != http.StatusBadRequest {
|
|
t.Fatalf("Mutated history status: got %d, want 400. body: %s", wMutated.Code, wMutated.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != initialSubmits {
|
|
t.Fatalf("Provider dispatched on mutated history rejection: got %d, want %d", got, initialSubmits)
|
|
}
|
|
|
|
// Case 4: Caller-metadata Spoof Attempt
|
|
// Caller passes spoofed metadata attempt: "iop_principal_ref": "user-2"
|
|
bodySpoof := `{
|
|
"model": "virtual-preset-rej",
|
|
"metadata": {"iop_principal_ref": "user-2", "iop_logical_request_id": "spoof-req"},
|
|
"messages": [{"role": "user", "content": "spoof attempt"}]
|
|
}`
|
|
reqSpoof := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodySpoof))
|
|
reqSpoof.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wSpoof := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wSpoof, reqSpoof)
|
|
|
|
if wSpoof.Code != http.StatusOK {
|
|
t.Fatalf("Spoof request status: got %d, body: %s", wSpoof.Code, wSpoof.Body.String())
|
|
}
|
|
|
|
// Verify that the new logical request was created under user-1 (authenticated bearer), not spoofed user-2
|
|
coord.mu.Lock()
|
|
for _, record := range coord.requests {
|
|
if record.principalRef == "user-2" {
|
|
coord.mu.Unlock()
|
|
t.Fatalf("Spoofed principal user-2 was recorded in coordinator!")
|
|
}
|
|
}
|
|
coord.mu.Unlock()
|
|
|
|
// Case 5: Legacy Bypass
|
|
// Non-preset route request should bypass coordinator completely
|
|
bodyLegacy := `{
|
|
"model": "legacy-route",
|
|
"messages": [{"role": "user", "content": "legacy"}]
|
|
}`
|
|
reqLegacy := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyLegacy))
|
|
reqLegacy.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wLegacy := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wLegacy, reqLegacy)
|
|
|
|
if wLegacy.Code != http.StatusOK {
|
|
t.Fatalf("Legacy route status: got %d, body: %s", wLegacy.Code, wLegacy.Body.String())
|
|
}
|
|
|
|
// Case 6: Cross-Owner Resume
|
|
// A waiting frontier owned by a DIFFERENT Edge must never resume here and
|
|
// must dispatch nothing.
|
|
t.Run("cross-owner waiting record", func(t *testing.T) {
|
|
crossLineage, err := newChatRequestLineage([]byte(`{"model":"virtual-preset-rej","messages":[{"role":"user","content":"cross-owner"}]}`))
|
|
if err != nil {
|
|
t.Fatalf("newChatRequestLineage: %v", err)
|
|
}
|
|
crossSnap, err := coord.create(logicalRequestAdmission{
|
|
OwnerEdgeID: "other-edge", PrincipalRef: "user-1", Lineage: crossLineage, PresetGeneration: "gen-1",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("seed create: %v", err)
|
|
}
|
|
crossStage, err := coord.newStageID()
|
|
if err != nil {
|
|
t.Fatalf("newStageID: %v", err)
|
|
}
|
|
if _, err := coord.activateStage(crossSnap.ID, "other-edge", crossStage); err != nil {
|
|
t.Fatalf("activateStage: %v", err)
|
|
}
|
|
if _, err := coord.awaitToolResults(crossSnap.ID, "other-edge", crossStage, []logicalRequestExpectedTool{
|
|
{PublicCallID: "call_cross", ProviderCallID: "prov_cross"},
|
|
}, "seed-issued-hash"); err != nil {
|
|
t.Fatalf("awaitToolResults: %v", err)
|
|
}
|
|
|
|
submitsBefore := fake.poolSubmitCountSnapshot()
|
|
bodyCrossOwner := `{
|
|
"model": "virtual-preset-rej",
|
|
"messages": [
|
|
{"role": "user", "content": "cross-owner"},
|
|
{"role": "assistant", "tool_calls": [{"id": "call_cross", "type": "function", "function": {"name": "search"}}]},
|
|
{"role": "tool", "tool_call_id": "call_cross", "content": "result"}
|
|
]
|
|
}`
|
|
reqCO := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(bodyCrossOwner))
|
|
reqCO.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wCO := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wCO, reqCO)
|
|
|
|
if wCO.Code != http.StatusBadRequest {
|
|
t.Fatalf("cross-owner status: got %d, want 400. body: %s", wCO.Code, wCO.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != submitsBefore {
|
|
t.Fatalf("provider dispatched on cross-owner rejection: got %d, want %d", got, submitsBefore)
|
|
}
|
|
})
|
|
|
|
// Case 7: Tool-Schema Mutation
|
|
// Resuming with a changed tools schema must be rejected before any provider
|
|
// dispatch.
|
|
t.Run("tool-schema mutation", func(t *testing.T) {
|
|
beginBody := `{
|
|
"model": "virtual-preset-rej",
|
|
"tools": [{"type": "function", "function": {"name": "search", "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}}}],
|
|
"messages": [{"role": "user", "content": "schema initial"}]
|
|
}`
|
|
reqBegin := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(beginBody))
|
|
reqBegin.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wBegin := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wBegin, reqBegin)
|
|
if wBegin.Code != http.StatusOK {
|
|
t.Fatalf("schema begin status: got %d, body: %s", wBegin.Code, wBegin.Body.String())
|
|
}
|
|
|
|
meta := fake.poolLastRunSnapshot().Metadata
|
|
schemaReqID := meta["iop_logical_request_id"]
|
|
schemaStageID := meta["iop_stage_id"]
|
|
if schemaReqID == "" || schemaStageID == "" {
|
|
t.Fatalf("schema begin identity incomplete: %+v", meta)
|
|
}
|
|
|
|
schemaAssistant := json.RawMessage(`{"role":"assistant","tool_calls":[{"id":"call_ts","type":"function","function":{"name":"search"}}]}`)
|
|
schemaHash, err := fingerprintCanonicalJSON(logicalRequestEndpointChat, schemaAssistant)
|
|
if err != nil {
|
|
t.Fatalf("fingerprintCanonicalJSON: %v", err)
|
|
}
|
|
if _, err := coord.awaitToolResults(schemaReqID, "edge-identity-test", schemaStageID, []logicalRequestExpectedTool{
|
|
{PublicCallID: "call_ts", ProviderCallID: "prov_ts"},
|
|
}, schemaHash); err != nil {
|
|
t.Fatalf("awaitToolResults: %v", err)
|
|
}
|
|
|
|
submitsBefore := fake.poolSubmitCountSnapshot()
|
|
// Continuation with a MUTATED tools schema (added "limit" property).
|
|
mutatedBody := `{
|
|
"model": "virtual-preset-rej",
|
|
"tools": [{"type": "function", "function": {"name": "search", "parameters": {"type": "object", "properties": {"q": {"type": "string"}, "limit": {"type": "number"}}}}}],
|
|
"messages": [
|
|
{"role": "user", "content": "schema initial"},
|
|
{"role": "assistant", "tool_calls": [{"id": "call_ts", "type": "function", "function": {"name": "search"}}]},
|
|
{"role": "tool", "tool_call_id": "call_ts", "content": "result"}
|
|
]
|
|
}`
|
|
reqMut := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(mutatedBody))
|
|
reqMut.Header.Set("Authorization", "Bearer "+rawToken1)
|
|
wMut := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(wMut, reqMut)
|
|
|
|
if wMut.Code != http.StatusBadRequest {
|
|
t.Fatalf("tool-schema mutation status: got %d, want 400. body: %s", wMut.Code, wMut.Body.String())
|
|
}
|
|
if got := fake.poolSubmitCountSnapshot(); got != submitsBefore {
|
|
t.Fatalf("provider dispatched on tool-schema mutation rejection: got %d, want %d", got, submitsBefore)
|
|
}
|
|
})
|
|
}
|
|
|
|
// TestPresetRequestIdentityAnthropicCountTokensBypassesCoordinator proves that a
|
|
// preset Anthropic count-tokens request served by the native tunnel fallback is
|
|
// not a Messages execution turn: it dispatches exactly one count-tokens
|
|
// submission, creates no logical execution state, and carries no
|
|
// request/call/stage identity metadata.
|
|
func TestPresetRequestIdentityAnthropicCountTokensBypassesCoordinator(t *testing.T) {
|
|
candidate := anthropicTestCandidate(t, "anthropic")
|
|
fake := &providerFakeRunService{
|
|
poolDispatchPath: string(edgeservice.ProviderPoolPathTunnel),
|
|
poolSelectedCandidate: candidate,
|
|
tunnelServedTarget: "upstream-claude",
|
|
tunnelFrames: anthropicTunnelFrames(http.StatusOK, "application/json", []byte(`{"input_tokens":11}`)),
|
|
}
|
|
|
|
preset := config.ExecutionPreset{
|
|
ID: "preset-anthropic-ct",
|
|
AllowedModes: []string{"direct"},
|
|
}
|
|
|
|
rawToken1 := "token-user-1"
|
|
sum1 := sha256.Sum256([]byte(rawToken1))
|
|
cfg := config.EdgeOpenAIConf{
|
|
PrincipalTokens: []config.OpenAIPrincipalTokenConf{
|
|
{TokenRef: "tok-1", TokenHashSHA256: hex.EncodeToString(sum1[:]), PrincipalRef: "user-1"},
|
|
},
|
|
}
|
|
|
|
srv := NewServer(cfg, fake, nil)
|
|
srv.SetEdgeID("edge-identity-test")
|
|
srv.SetExecutionPresets([]config.ExecutionPreset{preset})
|
|
srv.SetModelCatalog([]config.ModelCatalogEntry{
|
|
{
|
|
ID: "virtual-preset-anthropic-ct",
|
|
ExecutionPreset: "preset-anthropic-ct",
|
|
},
|
|
})
|
|
|
|
body := `{
|
|
"model": "virtual-preset-anthropic-ct",
|
|
"messages": [{"role": "user", "content": "count me"}]
|
|
}`
|
|
req := httptest.NewRequest(http.MethodPost, "/v1/messages/count_tokens", strings.NewReader(body))
|
|
req.Header.Set("X-Api-Key", rawToken1)
|
|
req.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
|
|
w := httptest.NewRecorder()
|
|
srv.routes().ServeHTTP(w, req)
|
|
|
|
if w.Code != http.StatusOK {
|
|
t.Fatalf("count-tokens status: got %d, body: %s", w.Code, w.Body.String())
|
|
}
|
|
if got := w.Body.String(); got != `{"input_tokens":11}` {
|
|
t.Fatalf("count-tokens body: got %s", got)
|
|
}
|
|
|
|
// Exactly one native count-tokens provider submission.
|
|
if got := fake.poolSubmitCountSnapshot(); got != 1 {
|
|
t.Fatalf("count-tokens pool submit count: got %d, want 1", got)
|
|
}
|
|
reqs := fake.tunnelReqsSnapshot()
|
|
if len(reqs) != 1 || reqs[0].Operation != string(config.OperationCountTokens) {
|
|
t.Fatalf("native count-tokens request mismatch: %+v", reqs)
|
|
}
|
|
|
|
// Zero logical execution state and no request/call/stage identity metadata.
|
|
coord := srv.logicalRequests()
|
|
coord.mu.Lock()
|
|
records := len(coord.requests)
|
|
coord.mu.Unlock()
|
|
if records != 0 {
|
|
t.Fatalf("count-tokens created %d coordinator records, want 0", records)
|
|
}
|
|
meta := fake.poolLastRunSnapshot().Metadata
|
|
for _, key := range []string{"iop_logical_request_id", "iop_call_id", "iop_stage_id"} {
|
|
if v, ok := meta[key]; ok && v != "" {
|
|
t.Fatalf("count-tokens leaked identity metadata %s=%q", key, v)
|
|
}
|
|
}
|
|
}
|