homelab/review-bot/anthropic_test.go

package main

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"
)

// fakeAnthropic answers each Messages request with the next scripted event stream.
type fakeAnthropic struct {
	t       *testing.T
	srv     *httptest.Server
	mu      sync.Mutex
	streams []string
	status  []int
	bodies  []map[string]any
	betas   []string
}

func block(index int, cb string) string {
	return `{"type":"content_block_start","index":` + itoa(index) + `,"content_block":` + cb + `}` + "\n" +
		`{"type":"content_block_stop","index":` + itoa(index) + `}`
}

func itoa(i int) string { b, _ := json.Marshal(i); return string(b) }

// messageStream renders event lines as the SSE stream of one message ending in stop.
func messageStream(model, stop string, events ...string) string {
	lines := []string{`{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"` + model +
		`","content":[],"stop_reason":null,"usage":{"input_tokens":1,"output_tokens":1}}}`}
	for _, e := range events {
		lines = append(lines, strings.Split(e, "\n")...)
	}
	lines = append(lines, `{"type":"message_delta","delta":{"stop_reason":"`+stop+`"},"usage":{"output_tokens":5}}`,
		`{"type":"message_stop"}`)
	var b strings.Builder
	for _, l := range lines {
		var ev struct {
			Type string `json:"type"`
		}
		_ = json.Unmarshal([]byte(l), &ev)
		b.WriteString("event: " + ev.Type + "\ndata: " + l + "\n\n")
	}
	return b.String()
}

func newFakeAnthropic(t *testing.T) *fakeAnthropic {
	f := &fakeAnthropic{t: t}
	f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/v1/messages" || r.Header.Get("X-Api-Key") != "anthropic-key" {
			http.Error(w, `{"type":"error","error":{"type":"authentication_error","message":"no"}}`, http.StatusUnauthorized)
			return
		}
		var in map[string]any
		_ = json.NewDecoder(r.Body).Decode(&in)
		f.mu.Lock()
		defer f.mu.Unlock()
		f.bodies = append(f.bodies, in)
		f.betas = append(f.betas, r.Header.Get("Anthropic-Beta"))
		if len(f.status) > 0 {
			st := f.status[0]
			f.status = f.status[1:]
			w.Header().Set("Retry-After", "0")
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(st)
			_, _ = io.WriteString(w, `{"type":"error","error":{"type":"rate_limit_error","message":"slow down"}}`)
			return
		}
		if len(f.streams) == 0 {
			http.Error(w, "no more scripted streams", http.StatusInternalServerError)
			return
		}
		s := f.streams[0]
		f.streams = f.streams[1:]
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = io.WriteString(w, s)
	}))
	t.Cleanup(f.srv.Close)
	return f
}

func (f *fakeAnthropic) backend() *Anthropic {
	return newAnthropic(Config{AnthropicURL: f.srv.URL, AnthropicKey: "anthropic-key", MCPURL: "https://mcp.example/mcp",
		MCPToken: "read-token", MCPTools: []string{"get_file_contents", "pull_request_read"}, AnthropicEffort: "high"})
}

const toolUse = `{"type":"mcp_tool_use","id":"t1","name":"get_file_contents","server_name":"gitea","input":{"path":"a.yaml"}}`
const toolResult = `{"type":"mcp_tool_result","tool_use_id":"t1","is_error":false,"content":[{"type":"text","text":"x: 1"}]}`

func textBlock(index int, text string) string {
	b, _ := json.Marshal(text)
	return `{"type":"content_block_start","index":` + itoa(index) + `,"content_block":{"type":"text","text":""}}` + "\n" +
		`{"type":"content_block_delta","index":` + itoa(index) + `,"delta":{"type":"text_delta","text":` + string(b) + `}}` + "\n" +
		`{"type":"content_block_stop","index":` + itoa(index) + `}`
}

func TestAnthropicResumesAPausedTurnAndTakesTheFinalText(t *testing.T) {
	f := newFakeAnthropic(t)
	f.streams = []string{
		messageStream("claude-opus-5-5", "pause_turn", textBlock(0, "Let me look."), block(1, toolUse), block(2, toolResult)),
		messageStream("claude-opus-5-5", "end_turn", block(0, toolUse), block(1, toolResult),
			textBlock(2, "## Summary\n\nok\n\nVerdict: comment")),
	}
	a, err := f.backend().Ask(context.Background(), "claude-opus-5-5", "Review pull request #7 of konairius/homelab.")
	if err != nil {
		t.Fatal(err)
	}
	if a.Text != "## Summary\n\nok\n\nVerdict: comment" || a.ToolCalls != 2 || len(a.Violations) != 0 {
		t.Errorf("answer %+v", a)
	}
	if len(f.bodies) != 2 {
		t.Fatalf("requests %d, want 2", len(f.bodies))
	}
	if msgs := f.bodies[1]["messages"].([]any); len(msgs) != 2 || msgs[1].(map[string]any)["role"] != "assistant" {
		t.Errorf("the paused turn must be sent back: %v", msgs)
	}

	body := f.bodies[0]
	servers := body["mcp_servers"].([]any)
	srv := servers[0].(map[string]any)
	if srv["url"] != "https://mcp.example/mcp" || srv["authorization_token"] != "read-token" || srv["name"] != "gitea" {
		t.Errorf("mcp server: %v", srv)
	}
	toolset := body["tools"].([]any)[0].(map[string]any)
	configs := toolset["configs"].(map[string]any)
	if toolset["type"] != "mcp_toolset" || toolset["default_config"].(map[string]any)["enabled"] != false ||
		len(configs) != 2 || configs["get_file_contents"].(map[string]any)["enabled"] != true {
		t.Errorf("toolset must enable only the listed tools: %v", toolset)
	}
	if body["output_config"].(map[string]any)["effort"] != "high" || body["fallbacks"] != "default" {
		t.Errorf("effort/fallbacks: %v %v", body["output_config"], body["fallbacks"])
	}
	if !strings.Contains(f.betas[0], "mcp-client-2025-11-20") || !strings.Contains(f.betas[0], "server-side-fallback-2026-07-01") {
		t.Errorf("betas: %q", f.betas[0])
	}
}

