iop/apps/node/internal/adapters/openai_compat/protocol_profile_test.go

package openai_compat

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"go.uber.org/zap"

	runtime "iop/packages/go/agentruntime"
	"iop/packages/go/config"
)

// TestProtocolProfileOperationURLFixtures executes each documented Chat or
// Messages variant against a loopback upstream. Catalog endpoints and auth
// were checked against official provider documentation on 2026-08-02; GLM
// General and Coding Plan endpoints are distinct, while MiMo pay-go endpoints
// are used and plan-specific MiMo endpoints are intentionally not guessed.
func TestProtocolProfileOperationURLFixtures(t *testing.T) {
	fixtures := []struct {
		id         string
		operation  config.ProtocolOperation
		basePrefix string
		wantURI    string
		authHeader string
	}{
		{id: "openai", operation: config.OperationChatCompletions, basePrefix: "/v1", wantURI: "/v1/chat/completions", authHeader: "Authorization"},
		{id: "gemini", operation: config.OperationChatCompletions, basePrefix: "/v1beta/openai", wantURI: "/v1beta/openai/chat/completions", authHeader: "Authorization"},
		{id: "glm", operation: config.OperationChatCompletions, basePrefix: "/api/paas/v4", wantURI: "/api/paas/v4/chat/completions", authHeader: "Authorization"},
		{id: "glm_coding", operation: config.OperationChatCompletions, basePrefix: "/api/coding/paas/v4", wantURI: "/api/coding/paas/v4/chat/completions", authHeader: "Authorization"},
		{id: "kimi", operation: config.OperationChatCompletions, basePrefix: "/v1", wantURI: "/v1/chat/completions", authHeader: "Authorization"},
		{id: "minimax_chat", operation: config.OperationChatCompletions, basePrefix: "/v1", wantURI: "/v1/chat/completions", authHeader: "Authorization"},
		{id: "mimo_chat", operation: config.OperationChatCompletions, basePrefix: "/v1", wantURI: "/v1/chat/completions", authHeader: "Authorization"},
		{id: "grok", operation: config.OperationChatCompletions, basePrefix: "/v1", wantURI: "/v1/chat/completions", authHeader: "Authorization"},
		{id: "seulgi_chat", operation: config.OperationChatCompletions, basePrefix: "/openai/v1", wantURI: "/openai/v1/chat/completions", authHeader: "Authorization"},
		{id: "anthropic", operation: config.OperationMessages, wantURI: "/v1/messages", authHeader: "X-Api-Key"},
		{id: "minimax_messages", operation: config.OperationMessages, basePrefix: "/anthropic", wantURI: "/anthropic/v1/messages", authHeader: "Authorization"},
		{id: "mimo_messages", operation: config.OperationMessages, basePrefix: "/anthropic", wantURI: "/anthropic/v1/messages", authHeader: "api-key"},
		{id: "seulgi_messages", operation: config.OperationMessages, basePrefix: "/anthropic/v1", wantURI: "/anthropic/v1/messages", authHeader: "X-Api-Key"},
	}

	for _, fx := range fixtures {
		t.Run(fx.id, func(t *testing.T) {
			outcomes := []struct {
				name        string
				status      int
				contentType string
				body        []byte
			}{
				{name: "json", status: http.StatusOK, contentType: "application/json", body: []byte(`{"ok":true}`)},
				{name: "sse", status: http.StatusOK, contentType: "text/event-stream", body: fixtureSSEBody(fx.operation)},
				{name: "provider_error", status: http.StatusUnprocessableEntity, contentType: "application/json", body: []byte(`{"error":{"message":"fixture"}}`)},
			}
			for _, outcome := range outcomes {
				t.Run(outcome.name, func(t *testing.T) {
					requests := 0
					server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
						requests++
						if r.RequestURI != fx.wantURI {
							t.Errorf("RequestURI = %q, want %q", r.RequestURI, fx.wantURI)
						}
						wantAuth := "test-secret"
						if fx.authHeader == "Authorization" {
							wantAuth = "Bearer test-secret"
						}
						if got := r.Header.Get(fx.authHeader); got != wantAuth {
							t.Errorf("%s = %q, want %q", fx.authHeader, got, wantAuth)
						}
						var payload map[string]any
						if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
							t.Errorf("decode body: %v", err)
						}
						if got := payload["model"]; got != "upstream-model" {
							t.Errorf("model = %v, want upstream-model", got)
						}
						w.Header().Set("Content-Type", outcome.contentType)
						w.WriteHeader(outcome.status)
						_, _ = w.Write(outcome.body)
					}))
					defer server.Close()

					profile := mustResolveProfile(t, fx.id)
					profile.BaseURL = server.URL + fx.basePrefix
					profile.ModelMapping = map[string]string{"canonical-model": "upstream-model"}
					adapter := New(config.OpenAICompatConf{
						Headers:        map[string]string{"Authorization": "Bearer test-secret"},
						RuntimeProfile: &profile,
					}, zap.NewNop())
					sink := &fakeTunnelSink{}
					err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
						RunID:     "run-" + fx.id + "-" + outcome.name,
						TunnelID:  "tunnel-" + fx.id + "-" + outcome.name,
						Method:    http.MethodPost,
						Operation: string(fx.operation),
						Body:      []byte(`{"model":"canonical-model","messages":[]}`),
					}, sink)
					if err != nil {
						t.Fatalf("TunnelProvider: %v", err)
					}
					if requests != 1 {
						t.Fatalf("requests = %d, want 1", requests)
					}

					frames := sink.all()
					if len(frames) < 3 {
						t.Fatalf("expected at least 3 frames (start/body/end), got %d", len(frames))
					}
					if frames[0].Kind != runtime.ProviderTunnelFrameKindResponseStart {
						t.Fatalf("expected RESPONSE_START, got %s", frames[0].Kind)
					}
					if frames[0].StatusCode != outcome.status {
						t.Fatalf("ResponseStart status = %d, want %d", frames[0].StatusCode, outcome.status)
					}
					if got := frames[0].Headers["Content-Type"]; got != outcome.contentType {
						t.Fatalf("ResponseStart Content-Type = %q, want %q", got, outcome.contentType)
					}

					var bodyBuffer bytes.Buffer
					for _, f := range frames[1 : len(frames)-1] {
						if f.Kind != runtime.ProviderTunnelFrameKindBody {
							t.Errorf("expected BODY kind, got %s", f.Kind)
						}
						bodyBuffer.Write(f.Body)
					}
					if !bytes.Equal(bodyBuffer.Bytes(), outcome.body) {
						t.Fatalf("body mismatch:\n got: %q\nwant: %q", bodyBuffer.Bytes(), outcome.body)
					}

					lastFrame := frames[len(frames)-1]
					if outcome.name == "provider_error" {
						if lastFrame.Kind != runtime.ProviderTunnelFrameKindEnd {
							t.Fatalf("provider HTTP error must end with END, got %s", lastFrame.Kind)
						}
					} else {
						if lastFrame.Kind != runtime.ProviderTunnelFrameKindEnd {
							t.Fatalf("expected END, got %s", lastFrame.Kind)
						}
						if !lastFrame.End {
							t.Fatal("expected End true")
						}
					}
					assertOrderedTunnelFrames(t, frames)
				})
			}
		})
	}
}