func TestAnthropicFlagsToolsOutsideTheList(t *testing.T) {
	f := newFakeAnthropic(t)
	f.streams = []string{messageStream("claude-haiku-5-5", "end_turn",
		block(0, strings.Replace(toolUse, "get_file_contents", "create_or_update_file", 1)), textBlock(1, "## Summary\n\nx"))}
	a, err := f.backend().Ask(context.Background(), "claude-haiku-5-5", "task")
	if err != nil || len(a.Violations) != 1 || a.Violations[0] != "create_or_update_file" {
		t.Errorf("answer %+v, err %v", a, err)
	}
	if _, ok := f.bodies[0]["fallbacks"]; ok {
		t.Error("haiku has no server-side fallback")
	}
}

func TestAnthropicRateLimitIsRetryable(t *testing.T) {
	f := newFakeAnthropic(t)
	f.status = []int{429}
	_, err := f.backend().Ask(context.Background(), "claude-opus-5-5", "task")
	api := asAPIError(err)
	if !retryable(err) || api == nil || api.Status != 429 || api.Provider != "anthropic" || !api.HasRetryAfter {
		t.Errorf("want a retryable anthropic 429 with Retry-After, got %#v (%v)", api, err)
	}
}

// With the selector on, the review runs on the model TypeSafe picks; when TypeSafe fails, on the
// default model.
func TestSelectorPicksTheBackend(t *testing.T) {
	for _, tc := range []struct {
		name      string
		answer    string
		status    int
		wantModel string
	}{
		{"chosen", `{"answers":{"model":{"type":"choice","choice":"anthropic:claude-opus-5-5","confidence":0.8,` +
			`"probabilities":{"anthropic:claude-opus-5-5":0.9,"mistral:mistral-large-4-0":0.1}}}}`, 200, "claude-opus-5-5"},
		{"failed", `{"error":"overloaded"}`, 529, "mistral-large-4-0"},
		{"unknown model", `{"answers":{"model":{"type":"choice","choice":"x:y"}}}`, 200, "mistral-large-4-0"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newFixture(t)
			fa := newFakeAnthropic(t)
			fa.streams = []string{messageStream("claude-opus-5-5", "end_turn",
				textBlock(0, "## Summary\n\nfine\n\n## Recommendation\n\nLooks reasonable.\n\nVerdict: comment"))}
			var state map[string]any
			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				var in map[string]any
				_ = json.NewDecoder(r.Body).Decode(&in)
				state = in["state"].(map[string]any)
				q := in["questions"].(map[string]any)["model"].(map[string]any)
				if r.Header.Get("Authorization") != "Bearer ts-key" || in["model"] != "jev-latest" || q["type"] != "choice" ||
					q["criteria"].(map[string]any)["anthropic:claude-opus-5-5"] != "strong" {
					http.Error(w, "unexpected request", http.StatusBadRequest)
					return
				}
				w.WriteHeader(tc.status)
				_, _ = io.WriteString(w, tc.answer)
			}))
			t.Cleanup(ts.Close)
			f.rev.cfg.Models = []Model{{Backend: "mistral", Name: "mistral-large-4-0", Description: "default"},
				{Backend: "anthropic", Name: "claude-opus-5-5", Description: "strong"}}
			f.rev.backends["anthropic"] = fa.backend()
			f.rev.setReady("anthropic", true)
			f.rev.selector = &TypeSafe{url: ts.URL, key: "ts-key", model: "jev-latest", http: &http.Client{Timeout: 5 * time.Second}}

			if err := f.rev.Review(context.Background(), "konairius/homelab", 7); err != nil {
				t.Fatal(err)
			}
			posted := f.postedReviews()
			if len(posted) != 1 || !strings.Contains(posted[0]["body"], "model `"+tc.wantModel+"`") {
				t.Fatalf("want one review by %s, got %v", tc.wantModel, posted)
			}
			if pr := state["pull_request"].(map[string]any); pr["title"] != "feat: x" || pr["repository"] != "konairius/homelab" {
				t.Errorf("selector state: %v", state)
			}
		})
	}
}