func TestProtocolProfileModelMappingPreservesRawBody(t *testing.T) {
	profile := mustResolveProfile(t, "openai")
	profile.ModelMapping = map[string]string{"canonical-model": "upstream-model"}
	adapter := New(config.OpenAICompatConf{RuntimeProfile: &profile}, zap.NewNop())
	body := []byte("{\n  \"model\" : \"canonical-model\", \"large\":9007199254740993, \"exp\":1.2300e+04, \"nested\":{\"model\":\"leave-me\"}, \"escaped\":\"a\\\\b\\\"c\"\n}")
	want := []byte("{\n  \"model\" : \"upstream-model\", \"large\":9007199254740993, \"exp\":1.2300e+04, \"nested\":{\"model\":\"leave-me\"}, \"escaped\":\"a\\\\b\\\"c\"\n}")
	got, err := adapter.mapProfileBodyModel(body)
	if err != nil {
		t.Fatalf("mapProfileBodyModel: %v", err)
	}
	if string(got) != string(want) {
		t.Fatalf("mapped body bytes changed unexpectedly:\n got: %s\nwant: %s", got, want)
	}

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		upstreamBody, readErr := io.ReadAll(r.Body)
		if readErr != nil {
			t.Errorf("read body: %v", readErr)
		}
		if string(upstreamBody) != string(want) {
			t.Errorf("upstream body bytes:\n got: %s\nwant: %s", upstreamBody, want)
		}
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"ok":true}`))
	}))
	defer server.Close()
	profile.BaseURL = server.URL + "/v1"
	adapter = New(config.OpenAICompatConf{RuntimeProfile: &profile}, zap.NewNop())
	if err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
		RunID: "raw-body", TunnelID: "raw-body", Method: http.MethodPost,
		Operation: string(config.OperationChatCompletions), Body: body,
	}, &fakeTunnelSink{}); err != nil {
		t.Fatalf("TunnelProvider: %v", err)
	}
}

func TestProtocolProfileModelMappingIdentityAndInvalidBodies(t *testing.T) {
	profile := mustResolveProfile(t, "openai")
	profile.ModelMapping = map[string]string{"canonical": "upstream", "identity": "identity"}
	adapter := New(config.OpenAICompatConf{RuntimeProfile: &profile}, zap.NewNop())
	for _, body := range [][]byte{
		[]byte(`{"other":1}`),
		[]byte(`{"model":123,"large":9007199254740993}`),
		[]byte(` { "model" : "identity", "n":1.00 } `),
	} {
		got, err := adapter.mapProfileBodyModel(body)
		if err != nil {
			t.Fatalf("identity body %s: %v", body, err)
		}
		if string(got) != string(body) {
			t.Fatalf("identity body changed: got %s want %s", got, body)
		}
	}
	for _, body := range [][]byte{
		[]byte(`{"model":"canonical"`),
		[]byte(`["canonical"]`),
		[]byte(`null`),
	} {
		if _, err := adapter.mapProfileBodyModel(body); err == nil || !strings.Contains(err.Error(), "decode profile request body") {
			t.Fatalf("malformed/non-object body %s error = %v", body, err)
		}
	}
}

// TestProtocolProfileAuthHeaders exercises actual HTTP requests. The expected
// MiniMax and MiMo Anthropic-compatible headers were checked on 2026-08-01:
// https://platform.minimax.io/docs/token-plan/other-tools
// https://mimo.mi.com/docs/en-US/quick-start/faq/api-integration
// https://mimo.mi.com/docs/en-US/quick-start/summary/first-api-call
func TestProtocolProfileAuthHeaders(t *testing.T) {
	for _, tc := range []struct {
		profile string
		seed    map[string]string
		wantKey string
		wantVal string
		absent  []string
	}{
		{profile: "minimax_messages", seed: map[string]string{"X-Api-Key": "test-secret"}, wantKey: "Authorization", wantVal: "Bearer test-secret", absent: []string{"X-Api-Key"}},
		{profile: "mimo_messages", seed: map[string]string{"Authorization": "Bearer test-secret", "X-Api-Key": "stale"}, wantKey: "api-key", wantVal: "test-secret", absent: []string{"Authorization", "X-Api-Key"}},
	} {
		t.Run(tc.profile, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if got := r.Header.Get(tc.wantKey); got != tc.wantVal {
					t.Errorf("%s = %q, want %q", tc.wantKey, got, tc.wantVal)
				}
				for _, key := range tc.absent {
					if got := r.Header.Get(key); got != "" {
						t.Errorf("stale %s = %q", key, got)
					}
				}
				w.Header().Set("Content-Type", "application/json")
				_, _ = w.Write([]byte(`{"ok":true}`))
			}))
			defer server.Close()
			profile := mustResolveProfile(t, tc.profile)
			profile.BaseURL = server.URL + "/anthropic"
			adapter := New(config.OpenAICompatConf{Headers: tc.seed, RuntimeProfile: &profile}, zap.NewNop())
			if err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
				RunID: tc.profile, TunnelID: tc.profile, Method: http.MethodPost,
				Operation: string(config.OperationMessages), Body: []byte(`{"model":"test","messages":[]}`),
			}, &fakeTunnelSink{}); err != nil {
				t.Fatalf("TunnelProvider: %v", err)
			}
		})
	}
}

// TestProtocolProfileUnsupportedOperationRejectedBeforeDispatch verifies that
// an unsupported operation (e.g. chat_completions on anthropic) makes zero
// upstream requests.
func TestProtocolProfileUnsupportedOperationRejectedBeforeDispatch(t *testing.T) {
	requestCount := 0
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		requestCount++
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = w.Write([]byte(`data: [DONE]`))
	}))
	defer server.Close()

	profile := mustResolveProfile(t, "anthropic")
	adapter := New(config.OpenAICompatConf{
		Endpoint:       server.URL,
		RuntimeProfile: &profile,
	}, zap.NewNop())

	sink := &fakeTunnelSink{}
	err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
		RunID:     "run-unsupported",
		TunnelID:  "tunnel-unsupported",
		Adapter:   "anthropic",
		Target:    "claude-3",
		Method:    http.MethodPost,
		Operation: string(config.OperationChatCompletions),
		Path:      "/chat/completions",
	}, sink)

	if err == nil {
		t.Fatal("expected error for unsupported operation, got nil")
	}
	if !strings.Contains(err.Error(), "not supported") {
		t.Fatalf("expected 'not supported' error, got %v", err)
	}
	if requestCount != 0 {
		t.Fatalf("expected zero upstream requests for unsupported operation, got %d", requestCount)
	}
	openAIProfile := mustResolveProfile(t, "openai")
	openAIProfile.BaseURL = server.URL + "/v1"
	openAIAdapter := New(config.OpenAICompatConf{RuntimeProfile: &openAIProfile}, zap.NewNop())
	err = openAIAdapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
		RunID: "run-empty-operation", TunnelID: "tunnel-empty-operation", Method: http.MethodPost,
		Path: "/v1/chat/completions", Body: []byte(`{"model":"canonical"}`),
	}, &fakeTunnelSink{})
	if err == nil || !strings.Contains(err.Error(), "operation must not be empty") {
		t.Fatalf("expected empty operation rejection, got %v", err)
	}
	if requestCount != 0 {
		t.Fatalf("expected zero upstream requests after empty operation, got %d", requestCount)
	}
}

// TestProtocolProfileRequestHeadersAndModelMapping covers production Models and
// Responses callers in addition to the catalog variant table above.
func TestProtocolProfileRequestHeadersAndModelMapping(t *testing.T) {
	t.Run("models", func(t *testing.T) {
		requests := 0
		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			requests++
			if r.RequestURI != "/v1/models" {
				t.Errorf("RequestURI = %q, want /v1/models", r.RequestURI)
			}
			if got := r.Header.Get("Authorization"); got != "Bearer test-secret" {
				t.Errorf("Authorization = %q", got)
			}
			_, _ = w.Write([]byte(`{"data":[{"id":"upstream-model"}]}`))
		}))
		defer server.Close()
		profile := mustResolveProfile(t, "openai")
		profile.BaseURL = server.URL + "/v1"
		adapter := New(config.OpenAICompatConf{
			Headers:        map[string]string{"Authorization": "test-secret"},
			RuntimeProfile: &profile,
		}, zap.NewNop())
		models, err := adapter.fetchTargets(context.Background())
		if err != nil {
			t.Fatalf("fetchTargets: %v", err)
		}
		if len(models) != 1 || models[0] != "upstream-model" || requests != 1 {
			t.Fatalf("models=%v requests=%d", models, requests)
		}
	})

	t.Run("responses_absolute_operation", func(t *testing.T) {
		requests := 0
		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			requests++
			if r.RequestURI != "/absolute/responses?version=1" {
				t.Errorf("RequestURI = %q", r.RequestURI)
			}
			var body map[string]any
			if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
				t.Fatalf("decode body: %v", err)
			}
			if body["model"] != "upstream-response" {
				t.Errorf("model = %v", body["model"])
			}
			_, _ = w.Write([]byte(`{"id":"resp"}`))
		}))
		defer server.Close()
		profile := mustResolveProfile(t, "openai")
		profile.Operations[string(config.OperationResponses)] = server.URL + "/absolute/responses?version=1"
		profile.ModelMapping = map[string]string{"canonical-response": "upstream-response"}
		adapter := New(config.OpenAICompatConf{
			Headers:        map[string]string{"Authorization": "Bearer test-secret"},
			RuntimeProfile: &profile,
		}, zap.NewNop())
		err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
			RunID: "run-responses", TunnelID: "tunnel-responses", Method: http.MethodPost,
			Operation: string(config.OperationResponses), Body: []byte(`{"model":"canonical-response"}`),
		}, &fakeTunnelSink{})
		if err != nil {
			t.Fatalf("TunnelProvider: %v", err)
		}
		if requests != 1 {
			t.Fatalf("requests = %d, want 1", requests)
		}
	})
}

// TestLegacyEndpointPathFallback verifies that when no profile is present,
// the legacy endpoint + path fallback works correctly.
func TestLegacyEndpointPathFallback(t *testing.T) {
	var gotPath string
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = w.Write([]byte(`data: [DONE]`))
	}))
	defer server.Close()

	adapter := New(config.OpenAICompatConf{
		Endpoint: server.URL,
	}, zap.NewNop())

	sink := &fakeTunnelSink{}
	err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
		RunID:    "run-legacy",
		TunnelID: "tunnel-legacy",
		Adapter:  "openai_compat",
		Target:   "gpt-4",
		Method:   http.MethodPost,
		Path:     "/v1/chat/completions",
	}, sink)
	if err != nil {
		t.Fatalf("TunnelProvider: %v", err)
	}

	if gotPath != "/v1/chat/completions" {
		t.Errorf("expected legacy path /v1/chat/completions, got %q", gotPath)
	}
}

// TestProtocolProfileTunnelOperationSurvivesHandler verifies that the operation
// field survives the parser/handler boundary.
func TestProtocolProfileTunnelOperationSurvivesHandler(t *testing.T) {
	var gotPath string
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = w.Write([]byte(`data: [DONE]`))
	}))
	defer server.Close()

	// Override the profile's base_url to point at the test server so the
	// profile-resolved URL reaches the test server.
	profile := mustResolveProfile(t, "openai")
	profile.BaseURL = server.URL + "/v1"
	adapter := New(config.OpenAICompatConf{
		Endpoint:       server.URL,
		RuntimeProfile: &profile,
	}, zap.NewNop())

	sink := &fakeTunnelSink{}
	err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
		RunID:     "run-op-survives",
		TunnelID:  "tunnel-op-survives",
		Adapter:   "openai",
		Target:    "gpt-4",
		Method:    http.MethodPost,
		Operation: string(config.OperationChatCompletions),
	}, sink)
	if err != nil {
		t.Fatalf("TunnelProvider: %v", err)
	}

	if gotPath != "/v1/chat/completions" {
		t.Errorf("expected /v1/chat/completions, got %q", gotPath)
	}
}

// TestProtocolProfileTunnelLegacyPathFallback verifies that legacy Path field
// still works when no profile is present.
func TestProtocolProfileTunnelLegacyPathFallback(t *testing.T) {
	var gotPath string
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = w.Write([]byte(`data: [DONE]`))
	}))
	defer server.Close()

	adapter := New(config.OpenAICompatConf{
		Endpoint: server.URL,
	}, zap.NewNop())

	sink := &fakeTunnelSink{}
	err := adapter.TunnelProvider(context.Background(), runtime.ProviderTunnelRequest{
		RunID:    "run-legacy-fallback",
		TunnelID: "tunnel-legacy-fallback",
		Adapter:  "openai_compat",
		Target:   "gpt-4",
		Method:   http.MethodPost,
		Path:     "/v1/chat/completions",
	}, sink)
	if err != nil {
		t.Fatalf("TunnelProvider: %v", err)
	}

	if gotPath != "/v1/chat/completions" {
		t.Errorf("expected /v1/chat/completions, got %q", gotPath)
	}
}

func TestProtocolProfileAdapterSnapshotClone(t *testing.T) {
	profile := mustResolveProfile(t, "openai")
	profile.Extensions = map[string]any{"nested": map[string]any{"value": "original"}}
	adapter := New(config.OpenAICompatConf{RuntimeProfile: &profile}, zap.NewNop())
	profile.BaseURL = "https://mutated.example.invalid"
	profile.Operations[string(config.OperationChatCompletions)] = "/mutated"
	profile.Extensions["nested"].(map[string]any)["value"] = "mutated"
	if adapter.profile.BaseURL == profile.BaseURL || adapter.profile.Operations[string(config.OperationChatCompletions)] == "/mutated" {
		t.Fatal("source mutation crossed adapter snapshot boundary")
	}
	if got := adapter.profile.Extensions["nested"].(map[string]any)["value"]; got != "original" {
		t.Fatalf("nested extension mutation crossed adapter boundary: %v", got)
	}
}

// fixtureSSEBody returns deterministic SSE bytes appropriate for the operation
// so every profile's streaming leg can be qualified without a provider.
func fixtureSSEBody(operation config.ProtocolOperation) []byte {
	if operation == config.OperationMessages {
		return []byte("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_fixture\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"model\":\"upstream-model\",\"stop_reason\":null,\"stop_sequence\":null,\"usage\":{\"input_tokens\":1,\"output_tokens\":0}}}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")
	}
	return []byte("data: {\"choices\":[{\"delta\":{\"content\":\"fixture\"}}]}\n\ndata: [DONE]\n\n")
}

func mustResolveProfile(t *testing.T, id string) config.ConcreteProtocolProfile {
	t.Helper()
	profile, err := config.ResolveProtocolProfile(id, "", config.BuiltInProtocolProfiles)
	if err != nil {
		t.Fatalf("ResolveProtocolProfile(%q): %v", id, err)
	}
	return profile
}