diff --git a/kubernetes/apps/review-bot/deployment.yaml b/kubernetes/apps/review-bot/deployment.yaml index 550d322e..368da6e0 100644 --- a/kubernetes/apps/review-bot/deployment.yaml +++ b/kubernetes/apps/review-bot/deployment.yaml @@ -40,10 +40,32 @@ spec: value: https://gitea.renner.uno - name: AGENT_NAME value: pr-review - - name: AGENT_MODEL - value: mistral-large-4-0 - name: CONNECTOR_ID value: 01a12adf-cdc3-7562-b32b-b1fa96be244c + # The first model is the default; a backend without its keys is off. + - name: MODELS + value: | + mistral:mistral-large-4-0 = Mistral Large 4. Solid on routine changes of small to medium size. + anthropic:claude-haiku-5-5 = Claude Haiku 5.5. Cheapest and fastest; only for trivial changes such as documentation, comments or version bumps. + anthropic:claude-sonnet-5-5 = Claude Sonnet 5.5. Fast and capable; good for most code and configuration changes. + anthropic:claude-opus-5-5 = Claude Opus 5.5. The strongest reviewer; for large or cross-cutting changes and anything touching security, secrets, network policy or data. + - name: SELECTOR + value: typesafe + - name: ANTHROPIC_API_KEY + valueFrom: + secretKeyRef: + name: review-bot-secrets + key: ANTHROPIC_API_KEY + - name: MCP_TOKEN + valueFrom: + secretKeyRef: + name: review-bot-secrets + key: MCP_TOKEN + - name: TYPESAFE_API_KEY + valueFrom: + secretKeyRef: + name: review-bot-secrets + key: TYPESAFE_API_KEY - name: WEBHOOK_SECRET valueFrom: secretKeyRef: diff --git a/kubernetes/apps/review-bot/externalsecret.yaml b/kubernetes/apps/review-bot/externalsecret.yaml index eddd5241..b6e80a9c 100644 --- a/kubernetes/apps/review-bot/externalsecret.yaml +++ b/kubernetes/apps/review-bot/externalsecret.yaml @@ -17,10 +17,23 @@ spec: property: webhook-secret - secretKey: MISTRAL_API_KEY remoteRef: - key: apps/wiki - property: minstral-token - # The review-bot token, rotated quarterly by the review-bot-secret-rotate CronJob. + key: apps/review-bot + property: minstral-api-key + - secretKey: ANTHROPIC_API_KEY + remoteRef: + key: apps/review-bot + property: antropic-api-key + - secretKey: TYPESAFE_API_KEY + remoteRef: + key: apps/review-bot + property: typesafe-api-key + # The review-bot tokens, rotated quarterly by the review-bot-secret-rotate CronJob; MCP_TOKEN is + # the read-only one for the Gitea MCP server. - secretKey: REVIEW_TOKEN remoteRef: key: infrastructure/review-bot property: token + - secretKey: MCP_TOKEN + remoteRef: + key: infrastructure/review-bot + property: mcp_token diff --git a/kubernetes/infrastructure/network-policies/review-bot-ns.yaml b/kubernetes/infrastructure/network-policies/review-bot-ns.yaml index 292257f7..09f07cdb 100644 --- a/kubernetes/infrastructure/network-policies/review-bot-ns.yaml +++ b/kubernetes/infrastructure/network-policies/review-bot-ns.yaml @@ -58,7 +58,7 @@ spec: - matchLabels: "k8s:io.kubernetes.pod.namespace": ingress app.kubernetes.io/name: traefik - # api.mistral.ai. + # api.mistral.ai, api.anthropic.com, api.typesafe.ai. - toEntities: - world toPorts: diff --git a/kubernetes/infrastructure/openbao/policies/rotator-review-bot.hcl b/kubernetes/infrastructure/openbao/policies/rotator-review-bot.hcl index e3aac450..2831725c 100644 --- a/kubernetes/infrastructure/openbao/policies/rotator-review-bot.hcl +++ b/kubernetes/infrastructure/openbao/policies/rotator-review-bot.hcl @@ -30,6 +30,8 @@ # # seed the target token field so gitea-token's assert_fields_exist passes on # # the first run (any non-empty placeholder; the first rotation overwrites it): # bao kv put homelab/infrastructure/review-bot token=placeholder +# # same for the read-only MCP token (rotate-mint-mcp): +# bao kv patch homelab/infrastructure/review-bot mcp_token=placeholder # # Do NOT set token_no_default_policy=true (kv-v2 preflight needs the default policy). diff --git a/kubernetes/infrastructure/openbao/rotation/review-bot.yaml b/kubernetes/infrastructure/openbao/rotation/review-bot.yaml index 5f8ce27f..34578254 100644 --- a/kubernetes/infrastructure/openbao/rotation/review-bot.yaml +++ b/kubernetes/infrastructure/openbao/rotation/review-bot.yaml @@ -127,6 +127,42 @@ spec: memory: 16Mi limits: memory: 48Mi + # Read-only review-bot PAT for the Anthropic backend on the Gitea MCP server. + - name: rotate-mint-mcp + image: gitea.renner.uno/gitea_admin/secret-rotator:2026-07-30 + args: ["gitea-token"] + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + env: + - name: BAO_ADDR + value: http://openbao.openbao.svc.cluster.local:8200 + - name: BAO_ROLE + value: rotator-review-bot + - name: BOOTSTRAP_PATH + value: homelab/infrastructure/secret-rotators-bootstrap/review-bot + - name: BAO_PATH + value: homelab/infrastructure/review-bot + - name: BAO_TOKEN_FIELD + value: mcp_token + - name: GITEA_URL + value: http://gitea.gitea.svc.cluster.local:3000 + - name: GITEA_VERIFY_REPO + value: konairius/shelly-manager + - name: GITEA_SCOPES + value: read:repository,read:issue,read:user + - name: GITEA_REQUIRE_PUSH + value: "false" + - name: GITEA_TOKEN_NAME_PREFIX + value: review-bot-mcp + resources: + requests: + cpu: 25m + memory: 16Mi + limits: + memory: 48Mi containers: - name: sync-woodpecker image: gitea.renner.uno/gitea_admin/secret-rotator:2026-07-30 diff --git a/kubernetes/infrastructure/post-crds/openbao-clustersecretstore.yaml b/kubernetes/infrastructure/post-crds/openbao-clustersecretstore.yaml index 6a3daf3b..6d9ea067 100644 --- a/kubernetes/infrastructure/post-crds/openbao-clustersecretstore.yaml +++ b/kubernetes/infrastructure/post-crds/openbao-clustersecretstore.yaml @@ -31,7 +31,7 @@ spec: - monitoring - paperless - renovate # renovate-bot token (apps/renovate/externalsecret.yaml) - - review-bot # webhook secret, Mistral key, review token (apps/review-bot/externalsecret.yaml) + - review-bot # webhook secret, Mistral/Anthropic/TypeSafe keys, review + MCP tokens (apps/review-bot/externalsecret.yaml) - trmnl - uptime-monitor # uptime-robot-api-key (apps/uptime-monitor/externalsecret.yaml) - velero diff --git a/review-bot/anthropic.go b/review-bot/anthropic.go new file mode 100644 index 00000000..37cfec22 --- /dev/null +++ b/review-bot/anthropic.go @@ -0,0 +1,137 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net/http" + "slices" + "strings" + + "github.com/anthropics/anthropic-sdk-go" + "github.com/anthropics/anthropic-sdk-go/option" + "github.com/anthropics/anthropic-sdk-go/packages/param" + "github.com/anthropics/anthropic-sdk-go/shared/constant" +) + +// mcpServer is the name the Gitea MCP server has inside a request. +const mcpServer = "gitea" + +// maxContinuations bounds how often a paused turn (the server-side tool loop reached its limit) +// is resumed. +const maxContinuations = 30 + +// fallbackModels serve a refused request with Anthropic's server-side fallback model. +var fallbackModels = []string{"claude-fable-5-1", "claude-opus-5-5", "claude-opus-5", "claude-sonnet-5-5"} + +// Anthropic runs the review on the Messages API. Anthropic connects to the Gitea MCP server +// itself, with a read-only token, and only the listed read tools are enabled. +type Anthropic struct { + client anthropic.Client + mcpURL string + mcpToken string + tools []string + effort string +} + +func newAnthropic(cfg Config) *Anthropic { + opts := []option.RequestOption{option.WithAPIKey(cfg.AnthropicKey), option.WithMaxRetries(0)} + if cfg.AnthropicURL != "" { + opts = append(opts, option.WithBaseURL(cfg.AnthropicURL)) + } + return &Anthropic{client: anthropic.NewClient(opts...), mcpURL: cfg.MCPURL, mcpToken: cfg.MCPToken, + tools: cfg.MCPTools, effort: cfg.AnthropicEffort} +} + +// Prepare has nothing to set up: the prompt and tools travel with every request. +func (a *Anthropic) Prepare(context.Context, []string) error { return nil } + +func (a *Anthropic) params(model, task string) anthropic.BetaMessageNewParams { + configs := map[string]anthropic.BetaMCPToolConfigParam{} + for _, t := range a.tools { + configs[t] = anthropic.BetaMCPToolConfigParam{Enabled: param.NewOpt(true)} + } + p := anthropic.BetaMessageNewParams{ + Model: model, + MaxTokens: 64000, + System: []anthropic.BetaTextBlockParam{{Text: reviewPrompt, CacheControl: anthropic.NewBetaCacheControlEphemeralParam()}}, + Messages: []anthropic.BetaMessageParam{anthropic.NewBetaUserMessage(anthropic.NewBetaTextBlock(task))}, + MCPServers: []anthropic.BetaRequestMCPServerURLDefinitionParam{{ + Name: mcpServer, URL: a.mcpURL, AuthorizationToken: param.NewOpt(a.mcpToken), + }}, + Tools: []anthropic.BetaToolUnionParam{{OfMCPToolset: &anthropic.BetaMCPToolsetParam{ + MCPServerName: mcpServer, + DefaultConfig: anthropic.BetaMCPToolDefaultConfigParam{Enabled: param.NewOpt(false)}, + Configs: configs, + }}}, + OutputConfig: anthropic.BetaOutputConfigParam{Effort: anthropic.BetaOutputConfigEffort(a.effort)}, + Betas: []anthropic.AnthropicBeta{anthropic.AnthropicBetaMCPClient2025_11_20}, + } + if slices.Contains(fallbackModels, model) { + p.Fallbacks = anthropic.BetaFallbacksParamUnion{OfDefault: constant.ValueOf[constant.Default]()} + p.Betas = append(p.Betas, anthropic.AnthropicBetaServerSideFallback2026_07_01) + } + return p +} + +// Ask streams the review, resuming paused turns, and takes the text after the last tool call. +func (a *Anthropic) Ask(ctx context.Context, model, task string) (Answer, error) { + p := a.params(model, task) + var ans Answer + var texts []string + for turn := 0; ; turn++ { + stream := a.client.Beta.Messages.NewStreaming(ctx, p) + msg := anthropic.BetaMessage{} + for stream.Next() { + if err := msg.Accumulate(stream.Current()); err != nil { + _ = stream.Close() + return Answer{}, &streamError{Provider: "anthropic", Err: err} + } + } + if err := stream.Err(); err != nil { + return Answer{}, anthropicError(err) + } + for _, block := range msg.Content { + switch b := block.AsAny().(type) { + case anthropic.BetaMCPToolUseBlock: + ans.ToolCalls++ + if !slices.Contains(a.tools, b.Name) { + ans.Violations = append(ans.Violations, b.Name) + } + texts = nil + case anthropic.BetaMCPToolResultBlock: + texts = nil + case anthropic.BetaTextBlock: + texts = append(texts, b.Text) + } + } + switch msg.StopReason { + case anthropic.BetaStopReasonPauseTurn: + if turn >= maxContinuations { + return Answer{}, fmt.Errorf("anthropic: the turn was still paused after %d continuations", maxContinuations) + } + p.Messages = append(p.Messages, msg.ToParam()) + continue + case anthropic.BetaStopReasonRefusal: + return Answer{}, fmt.Errorf("anthropic refused the review (%s): %s", msg.StopDetails.Category, msg.StopDetails.Explanation) + case anthropic.BetaStopReasonMaxTokens: + return Answer{}, errors.New("anthropic: the answer hit max_tokens") + } + ans.Text = strings.Join(texts, "") + return ans, nil + } +} + +// anthropicError makes an API error an apiError, so the retry policy sees its status, and any +// other failure of the stream a streamError. +func anthropicError(err error) error { + var api *anthropic.Error + if !errors.As(err, &api) { + return &streamError{Provider: "anthropic", Err: err} + } + resp := api.Response + if resp == nil { + resp = &http.Response{StatusCode: api.StatusCode, Header: http.Header{}} + } + return newAPIError("anthropic", http.MethodPost, "/v1/messages", resp, []byte(api.RawJSON())) +} diff --git a/review-bot/anthropic_test.go b/review-bot/anthropic_test.go new file mode 100644 index 00000000..5db587e3 --- /dev/null +++ b/review-bot/anthropic_test.go @@ -0,0 +1,219 @@ +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 k.renner/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(), "k.renner/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"] != "k.renner/homelab" { + t.Errorf("selector state: %v", state) + } + }) + } +} diff --git a/review-bot/backend.go b/review-bot/backend.go new file mode 100644 index 00000000..64f895b4 --- /dev/null +++ b/review-bot/backend.go @@ -0,0 +1,102 @@ +package main + +import ( + "context" + _ "embed" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" +) + +//go:embed prompt.md +var reviewPrompt string + +// A Backend runs the review agent of one provider. The agent reads the repository through a +// read-only Gitea connection of that provider and answers with the report. +type Backend interface { + // Prepare makes the backend ready to review with the given models of it. It is retried until + // it succeeds. + Prepare(ctx context.Context, models []string) error + // Ask runs one review conversation with model and returns the agent's answer. + Ask(ctx context.Context, model, task string) (Answer, error) +} + +type Answer struct { + Text string + ToolCalls int + Violations []string +} + +// apiError is a non-2xx answer of a provider API. Whatever limit headers come back are kept so +// they can be logged and used. +type apiError struct { + Provider string + Method, Path string + Status int + Body string + RetryAfter time.Duration + HasRetryAfter bool + Limits map[string]string +} + +func (e *apiError) Error() string { + return fmt.Sprintf("%s %s %s: HTTP %d: %s", e.Provider, e.Method, e.Path, e.Status, e.Body) +} + +// newAPIError reads the status, the Retry-After and the limit headers of a failed response. +func newAPIError(provider, method, path string, resp *http.Response, body []byte) *apiError { + e := &apiError{Provider: provider, Method: method, Path: path, Status: resp.StatusCode, Body: snippet(body), + Limits: limitHeaders(resp.Header)} + if secs, err := strconv.Atoi(strings.TrimSpace(resp.Header.Get("Retry-After"))); err == nil && secs >= 0 { + e.RetryAfter, e.HasRetryAfter = time.Duration(secs)*time.Second, true + } + return e +} + +// limitHeaders collects the Retry-After and rate limit headers of a response. +func limitHeaders(h http.Header) map[string]string { + out := map[string]string{} + for k, v := range h { + l := strings.ToLower(k) + if l == "retry-after" || strings.Contains(l, "ratelimit") || strings.Contains(l, "rate-limit") { + out[k] = strings.Join(v, ",") + } + } + return out +} + +// streamError is a response stream that broke off or reported an error before the answer was +// complete. +type streamError struct { + Provider string + Err error +} + +func (e *streamError) Error() string { return fmt.Sprintf("%s stream: %v", e.Provider, e.Err) } +func (e *streamError) Unwrap() error { return e.Err } + +// retryable says whether err is worth asking again after: a rate limit (429), a gateway or +// overload failure (502, 503, 504, 529), or a response stream that broke off. +func retryable(err error) bool { + if api := asAPIError(err); api != nil { + switch api.Status { + case http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, + http.StatusGatewayTimeout, 529: + return true + } + return false + } + var s *streamError + return errors.As(err, &s) && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) +} + +func asAPIError(err error) *apiError { + var api *apiError + if errors.As(err, &api) { + return api + } + return nil +} diff --git a/review-bot/config.go b/review-bot/config.go index 8b8840ac..7e101f4b 100644 --- a/review-bot/config.go +++ b/review-bot/config.go @@ -8,6 +8,14 @@ import ( "time" ) +// defaultMCPTools are the read tools of the Gitea MCP server that the Anthropic backend enables. +var defaultMCPTools = []string{ + "get_commit", "get_dir_contents", "get_file_contents", "get_repository_tree", "get_tag", "get_release", + "get_latest_release", "issue_read", "label_read", "list_branches", "list_commits", "list_issues", + "list_pull_requests", "list_releases", "list_tags", "milestone_read", "pull_request_read", "search_issues", + "search_repos", +} + // Config is read from the environment. A secret comes from NAME, or from the file named by // NAME_FILE when that is set (a mounted Kubernetes Secret). type Config struct { @@ -16,20 +24,40 @@ type Config struct { BotLogin string ReviewToken string WebhookSecret string - MistralURL string - MistralKey string - AgentName string - AgentModel string - ConnectorID string + // Models the bot reviews with; the first is the default. A backend whose credentials are + // missing drops its models. + Models []Model + + MistralURL string + MistralKey string + AgentName string // prefix of the Mistral agent names, one agent per model + ConnectorID string + + AnthropicURL string + AnthropicKey string + AnthropicEffort string + MCPURL string + MCPToken string // read-only Gitea token Anthropic uses on the MCP server + MCPTools []string + + // Selector "typesafe" picks the model per pull request; empty uses the default model. + Selector string + TypeSafeURL string + TypeSafeKey string + TypeSafeModel string + Repos []string Workers int QueueSize int Sweep time.Duration ReviewTimeout time.Duration ShutdownWait time.Duration - // A review that runs into Mistral's connector rate limit is asked again after a wait. + // A review that runs into a rate limit or a gateway failure is asked again after a wait. RateLimitRetries int RateLimitWait time.Duration + + // Warnings are configuration problems the bot works around. + Warnings []string } func loadConfig(getenv func(string) string, readFile func(string) ([]byte, error)) (Config, error) { @@ -40,7 +68,7 @@ func loadConfig(getenv func(string) string, readFile func(string) ([]byte, error } return def } - secret := func(name string) string { + optional := func(name string) string { if path := strings.TrimSpace(getenv(name + "_FILE")); path != "" { b, err := readFile(path) if err != nil { @@ -49,8 +77,11 @@ func loadConfig(getenv func(string) string, readFile func(string) ([]byte, error } return strings.TrimSpace(string(b)) } - v := strings.TrimSpace(getenv(name)) - if v == "" { + return strings.TrimSpace(getenv(name)) + } + secret := func(name string) string { + v := optional(name) + if v == "" && strings.TrimSpace(getenv(name+"_FILE")) == "" { errs = append(errs, fmt.Errorf("%s is not set", name)) } return v @@ -79,6 +110,13 @@ func loadConfig(getenv func(string) string, readFile func(string) ([]byte, error } return d } + list := func(name string, def []string) []string { + v := text(name, "") + if v == "" { + return def + } + return strings.FieldsFunc(v, func(r rune) bool { return r == ',' || r == ' ' || r == '\n' }) + } c := Config{ Listen: text("LISTEN", ":8080"), @@ -86,11 +124,24 @@ func loadConfig(getenv func(string) string, readFile func(string) ([]byte, error BotLogin: text("BOT_LOGIN", "review-bot"), ReviewToken: secret("REVIEW_TOKEN"), WebhookSecret: secret("WEBHOOK_SECRET"), - MistralURL: strings.TrimRight(text("MISTRAL_URL", "https://api.mistral.ai/v1"), "/"), - MistralKey: secret("MISTRAL_API_KEY"), - AgentName: text("AGENT_NAME", "pr-review"), - AgentModel: text("AGENT_MODEL", "mistral-large-4-0"), - ConnectorID: text("CONNECTOR_ID", ""), + + MistralURL: strings.TrimRight(text("MISTRAL_URL", "https://api.mistral.ai/v1"), "/"), + MistralKey: optional("MISTRAL_API_KEY"), + AgentName: text("AGENT_NAME", "pr-review"), + ConnectorID: text("CONNECTOR_ID", ""), + + AnthropicURL: strings.TrimRight(text("ANTHROPIC_URL", ""), "/"), + AnthropicKey: optional("ANTHROPIC_API_KEY"), + AnthropicEffort: text("ANTHROPIC_EFFORT", "high"), + MCPURL: text("MCP_URL", "https://gitea-mcp.renner.uno/mcp"), + MCPToken: optional("MCP_TOKEN"), + MCPTools: list("MCP_TOOLS", defaultMCPTools), + + Selector: text("SELECTOR", ""), + TypeSafeURL: text("TYPESAFE_URL", "https://api.typesafe.ai/v1/systemone"), + TypeSafeKey: optional("TYPESAFE_API_KEY"), + TypeSafeModel: text("TYPESAFE_MODEL", "jev-latest"), + Workers: number("WORKERS", 1), QueueSize: number("QUEUE_SIZE", 32), Sweep: duration("SWEEP_INTERVAL", 30*time.Minute), @@ -100,10 +151,45 @@ func loadConfig(getenv func(string) string, readFile func(string) ([]byte, error RateLimitRetries: number("RATE_LIMIT_RETRIES", 3), RateLimitWait: duration("RATE_LIMIT_WAIT", time.Minute), } - if c.ConnectorID == "" { - errs = append(errs, errors.New("CONNECTOR_ID is not set")) + + models, err := parseModels(text("MODELS", "mistral:mistral-large-4-0")) + if err != nil { + errs = append(errs, err) } - for _, r := range strings.FieldsFunc(text("REPOS", ""), func(r rune) bool { return r == ',' || r == ' ' || r == '\n' }) { + missing := map[string]string{} + if c.MistralKey == "" { + missing["mistral"] = "MISTRAL_API_KEY is not set" + } else if c.ConnectorID == "" { + missing["mistral"] = "CONNECTOR_ID is not set" + } + if c.AnthropicKey == "" { + missing["anthropic"] = "ANTHROPIC_API_KEY is not set" + } else if c.MCPToken == "" { + missing["anthropic"] = "MCP_TOKEN is not set" + } + for _, m := range models { + if why, ok := missing[m.Backend]; ok { + c.Warnings = append(c.Warnings, fmt.Sprintf("%s is off: %s", m.ID(), why)) + continue + } + c.Models = append(c.Models, m) + } + if err == nil && len(c.Models) == 0 { + errs = append(errs, errors.New("no model of MODELS is usable: "+strings.Join(c.Warnings, "; "))) + } + + switch c.Selector { + case "": + case "typesafe": + if c.TypeSafeKey == "" { + c.Warnings = append(c.Warnings, "the typesafe selector is off: TYPESAFE_API_KEY is not set") + c.Selector = "" + } + default: + errs = append(errs, fmt.Errorf("SELECTOR must be empty or typesafe, got %q", c.Selector)) + } + + for _, r := range list("REPOS", nil) { if strings.Count(r, "/") != 1 { errs = append(errs, fmt.Errorf("REPOS entry %q is not owner/repo", r)) continue diff --git a/review-bot/config_test.go b/review-bot/config_test.go index 20a1b2ee..e8bdae5a 100644 --- a/review-bot/config_test.go +++ b/review-bot/config_test.go @@ -24,10 +24,13 @@ func TestLoadConfigDefaults(t *testing.T) { t.Fatal(err) } if c.Listen != ":8080" || c.GiteaURL != "https://gitea.renner.uno" || c.BotLogin != "review-bot" || - c.AgentModel != "mistral-large-4-0" || c.Workers != 1 || c.Sweep != 30*time.Minute || - c.RateLimitRetries != 3 || c.RateLimitWait != time.Minute { + c.Workers != 1 || c.Sweep != 30*time.Minute || c.RateLimitRetries != 3 || c.RateLimitWait != time.Minute || + c.Selector != "" || c.MCPURL != "https://gitea-mcp.renner.uno/mcp" || c.AnthropicEffort != "high" { t.Errorf("unexpected defaults: %+v", c) } + if len(c.Models) != 1 || c.Models[0].ID() != "mistral:mistral-large-4-0" { + t.Errorf("models: %+v", c.Models) + } if len(c.Repos) != 2 || c.Repos[1] != "konairius/shelly-manager" { t.Errorf("repos: %v", c.Repos) } @@ -44,14 +47,53 @@ func TestLoadConfigSecretFromFile(t *testing.T) { } func TestLoadConfigErrors(t *testing.T) { - _, err := loadConfig(env(map[string]string{"WORKERS": "0", "REPOS": "nonsense", "SWEEP_INTERVAL": "soon"}), noFiles) + _, err := loadConfig(env(map[string]string{"WORKERS": "0", "REPOS": "nonsense", "SWEEP_INTERVAL": "soon", + "SELECTOR": "dice"}), noFiles) if err == nil { t.Fatal("expected errors") } - for _, want := range []string{"REVIEW_TOKEN is not set", "WEBHOOK_SECRET is not set", "MISTRAL_API_KEY is not set", - "CONNECTOR_ID is not set", "WORKERS must be", "SWEEP_INTERVAL must be", `REPOS entry "nonsense"`} { + for _, want := range []string{"REVIEW_TOKEN is not set", "WEBHOOK_SECRET is not set", "no model of MODELS is usable", + "MISTRAL_API_KEY is not set", "WORKERS must be", "SWEEP_INTERVAL must be", `REPOS entry "nonsense"`, "SELECTOR must be"} { if !strings.Contains(err.Error(), want) { t.Errorf("missing %q in: %v", want, err) } } } + +// A backend without its credentials drops its models with a warning instead of stopping the bot. +func TestLoadConfigDropsBackendsWithoutCredentials(t *testing.T) { + e := baseEnv() + e["MODELS"] = "anthropic:claude-opus-5-5 = strong\nmistral:mistral-large-4-0\n# comment\n\nanthropic:claude-sonnet-5-5" + e["SELECTOR"] = "typesafe" + c, err := loadConfig(env(e), noFiles) + if err != nil { + t.Fatal(err) + } + if len(c.Models) != 1 || c.Models[0].ID() != "mistral:mistral-large-4-0" { + t.Errorf("models: %+v", c.Models) + } + if c.Selector != "" || len(c.Warnings) != 3 || !strings.Contains(strings.Join(c.Warnings, "\n"), "ANTHROPIC_API_KEY is not set") { + t.Errorf("selector %q, warnings %v", c.Selector, c.Warnings) + } + + e["ANTHROPIC_API_KEY"], e["TYPESAFE_API_KEY"] = "a", "ts" + c, _ = loadConfig(env(e), noFiles) + if len(c.Models) != 1 || !strings.Contains(strings.Join(c.Warnings, "\n"), "MCP_TOKEN is not set") { + t.Errorf("anthropic without MCP_TOKEN must stay off: %+v %v", c.Models, c.Warnings) + } + + e["MCP_TOKEN"] = "m" + c, err = loadConfig(env(e), noFiles) + if err != nil || len(c.Models) != 3 || c.Models[0].ID() != "anthropic:claude-opus-5-5" || + c.Models[0].Description != "strong" || c.Selector != "typesafe" || len(c.Warnings) != 0 { + t.Errorf("models %+v, selector %q, warnings %v, err %v", c.Models, c.Selector, c.Warnings, err) + } +} + +func TestParseModelsErrors(t *testing.T) { + for _, bad := range []string{"claude-opus-5-5", "openai:gpt = x", "mistral:", "mistral:a\nmistral:a"} { + if _, err := parseModels(bad); err == nil { + t.Errorf("%q: want an error", bad) + } + } +} diff --git a/review-bot/gitea.go b/review-bot/gitea.go index f00d08a7..5c5fc686 100644 --- a/review-bot/gitea.go +++ b/review-bot/gitea.go @@ -16,6 +16,7 @@ type PR struct { State string `json:"state"` Draft bool `json:"draft"` Title string `json:"title"` + Body string `json:"body"` Additions int `json:"additions"` Deletions int `json:"deletions"` ChangedFiles int `json:"changed_files"` @@ -93,6 +94,13 @@ func (g *Gitea) Reviews(ctx context.Context, slug string, n int) ([]Review, erro return rv, err } +// Files lists the changed files of a pull request (the first 100). +func (g *Gitea) Files(ctx context.Context, slug string, n int) ([]PRFile, error) { + var files []PRFile + err := g.do(ctx, http.MethodGet, fmt.Sprintf("/repos/%s/pulls/%d/files?limit=100", slug, n), nil, &files) + return files, err +} + // PostReview posts the review and returns its web address. func (g *Gitea) PostReview(ctx context.Context, slug string, n int, body, event, commit string) (string, error) { in := map[string]string{"body": body, "event": event, "commit_id": commit} diff --git a/review-bot/go.mod b/review-bot/go.mod index 7a4c8e05..8ddf2850 100644 --- a/review-bot/go.mod +++ b/review-bot/go.mod @@ -2,6 +2,28 @@ module gitea.renner.uno/k.renner/homelab/review-bot go 1.27 -require github.com/failsafe-go/failsafe-go v0.9.8 +require ( + github.com/anthropics/anthropic-sdk-go v1.82.0 + github.com/failsafe-go/failsafe-go v0.9.8 +) -require github.com/bits-and-blooms/bitset v1.24.4 // indirect +require ( + github.com/bahlo/generic-list-go v0.2.0 // indirect + github.com/bits-and-blooms/bitset v1.24.4 // indirect + github.com/buger/jsonparser v1.1.2 // indirect + github.com/go-logr/logr v1.4.2 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/invopop/jsonschema v0.14.0 // indirect + github.com/pb33f/ordered-map/v2 v2.3.1 // indirect + github.com/standard-webhooks/standard-webhooks/libraries v0.0.1 // indirect + github.com/tidwall/gjson v1.18.0 // indirect + github.com/tidwall/match v1.1.1 // indirect + github.com/tidwall/pretty v1.2.1 // indirect + github.com/tidwall/sjson v1.2.5 // indirect + go.opentelemetry.io/auto/sdk v1.1.0 // indirect + go.opentelemetry.io/otel v1.33.0 // indirect + go.opentelemetry.io/otel/metric v1.33.0 // indirect + go.opentelemetry.io/otel/trace v1.33.0 // indirect + go.yaml.in/yaml/v4 v4.0.0-rc.2 // indirect + golang.org/x/sync v0.16.0 // indirect +) diff --git a/review-bot/go.sum b/review-bot/go.sum index 5c68a2d8..b8db9447 100644 --- a/review-bot/go.sum +++ b/review-bot/go.sum @@ -1,26 +1,143 @@ +cloud.google.com/go/auth v0.7.2 h1:uiha352VrCDMXg+yoBtaD0tUF4Kv9vrtrWPYXwutnDE= +cloud.google.com/go/auth v0.7.2/go.mod h1:VEc4p5NNxycWQTMQEDQF0bd6aTMb6VgYDXEwiJJQAbs= +cloud.google.com/go/auth/oauth2adapt v0.2.3 h1:MlxF+Pd3OmSudg/b1yZ5lJwoXCEaeedAguodky1PcKI= +cloud.google.com/go/auth/oauth2adapt v0.2.3/go.mod h1:tMQXOfZzFuNuUxOypHlQEXgdfX5cuhwU+ffUuXRJE8I= +cloud.google.com/go/compute/metadata v0.5.0 h1:Zr0eK8JbFv6+Wi4ilXAR8FJ3wyNdpxHKJNPos6LTZOY= +cloud.google.com/go/compute/metadata v0.5.0/go.mod h1:aHnloV2TPI38yx4s9+wAZhHykWvVCfu7hQbF+9CWoiY= +github.com/anthropics/anthropic-sdk-go v1.82.0 h1:A82J+yHEMbQ3+7ObCagOX4tVm1uyBhELCHd2dDYZYuo= +github.com/anthropics/anthropic-sdk-go v1.82.0/go.mod h1:GThfYqPJoaQ/6pmibCI98Cr4y5su2FXMUHn3NrSSnIc= +github.com/aws/aws-sdk-go-v2 v1.38.0 h1:UCRQ5mlqcFk9HJDIqENSLR3wiG1VTWlyUfLDEvY7RxU= +github.com/aws/aws-sdk-go-v2 v1.38.0/go.mod h1:9Q0OoGQoboYIAJyslFyF1f5K1Ryddop8gqMhWx/n4Wg= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 h1:tW1/Rkad38LA15X4UQtjXZXNKsCgkshC3EbmcUmghTg= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3/go.mod h1:UbnqO+zjqk3uIt9yCACHJ9IVNhyhOCnYk8yA19SAWrM= +github.com/aws/aws-sdk-go-v2/config v1.27.27 h1:HdqgGt1OAP0HkEDDShEl0oSYa9ZZBSOmKpdpsDMdO90= +github.com/aws/aws-sdk-go-v2/config v1.27.27/go.mod h1:MVYamCg76dFNINkZFu4n4RjDixhVr51HLj4ErWzrVwg= +github.com/aws/aws-sdk-go-v2/credentials v1.17.27 h1:2raNba6gr2IfA0eqqiP2XiQ0UVOpGPgDSi0I9iAP+UI= +github.com/aws/aws-sdk-go-v2/credentials v1.17.27/go.mod h1:gniiwbGahQByxan6YjQUMcW4Aov6bLC3m+evgcoN4r4= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11 h1:KreluoV8FZDEtI6Co2xuNk/UqI9iwMrOx/87PBNIKqw= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11/go.mod h1:SeSUYBLsMYFoRvHE0Tjvn7kbxaUhl75CJi1sbfhMxkU= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15 h1:SoNJ4RlFEQEbtDcCEt+QG56MY4fm4W8rYirAmq+/DdU= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15/go.mod h1:U9ke74k1n2bf+RIgoX1SXFed1HLs51OgUSs+Ph0KJP8= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15 h1:C6WHdGnTDIYETAm5iErQUiVNsclNx9qbJVPIt03B6bI= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15/go.mod h1:ZQLZqhcu+JhSrA9/NXRm8SkDvsycE+JkV3WGY41e+IM= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 h1:hT8rVHwugYE2lEfdFE0QWVo81lF7jMrYJVDWI+f+VxU= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0/go.mod h1:8tu/lYfQfFe6IGnaOdrpVgEL2IrrDOf6/m9RQum4NkY= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3 h1:dT3MqvGhSoaIhRseqw2I0yH81l7wiR2vjs57O51EAm8= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3/go.mod h1:GlAeCkHwugxdHaueRr4nhPuY+WW+gR8UjlcqzPr1SPI= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17 h1:HGErhhrxZlQ044RiM+WdoZxp0p+EGM62y3L6pwA4olE= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17/go.mod h1:RkZEx4l0EHYDJpWppMJ3nD9wZJAa8/0lq9aVC+r2UII= +github.com/aws/aws-sdk-go-v2/service/sso v1.22.4 h1:BXx0ZIxvrJdSgSvKTZ+yRBeSqqgPM89VPlulEcl37tM= +github.com/aws/aws-sdk-go-v2/service/sso v1.22.4/go.mod h1:ooyCOXjvJEsUw7x+ZDHeISPMhtwI3ZCB7ggFMcFfWLU= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4 h1:yiwVzJW2ZxZTurVbYWA7QOrAaCYQR72t0wrSBfoesUE= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4/go.mod h1:0oxfLkpz3rQ/CHlx5hB7H69YUpFiI1tql6Q6Ne+1bCw= +github.com/aws/aws-sdk-go-v2/service/sts v1.30.3 h1:ZsDKRLXGWHk8WdtyYMoGNO7bTudrvuKpDKgMVRlepGE= +github.com/aws/aws-sdk-go-v2/service/sts v1.30.3/go.mod h1:zwySh8fpFyXp9yOr/KVzxOl8SRqgf/IDw5aUt9UKFcQ= +github.com/aws/smithy-go v1.22.5 h1:P9ATCXPMb2mPjYBgueqJNCA5S9UfktsW0tTxi+a7eqw= +github.com/aws/smithy-go v1.22.5/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI= +github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk= +github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg= github.com/bits-and-blooms/bitset v1.24.4 h1:95H15Og1clikBrKr/DuzMXkQzECs1M6hhoGXLwLQOZE= github.com/bits-and-blooms/bitset v1.24.4/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= +github.com/buger/jsonparser v1.1.2 h1:frqHqw7otoVbk5M8LlE/L7HTnIq2v9RX6EJ48i9AxJk= +github.com/buger/jsonparser v1.1.2/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dnaeon/go-vcr v1.2.0 h1:zHCHvJYTMh1N7xnV7zf1m1GPBF9Ad0Jk/whtQ1663qI= +github.com/dnaeon/go-vcr v1.2.0/go.mod h1:R4UdLID7HZT3taECzJs4YgbbH6PIGXB6W/sc5OLb6RQ= github.com/failsafe-go/failsafe-go v0.9.8 h1:NzahTEc+vg6FqCJV0Sy9mgwcaCemB81c8iSHWTF/urU= github.com/failsafe-go/failsafe-go v0.9.8/go.mod h1:wMKUFRbxxjWwvwhJttSiim9BaQssrlZV4lk0IwPkwUc= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= +github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE= +github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/jsonschema-go v0.4.2 h1:tmrUohrwoLZZS/P3x7ex0WAVknEkBZM46iALbcqoRA8= +github.com/google/jsonschema-go v0.4.2/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= +github.com/google/s2a-go v0.1.7 h1:60BLSyTrOV4/haCDW4zb1guZItoSq8foHCXrAnjBo/o= +github.com/google/s2a-go v0.1.7/go.mod h1:50CgR4k1jNlWBu4UfS4AcfhVe1r6pdZPygJ3R8F0Qdw= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/googleapis/enterprise-certificate-proxy v0.3.2 h1:Vie5ybvEvT75RniqhfFxPRy3Bf7vr3h0cechB90XaQs= +github.com/googleapis/enterprise-certificate-proxy v0.3.2/go.mod h1:VLSiSSBs/ksPL8kq3OBOQ6WRI2QnaFynd1DCjZ62+V0= github.com/influxdata/tdigest v0.0.1 h1:XpFptwYmnEKUqmkcDjrzffswZ3nvNeevbUSLPP/ZzIY= github.com/influxdata/tdigest v0.0.1/go.mod h1:Z0kXnxzbTC2qrx4NaIzYkE1k66+6oEDQTvL95hQFh5Y= +github.com/invopop/jsonschema v0.14.0 h1:MHQqLhvpNUZfw+hM3AZDYK7jxO8FZoQeQM77g8iyZjg= +github.com/invopop/jsonschema v0.14.0/go.mod h1:ygm6C2EaVNMBDPpaPlnOA2pFAxBnxGjFlMZABxm9n2I= +github.com/modelcontextprotocol/go-sdk v1.3.1 h1:TfqtNKOIWN4Z1oqmPAiWDC2Jq7K9OdJaooe0teoXASI= +github.com/modelcontextprotocol/go-sdk v1.3.1/go.mod h1:DgVX498dMD8UJlseK1S5i1T4tFz2fkBk4xogC3D15nw= +github.com/pb33f/ordered-map/v2 v2.3.1 h1:5319HDO0aw4DA4gzi+zv4FXU9UlSs3xGZ40wcP1nBjY= +github.com/pb33f/ordered-map/v2 v2.3.1/go.mod h1:qxFQgd0PkVUtOMCkTapqotNgzRhMPL7VvaHKbd1HnmQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= +github.com/standard-webhooks/standard-webhooks/libraries v0.0.1 h1:uOfcYT+3QungH6tIGSVCR/Y3KJmgJiHcojJbMTPDZAI= +github.com/standard-webhooks/standard-webhooks/libraries v0.0.1/go.mod h1:L1MQhA6x4dn9r007T033lsaZMv9EmBAdXyU/+EF40fo= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -golang.org/x/net v0.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE= -golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg= -golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg= -golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc= -golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= +github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY= +github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA= +github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= +github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= +github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY= +github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28= +github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= +github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= +go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= +go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= +go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= +go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 h1:4Pp6oUg3+e/6M4C0A/3kJ2VYa++dsWVTtGgLVj5xtHg= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0/go.mod h1:Mjt1i1INqiaoZOMGR1RIUJN+i3ChKoFRqzrRQhlkbs0= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 h1:jq9TW8u3so/bN+JPT166wjOI6/vQPF6Xe7nMNIltagk= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0/go.mod h1:p8pYQP+m5XfbZm9fxtSKAbM6oIllS7s2AfxrChvc7iw= +go.opentelemetry.io/otel v1.33.0 h1:/FerN9bax5LoK51X/sI0SVYrjSE0/yUL7DpxW4K3FWw= +go.opentelemetry.io/otel v1.33.0/go.mod h1:SUUkR6csvUQl+yjReHu5uM3EtVV7MBm5FHKRlNx4I8I= +go.opentelemetry.io/otel/metric v1.33.0 h1:r+JOocAyeRVXD8lZpjdQjzMadVZp2M4WmQ+5WtEnklQ= +go.opentelemetry.io/otel/metric v1.33.0/go.mod h1:L9+Fyctbp6HFTddIxClbQkjtubW6O9QS3Ann/M82u6M= +go.opentelemetry.io/otel/sdk v1.33.0 h1:iax7M131HuAm9QkZotNHEfstof92xM+N8sr3uHXc2IM= +go.opentelemetry.io/otel/sdk v1.33.0/go.mod h1:A1Q5oi7/9XaMlIWzPSxLRWOI8nG3FnzHJNbiENQuihM= +go.opentelemetry.io/otel/trace v1.33.0 h1:cCJuF7LRjUFso9LPnEAHJDB2pqzp+hbO8eu1qqW2d/s= +go.opentelemetry.io/otel/trace v1.33.0/go.mod h1:uIcdVUZMpTAmz0tI1z04GoVSezK37CbGV4fr1f2nBck= +go.yaml.in/yaml/v4 v4.0.0-rc.2 h1:/FrI8D64VSr4HtGIlUtlFMGsm7H7pWTbj6vOLVZcA6s= +go.yaml.in/yaml/v4 v4.0.0-rc.2/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= +golang.org/x/crypto v0.40.0 h1:r4x+VvoG5Fm+eJcxMaY8CQM7Lb0l1lsmjGBQ6s8BfKM= +golang.org/x/crypto v0.40.0/go.mod h1:Qr1vMER5WyS2dfPHAlsOj01wgLbsyWtFn/aY+5+ZdxY= +golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw= +golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA= +golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI= +golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU= +golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw= +golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4= +golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU= +golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk= +golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= +google.golang.org/api v0.189.0 h1:equMo30LypAkdkLMBqfeIqtyAnlyig1JSZArl4XPwdI= +google.golang.org/api v0.189.0/go.mod h1:FLWGJKb0hb+pU2j+rJqwbnsF+ym+fQs73rbJ+KAUgy8= google.golang.org/genproto/googleapis/rpc v0.0.0-20240814211410-ddb44dafa142 h1:e7S5W7MGGLaSu8j3YjdezkZ+m1/Nm0uRVRMEMGk26Xs= google.golang.org/genproto/googleapis/rpc v0.0.0-20240814211410-ddb44dafa142/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU= google.golang.org/grpc v1.67.1 h1:zWnc1Vrcno+lHZCOofnIMvycFcc0QRGIzm9dhnDX68E= google.golang.org/grpc v1.67.1/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA= google.golang.org/protobuf v1.36.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM= google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= +gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/review-bot/main.go b/review-bot/main.go index bd2ac7aa..2554e551 100644 --- a/review-bot/main.go +++ b/review-bot/main.go @@ -1,7 +1,8 @@ -// review-bot reviews pull requests with a Mistral agent. Gitea's pull_request webhook queues the -// review at once; a periodic sweep over the configured repositories catches deliveries that were -// lost. The result is posted as the review bot. Nothing is stored: the state is the marker in the -// bot's own review. +// review-bot reviews pull requests with an agent that reads the repository: a Mistral agent with +// the Gitea connector, or Claude with the Gitea MCP server; TypeSafe can pick the model per pull +// request. Gitea's pull_request webhook queues the review at once; a periodic sweep over the +// configured repositories catches deliveries that were lost. The result is posted as the review +// bot. Nothing is stored: the state is the marker in the bot's own review. package main import ( @@ -35,18 +36,34 @@ func run(ctx context.Context, cfg Config, log *slog.Logger) error { gitea := &Gitea{base: cfg.GiteaURL, token: cfg.ReviewToken, http: &http.Client{Timeout: time.Minute}} status := newTracker(gitea, log, metrics) go status.Run(ctx) + for _, w := range cfg.Warnings { + log.Warn("configuration", "warning", w) + } reviewer := &Reviewer{ - cfg: cfg, - gitea: gitea, - mistral: &Mistral{base: cfg.MistralURL, key: cfg.MistralKey, http: &http.Client{Timeout: cfg.ReviewTimeout + time.Minute}}, + cfg: cfg, + gitea: gitea, + // The conversations stream; the review timeout bounds them, not the client. + backends: map[string]Backend{ + "mistral": &Mistral{base: cfg.MistralURL, key: cfg.MistralKey, agentPrefix: cfg.AgentName, + connectorID: cfg.ConnectorID, http: &http.Client{}, recoverTries: 3, recoverWait: 30 * time.Second}, + "anthropic": newAnthropic(cfg), + }, log: log, metrics: metrics, status: status, } + if cfg.Selector == "typesafe" { + reviewer.selector = &TypeSafe{url: cfg.TypeSafeURL, key: cfg.TypeSafeKey, model: cfg.TypeSafeModel, + http: &http.Client{Timeout: time.Minute}} + } queue := newQueue(reviewer, cfg.QueueSize, log, metrics) queue.Start(ctx, cfg.Workers) - go syncAgent(ctx, cfg, reviewer, log) + for _, b := range backends { + if models := modelsOf(cfg.Models, b); len(models) > 0 { + go prepare(ctx, b, models, reviewer, log) + } + } go sweepLoop(ctx, cfg, reviewer, queue) srv := &http.Server{ @@ -75,18 +92,18 @@ func run(ctx context.Context, cfg Config, log *slog.Logger) error { return nil } -// syncAgent makes the listed Mistral agent match this binary, and retries until it works. -func syncAgent(ctx context.Context, cfg Config, r *Reviewer, log *slog.Logger) { +// prepare readies one backend for its models, and retries until it works. +func prepare(ctx context.Context, backend string, models []string, r *Reviewer, log *slog.Logger) { for wait := 10 * time.Second; ; wait = min(wait*2, 5*time.Minute) { attempt, cancel := context.WithTimeout(ctx, time.Minute) - agent, err := r.mistral.EnsureAgent(attempt, cfg.AgentName, cfg.AgentModel, cfg.ConnectorID, reviewPrompt) + err := r.backends[backend].Prepare(attempt, models) cancel() if err == nil { - r.setAgent(agent) - log.Info("agent ready", "agent", agent.ID, "model", agent.Model) + r.setReady(backend, true) + log.Info("backend ready", "backend", backend, "models", models) return } - log.Error("cannot sync the agent", "err", err, "retry_in", wait.String()) + log.Error("cannot prepare the backend", "backend", backend, "err", err, "retry_in", wait.String()) select { case <-ctx.Done(): return diff --git a/review-bot/mistral.go b/review-bot/mistral.go index a9802da8..1251071a 100644 --- a/review-bot/mistral.go +++ b/review-bot/mistral.go @@ -1,28 +1,38 @@ package main import ( + "bufio" "bytes" "context" "crypto/sha256" - _ "embed" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "net/http" "regexp" - "strconv" "strings" + "sync" "time" ) -//go:embed prompt.md -var reviewPrompt string - +// Mistral runs the review as a Mistral agent (one per model) with the read-only Gitea connector. +// The conversation is streamed: a request that waits silently for the whole review is cut off +// by Mistral's edge after ten minutes. type Mistral struct { - base string - key string - http *http.Client + base string + key string + agentPrefix string + connectorID string + http *http.Client + // recoverTries and recoverWait bound how often the history of a broken-off conversation is + // read to find its answer. + recoverTries int + recoverWait time.Duration + + mu sync.RWMutex + agents map[string]string // model -> agent id } type Agent struct { @@ -32,23 +42,31 @@ type Agent struct { Description string `json:"description"` } -func (m *Mistral) do(ctx context.Context, method, path string, in, out any) error { +func (m *Mistral) request(ctx context.Context, method, path string, in any) (*http.Request, error) { var body io.Reader if in != nil { b, err := json.Marshal(in) if err != nil { - return err + return nil, err } body = bytes.NewReader(b) } req, err := http.NewRequestWithContext(ctx, method, m.base+path, body) if err != nil { - return err + return nil, err } req.Header.Set("Authorization", "Bearer "+m.key) if in != nil { req.Header.Set("Content-Type", "application/json") } + return req, nil +} + +func (m *Mistral) do(ctx context.Context, method, path string, in, out any) error { + req, err := m.request(ctx, method, path, in) + if err != nil { + return err + } resp, err := m.http.Do(req) if err != nil { return err @@ -59,11 +77,7 @@ func (m *Mistral) do(ctx context.Context, method, path string, in, out any) erro return err } if resp.StatusCode != http.StatusOK { - e := &apiError{Method: method, Path: path, Status: resp.StatusCode, Body: snippet(data), Limits: limitHeaders(resp.Header)} - if secs, err := strconv.Atoi(strings.TrimSpace(resp.Header.Get("Retry-After"))); err == nil && secs >= 0 { - e.RetryAfter, e.HasRetryAfter = time.Duration(secs)*time.Second, true - } - return e + return newAPIError("mistral", method, path, resp, data) } if out == nil { return nil @@ -71,38 +85,36 @@ func (m *Mistral) do(ctx context.Context, method, path string, in, out any) erro return json.Unmarshal(data, out) } -// apiError is a non-200 answer of the Mistral API. Mistral documents X-RateLimit-Remaining and -// no Retry-After; whatever limit headers come back are kept so they can be logged and used. -type apiError struct { - Method, Path string - Status int - Body string - RetryAfter time.Duration - HasRetryAfter bool - Limits map[string]string -} - -// limitHeaders collects the Retry-After and rate limit headers of a response. -func limitHeaders(h http.Header) map[string]string { - out := map[string]string{} - for k, v := range h { - l := strings.ToLower(k) - if l == "retry-after" || strings.Contains(l, "ratelimit") || strings.Contains(l, "rate-limit") { - out[k] = strings.Join(v, ",") - } - } - return out -} - -func (e *apiError) Error() string { - return fmt.Sprintf("mistral %s %s: HTTP %d: %s", e.Method, e.Path, e.Status, e.Body) -} - func promptHash(prompt string) string { sum := sha256.Sum256([]byte(prompt)) return hex.EncodeToString(sum[:])[:12] } +func (m *Mistral) agentName(model string) string { return m.agentPrefix + "-" + model } + +// Prepare makes one agent per model match this binary. +func (m *Mistral) Prepare(ctx context.Context, models []string) error { + for _, model := range models { + a, err := m.EnsureAgent(ctx, m.agentName(model), model, m.connectorID, reviewPrompt) + if err != nil { + return fmt.Errorf("agent for %s: %w", model, err) + } + m.mu.Lock() + if m.agents == nil { + m.agents = map[string]string{} + } + m.agents[model] = a.ID + m.mu.Unlock() + } + return nil +} + +func (m *Mistral) agent(model string) string { + m.mu.RLock() + defer m.mu.RUnlock() + return m.agents[model] +} + // EnsureAgent makes the listed agent match this binary: it creates the agent when it does not // exist and updates it when its model or the hash of its prompt differs, so git stays the source // of truth for the prompt. @@ -145,69 +157,217 @@ func (m *Mistral) EnsureAgent(ctx context.Context, name, model, connectorID, pro return created, nil } -type output struct { +// A tool is read-only when its name contains _get_, _list_ or _search_, or ends in _read. +var readOnlyTool = regexp.MustCompile(`_(get|list|search)_|_read$`) + +// answerBuilder collects the answer from conversation outputs: the text of message outputs +// (thinking is dropped), and the tool calls, listing every one that is not read-only. +type answerBuilder struct { + a Answer + texts []string + lastID string +} + +func (b *answerBuilder) tool(name string) { + b.a.ToolCalls++ + if !readOnlyTool.MatchString(name) { + b.a.Violations = append(b.a.Violations, name) + } +} + +// text adds content (a string, a chunk or a list of chunks) of message id. Content of the same +// message is joined directly, different messages are joined by a newline. +func (b *answerBuilder) text(id string, content json.RawMessage) { + var parts []string + var s string + if json.Unmarshal(content, &s) == nil { + parts = append(parts, s) + } else { + type chunk struct { + Type string `json:"type"` + Text string `json:"text"` + } + var chunks []chunk + if json.Unmarshal(content, &chunks) != nil { + var one chunk + if json.Unmarshal(content, &one) == nil { + chunks = []chunk{one} + } + } + for _, ch := range chunks { + if ch.Type == "text" { + parts = append(parts, ch.Text) + } + } + } + if len(parts) == 0 { + return + } + joined := strings.Join(parts, "") + if id != "" && id == b.lastID && len(b.texts) > 0 { + b.texts[len(b.texts)-1] += joined + return + } + b.texts = append(b.texts, joined) + b.lastID = id +} + +func (b *answerBuilder) answer() Answer { + b.a.Text = strings.Join(b.texts, "\n") + return b.a +} + +type entry struct { Type string `json:"type"` + ID string `json:"id"` Name string `json:"name"` Content json.RawMessage `json:"content"` } -type conversation struct { - Outputs []output `json:"outputs"` -} - -// A tool is read-only when its name contains _get_, _list_ or _search_, or ends in _read. -var readOnlyTool = regexp.MustCompile(`_(get|list|search)_|_read$`) - -type Answer struct { - Text string - ToolCalls int - Violations []string -} - -// parseAnswer extracts the answer text (the text blocks of message outputs; thinking blocks are -// dropped) and the tool calls, and lists every tool call that is not read-only. +// parseAnswer reads the answer from a conversation response (`outputs`) or history (`entries`). func parseAnswer(raw []byte) (Answer, error) { - var c conversation + var c struct { + Outputs []entry `json:"outputs"` + Entries []entry `json:"entries"` + } if err := json.Unmarshal(raw, &c); err != nil { return Answer{}, err } - var a Answer - var texts []string - for _, o := range c.Outputs { + var b answerBuilder + for _, o := range append(c.Outputs, c.Entries...) { switch o.Type { case "tool.execution": - a.ToolCalls++ - if !readOnlyTool.MatchString(o.Name) { - a.Violations = append(a.Violations, o.Name) - } + b.tool(o.Name) case "message.output": - var s string - if json.Unmarshal(o.Content, &s) == nil { - texts = append(texts, s) - continue + // Every entry is its own message. + b.lastID = "" + b.text("", o.Content) + } + } + return b.answer(), nil +} + +type event struct { + Type string `json:"type"` + ConversationID string `json:"conversation_id"` + ID string `json:"id"` + Name string `json:"name"` + Content json.RawMessage `json:"content"` + Message string `json:"message"` + Code int `json:"code"` +} + +// Ask streams one conversation with the model's agent. When the stream breaks off after the +// conversation started, the answer is looked up in the stored history before giving up. +func (m *Mistral) Ask(ctx context.Context, model, task string) (Answer, error) { + agentID := m.agent(model) + if agentID == "" { + return Answer{}, fmt.Errorf("no Mistral agent for %s yet", model) + } + in := map[string]any{"agent_id": agentID, "inputs": task, "store": true, "stream": true} + req, err := m.request(ctx, http.MethodPost, "/conversations", in) + if err != nil { + return Answer{}, err + } + req.Header.Set("Accept", "text/event-stream") + resp, err := m.http.Do(req) + if err != nil { + return Answer{}, &streamError{Provider: "mistral", Err: err} + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + return Answer{}, newAPIError("mistral", http.MethodPost, "/conversations", resp, data) + } + conv, ans, err := readStream(resp.Body) + if err == nil { + return ans, nil + } + var broken *streamError + if conv == "" || !errors.As(err, &broken) || ctx.Err() != nil { + return Answer{}, err + } + if rec, ok := m.recover(ctx, conv); ok { + return rec, nil + } + return Answer{}, err +} + +// readStream reads the server-sent events of a conversation until conversation.response.done. +// It returns the conversation id as soon as it is known, also with an error. +func readStream(body io.Reader) (conv string, _ Answer, _ error) { + r := bufio.NewReaderSize(body, 64<<10) + var b answerBuilder + var data []string + dispatch := func() (done bool, err error) { + if len(data) == 0 { + return false, nil + } + raw := strings.Join(data, "\n") + data = data[:0] + var ev event + if err := json.Unmarshal([]byte(raw), &ev); err != nil { + return false, &streamError{Provider: "mistral", Err: fmt.Errorf("unreadable event: %w", err)} + } + switch ev.Type { + case "conversation.response.started": + conv = ev.ConversationID + case "message.output.delta": + b.text(ev.ID, ev.Content) + case "tool.execution.started": + b.tool(ev.Name) + case "conversation.response.error": + return false, fmt.Errorf("mistral conversation error %d: %s", ev.Code, ev.Message) + case "conversation.response.done": + return true, nil + } + return false, nil + } + for { + line, err := r.ReadString('\n') + line = strings.TrimRight(line, "\r\n") + switch { + case line == "": + if done, derr := dispatch(); derr != nil { + return conv, Answer{}, derr + } else if done { + return conv, b.answer(), nil } - var chunks []struct { - Type string `json:"type"` - Text string `json:"text"` + case strings.HasPrefix(line, "data:"): + data = append(data, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " ")) + } + if err != nil { + if done, derr := dispatch(); derr != nil { + return conv, Answer{}, derr + } else if done { + return conv, b.answer(), nil } - if err := json.Unmarshal(o.Content, &chunks); err == nil { - for _, ch := range chunks { - if ch.Type == "text" { - texts = append(texts, ch.Text) - } - } + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + return conv, Answer{}, &streamError{Provider: "mistral", Err: err} + } + } +} + +// recover reads the stored history of a conversation whose stream broke off, a few times, and +// takes its answer once it ends in a verdict. +func (m *Mistral) recover(ctx context.Context, conv string) (Answer, bool) { + for try := 0; try < m.recoverTries; try++ { + select { + case <-ctx.Done(): + return Answer{}, false + case <-time.After(m.recoverWait): + } + var raw json.RawMessage + if err := m.do(ctx, http.MethodGet, "/conversations/"+conv+"/history", nil, &raw); err != nil { + continue + } + if a, err := parseAnswer(raw); err == nil { + if _, verdict := splitVerdict(a.Text); verdict != "" { + return a, true } } } - a.Text = strings.Join(texts, "\n") - return a, nil -} - -func (m *Mistral) Ask(ctx context.Context, agentID, task string) (Answer, error) { - in := map[string]any{"agent_id": agentID, "inputs": task, "store": false} - var raw json.RawMessage - if err := m.do(ctx, http.MethodPost, "/conversations", in, &raw); err != nil { - return Answer{}, err - } - return parseAnswer(raw) + return Answer{}, false } diff --git a/review-bot/mistral_test.go b/review-bot/mistral_test.go new file mode 100644 index 00000000..bdf8abba --- /dev/null +++ b/review-bot/mistral_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "context" + "strings" + "testing" +) + +// A stream that breaks off after the conversation started is answered from its stored history. +func TestBrokenStreamIsAnsweredFromTheHistory(t *testing.T) { + f := newFixture(t) + f.breakStreams = 1 + f.history = `{"conversation_id":"conv_1","entries":[ + {"type":"message.input","content":"Review pull request #7"}, + {"type":"tool.execution","name":"conn_get_file_contents"}, + {"type":"message.output","content":"## Summary\n\nfine\n\n## Recommendation\n\nLooks reasonable.\n\nVerdict: comment"}]}` + if err := f.rev.Review(context.Background(), "k.renner/homelab", 7); err != nil { + t.Fatal(err) + } + if f.convCalls != 1 || f.historyCalls != 1 || len(f.postedReviews()) != 1 { + t.Errorf("conversations %d, history reads %d, reviews %d; want 1, 1 and 1", f.convCalls, f.historyCalls, len(f.postedReviews())) + } +} + +// Without a finished answer in the history, a broken stream is asked again. +func TestBrokenStreamWithoutHistoryIsAskedAgain(t *testing.T) { + f := newFixture(t) + f.breakStreams = 1 + if err := f.rev.Review(context.Background(), "k.renner/homelab", 7); err != nil { + t.Fatal(err) + } + if f.convCalls != 2 || f.historyCalls != 2 || len(f.postedReviews()) != 1 { + t.Errorf("conversations %d, history reads %d, reviews %d; want 2, 2 and 1", f.convCalls, f.historyCalls, len(f.postedReviews())) + } +} + +func TestStreamErrorEventIsNotRetried(t *testing.T) { + f := newFixture(t) + f.streamError = "Failed to generate response." + err := f.rev.Review(context.Background(), "k.renner/homelab", 7) + if err == nil || !strings.Contains(err.Error(), "Failed to generate response.") { + t.Fatalf("want the stream error, got %v", err) + } + if f.convCalls != 1 || len(f.postedReviews()) != 0 { + t.Errorf("conversations %d, reviews %d; want 1 and 0", f.convCalls, len(f.postedReviews())) + } +} + +func TestReadStreamJoinsDeltasOfOneMessage(t *testing.T) { + stream := "event: conversation.response.started\ndata: {\"type\":\"conversation.response.started\",\"conversation_id\":\"c9\"}\n\n" + + "data: {\"type\":\"message.output.delta\",\"id\":\"m1\",\"content\":\"## Sum\"}\n\n" + + "data: {\"type\":\"message.output.delta\",\"id\":\"m1\",\"content\":{\"type\":\"thinking\",\"thinking\":[]}}\n\n" + + "data: {\"type\":\"message.output.delta\",\"id\":\"m1\",\"content\":{\"type\":\"text\",\"text\":\"mary\"}}\n\n" + + "data: {\"type\":\"tool.execution.started\",\"id\":\"t1\",\"name\":\"conn_list_commits\",\"arguments\":\"{}\"}\n\n" + + "data: {\"type\":\"message.output.delta\",\"id\":\"m2\",\"content\":\"next\"}\r\n\r\n" + + "data: {\"type\":\"conversation.response.done\"}\n\n" + conv, a, err := readStream(strings.NewReader(stream)) + if err != nil || conv != "c9" || a.Text != "## Summary\nnext" || a.ToolCalls != 1 { + t.Errorf("conv %q, answer %+v, err %v", conv, a, err) + } +} + +func TestReadStreamWithoutDoneIsBroken(t *testing.T) { + _, _, err := readStream(strings.NewReader("data: {\"type\":\"conversation.response.started\",\"conversation_id\":\"c9\"}\n\n")) + if !retryable(err) { + t.Errorf("a stream that ends without done must be retryable, got %v", err) + } +} diff --git a/review-bot/models.go b/review-bot/models.go new file mode 100644 index 00000000..14924cc6 --- /dev/null +++ b/review-bot/models.go @@ -0,0 +1,61 @@ +package main + +import ( + "fmt" + "strings" +) + +// Model is one model the bot can review with, on one backend. +type Model struct { + Backend string // "mistral" or "anthropic" + Name string // the provider's model id + Description string // what it is good for; the selector reads it +} + +func (m Model) ID() string { return m.Backend + ":" + m.Name } + +var backends = []string{"mistral", "anthropic"} + +// parseModels reads one model per line, "backend:model = description", the description being +// optional. Blank lines and lines starting with # are skipped. The first model is the default. +func parseModels(text string) ([]Model, error) { + var out []Model + seen := map[string]bool{} + for _, line := range strings.Split(text, "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + id, desc, _ := strings.Cut(line, "=") + backend, name, ok := strings.Cut(strings.TrimSpace(id), ":") + backend, name = strings.TrimSpace(backend), strings.TrimSpace(name) + if !ok || name == "" { + return nil, fmt.Errorf("MODELS line %q is not backend:model", line) + } + known := false + for _, b := range backends { + known = known || b == backend + } + if !known { + return nil, fmt.Errorf("MODELS line %q: unknown backend %q (want one of %s)", line, backend, strings.Join(backends, ", ")) + } + m := Model{Backend: backend, Name: name, Description: strings.TrimSpace(desc)} + if seen[m.ID()] { + return nil, fmt.Errorf("MODELS lists %s twice", m.ID()) + } + seen[m.ID()] = true + out = append(out, m) + } + return out, nil +} + +// modelsOf lists the model names of one backend. +func modelsOf(models []Model, backend string) []string { + var out []string + for _, m := range models { + if m.Backend == backend { + out = append(out, m.Name) + } + } + return out +} diff --git a/review-bot/prompt.md b/review-bot/prompt.md index 42e84042..eb97e7d1 100644 --- a/review-bot/prompt.md +++ b/review-bot/prompt.md @@ -1,6 +1,6 @@ -# PR Review (Mistral agent with read-only repository access) +# PR Review (agent with read-only repository access) -You review one pull request. You can read the repository through the Gitea connector tools attached to this +You review one pull request. You can read the repository through the Gitea tools attached to this conversation: pull request details and diff, file contents at any ref, directory listings, the repository tree, branches, commits, issues and comments. Use them. Do not guess what you can read. diff --git a/review-bot/review.go b/review-bot/review.go index ddb79b37..c0adeffc 100644 --- a/review-bot/review.go +++ b/review-bot/review.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "log/slog" - "net/http" "strings" "sync" "time" @@ -14,40 +13,88 @@ import ( "github.com/failsafe-go/failsafe-go/retrypolicy" ) -// Reviewer reviews one pull request: it asks the Mistral agent, cleans the answer, derives the -// review event and posts the result as the review bot. +// Reviewer reviews one pull request: it picks the model, asks that model's backend, cleans the +// answer, derives the review event and posts the result as the review bot. type Reviewer struct { - cfg Config - gitea *Gitea - mistral *Mistral - log *slog.Logger - metrics *Metrics - status *Tracker + cfg Config + gitea *Gitea + backends map[string]Backend + selector *TypeSafe // nil: always the default model + log *slog.Logger + metrics *Metrics + status *Tracker - mu sync.RWMutex - agentID string - model string + mu sync.RWMutex + ready map[string]bool // backend -> prepared } -func (r *Reviewer) setAgent(a Agent) { +func (r *Reviewer) setReady(backend string, ok bool) { r.mu.Lock() defer r.mu.Unlock() - r.agentID, r.model = a.ID, a.Model + if r.ready == nil { + r.ready = map[string]bool{} + } + r.ready[backend] = ok } -func (r *Reviewer) agent() (id, model string) { +func (r *Reviewer) isReady(backend string) bool { r.mu.RLock() defer r.mu.RUnlock() - return r.agentID, r.model + return r.ready[backend] } -func (r *Reviewer) Ready() bool { id, _ := r.agent(); return id != "" } +// fallback is the first configured model whose backend is ready. +func (r *Reviewer) fallback() (Model, bool) { + for _, m := range r.cfg.Models { + if r.isReady(m.Backend) { + return m, true + } + } + return Model{}, false +} + +func (r *Reviewer) Ready() bool { _, ok := r.fallback(); return ok } + +// pick chooses the model for a pull request: the selector's choice when it is on and its choice +// is ready, else the first ready model. +func (r *Reviewer) pick(ctx context.Context, slug string, pr PR) (Model, error) { + def, ok := r.fallback() + if !ok { + return Model{}, errors.New("no review backend is ready") + } + if r.selector == nil { + return def, nil + } + files, err := r.gitea.Files(ctx, slug, pr.Number) + if err != nil { + r.log.Warn("selector: cannot list the changed files", "pr", key(slug, pr.Number), "err", err) + } + sctx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + c, err := r.selector.Select(sctx, slug, pr, files, r.cfg.Models) + if err != nil { + r.metrics.Inc("selections_total", "result", "failed") + r.log.Warn("selector failed, using the default model", "pr", key(slug, pr.Number), "model", def.ID(), "err", err) + return def, nil + } + for _, m := range r.cfg.Models { + if m.ID() == c.Model && r.isReady(m.Backend) { + r.metrics.Inc("selections_total", "result", "chosen") + r.log.Info("model selected", "pr", key(slug, pr.Number), "model", m.ID(), "confidence", c.Confidence, + "probabilities", c.Probabilities) + return m, nil + } + } + r.metrics.Inc("selections_total", "result", "unusable") + r.log.Warn("selector chose a model that is not ready, using the default", "pr", key(slug, pr.Number), + "chosen", c.Model, "model", def.ID()) + return def, nil +} // Review reviews slug#n unless it needs none (closed, draft, WIP, already reviewed at this head). func (r *Reviewer) Review(ctx context.Context, slug string, n int) error { - agentID, model := r.agent() - if agentID == "" { - return errors.New("the review agent is not ready") + if !r.Ready() { + return errors.New("no review backend is ready") } pr, err := r.gitea.PR(ctx, slug, n) if err != nil { @@ -67,18 +114,23 @@ func (r *Reviewer) Review(ctx context.Context, slug string, n int) error { } return nil } - r.status.Set(slug, n, pr.Head.SHA, PhaseRunning, "", pr.HTMLURL) + model, err := r.pick(ctx, slug, pr) + if err != nil { + return err + } + r.status.Set(slug, n, pr.Head.SHA, PhaseRunning, model.ID(), pr.HTMLURL) start := time.Now() task := fmt.Sprintf("Review pull request #%d of %s. Head commit: %s.", n, slug, pr.Head.SHA) - ans, err := r.ask(ctx, agentID, task, func(detail string) { + ans, err := r.ask(ctx, model, task, func(detail string) { r.status.Set(slug, n, pr.Head.SHA, PhaseRetrying, detail, pr.HTMLURL) }) if err != nil { r.metrics.Inc("reviews_total", "result", "failed") return err } - r.log.Info("agent answered", "pr", key(slug, n), "tool_calls", ans.ToolCalls, "seconds", int(time.Since(start).Seconds())) + r.log.Info("agent answered", "pr", key(slug, n), "model", model.ID(), "tool_calls", ans.ToolCalls, + "seconds", int(time.Since(start).Seconds())) if len(ans.Violations) > 0 { r.metrics.Inc("reviews_total", "result", "violation") return fmt.Errorf("the agent called tools that are not read-only (%s); nothing is posted", strings.Join(ans.Violations, ", ")) @@ -111,7 +163,7 @@ func (r *Reviewer) Review(ctx context.Context, slug string, n int) error { body := fmt.Sprintf("🤖 **Automated review** — model `%s`, commit `%s`.\n\n"+ "> Automated review by the `review-bot`: it approves only when it found nothing red or yellow, "+ "and a human still merges. Push new commits to get a fresh review.\n\n---\n\n%s\n\n\n", - model, pr.Head.SHA, report, marker(pr.Head.SHA)) + model.Name, pr.Head.SHA, report, marker(pr.Head.SHA)) url, err := r.gitea.PostReview(ctx, slug, n, body, event, pr.Head.SHA) if err != nil { r.metrics.Inc("reviews_total", "result", "failed") @@ -128,15 +180,16 @@ func (r *Reviewer) Review(ctx context.Context, slug string, n int) error { func short(sha string) string { return sha[:min(len(sha), 8)] } -// ask asks the agent, and asks again after a wait when Mistral answers HTTP 429 (its connector rate -// limit) or a gateway failure. Only the status code decides; the text of tool results is never -// inspected. onRetry hears of each scheduled retry. -func (r *Reviewer) ask(ctx context.Context, agentID, task string, onRetry func(detail string)) (Answer, error) { +// ask asks the model's backend, and asks again after a wait on a rate limit, a gateway failure or +// a broken response stream. Only the status code or the transport decides; the text of tool +// results is never inspected. onRetry hears of each scheduled retry. +func (r *Reviewer) ask(ctx context.Context, model Model, task string, onRetry func(detail string)) (Answer, error) { + backend := r.backends[model.Backend] // One conversation's timeout plus the longest waits caps the whole review. ctx, cancelAll := context.WithTimeout(ctx, r.cfg.ReviewTimeout+time.Duration(r.cfg.RateLimitRetries)*maxLimitWait) defer cancelAll() policy := retrypolicy.NewBuilder[Answer](). - HandleIf(func(_ Answer, err error) bool { return retryable(err) != nil }). + HandleIf(func(_ Answer, err error) bool { return retryable(err) }). WithMaxRetries(r.cfg.RateLimitRetries). WithDelayFunc(func(e failsafe.ExecutionAttempt[Answer]) time.Duration { return limitDelay(r.cfg.RateLimitWait, e.Attempts(), e.LastError()) @@ -144,8 +197,8 @@ func (r *Reviewer) ask(ctx context.Context, agentID, task string, onRetry func(d OnRetryScheduled(func(e failsafe.ExecutionScheduledEvent[Answer]) { r.metrics.Inc("retries_total") var headers map[string]string - reason := "error" - if api := retryable(e.LastError()); api != nil { + reason := "broken stream" + if api := asAPIError(e.LastError()); api != nil { headers = api.Limits reason = fmt.Sprintf("HTTP %d", api.Status) } @@ -160,9 +213,9 @@ func (r *Reviewer) ask(ctx context.Context, agentID, task string, onRetry func(d // Every conversation gets the whole timeout; the waits between them do not eat into it. actx, cancel := context.WithTimeout(ctx, r.cfg.ReviewTimeout) defer cancel() - return r.mistral.Ask(actx, agentID, task) + return backend.Ask(actx, model.Name, task) }) - if retryable(err) != nil { + if retryable(err) { r.metrics.Inc("retry_giveups_total") return Answer{}, fmt.Errorf("still failing after %d retries: %w", r.cfg.RateLimitRetries, err) } @@ -173,30 +226,16 @@ func (r *Reviewer) ask(ctx context.Context, agentID, task string, onRetry func(d // configured wait. const maxLimitWait = 5 * time.Minute -// limitDelay is the wait before retry number `retry` (1 for the first): Mistral's own Retry-After -// when the last answer carried one, else the configured wait times the retry number, capped at -// maxLimitWait either way. +// limitDelay is the wait before retry number `retry` (1 for the first): the provider's own +// Retry-After when the last answer carried one, else the configured wait times the retry number, +// capped at maxLimitWait either way. func limitDelay(wait time.Duration, retry int, last error) time.Duration { - if api := retryable(last); api != nil && api.HasRetryAfter { + if api := asAPIError(last); api != nil && retryable(last) && api.HasRetryAfter { return min(api.RetryAfter, maxLimitWait) } return min(wait*time.Duration(retry), maxLimitWait) } -// retryable returns the apiError when err is a status worth asking again after: 429 (the connector -// rate limit) or a gateway failure (502, 503, 504), else nil. -func retryable(err error) *apiError { - var api *apiError - if !errors.As(err, &api) { - return nil - } - switch api.Status { - case http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout: - return api - } - return nil -} - func key(slug string, n int) string { return fmt.Sprintf("%s#%d", slug, n) } type job struct { diff --git a/review-bot/review_test.go b/review-bot/review_test.go index a7bd398b..4c4b1c47 100644 --- a/review-bot/review_test.go +++ b/review-bot/review_test.go @@ -42,6 +42,50 @@ type fixture struct { retryAfter string // Retry-After header sent with a simulated 429 head string // head commit of PR 7 pushAfter string // when set, the next conversation answer is followed by a push of this head + // breakStreams breaks the next conversation streams off after their first event. + breakStreams int + history string // JSON returned for a conversation's history; "" answers 404 + historyCalls int + streamError string // when set, conversations end in a conversation.response.error event + + mistralBackend *Mistral +} + +// sse renders a conversation answer ({"outputs":[...]}) as the event stream Mistral sends. +func sse(t *testing.T, answer string, broken bool, streamErr string) string { + var c struct { + Outputs []map[string]json.RawMessage `json:"outputs"` + } + if err := json.Unmarshal([]byte(answer), &c); err != nil { + t.Fatalf("bad answer fixture: %v", err) + } + var b strings.Builder + write := func(v map[string]any) { + data, _ := json.Marshal(v) + b.WriteString("event: " + v["type"].(string) + "\ndata: " + string(data) + "\n\n") + } + write(map[string]any{"type": "conversation.response.started", "conversation_id": "conv_1"}) + if broken { + return b.String() + } + for i, o := range c.Outputs { + var typ string + _ = json.Unmarshal(o["type"], &typ) + switch typ { + case "tool.execution": + var name string + _ = json.Unmarshal(o["name"], &name) + write(map[string]any{"type": "tool.execution.started", "id": "t", "name": name, "arguments": "{}"}) + case "message.output": + write(map[string]any{"type": "message.output.delta", "id": "m" + string(rune('0'+i)), "content": o["content"]}) + } + } + if streamErr != "" { + write(map[string]any{"type": "conversation.response.error", "message": streamErr, "code": 3000}) + return b.String() + } + write(map[string]any{"type": "conversation.response.done", "usage": map[string]int{"total_tokens": 1}}) + return b.String() } const goodAnswer = `{"outputs":[ @@ -107,7 +151,8 @@ func newFixture(t *testing.T) *fixture { case r.Method == "POST" && r.URL.Path == "/conversations": var in map[string]any _ = json.NewDecoder(r.Body).Decode(&in) - if in["agent_id"] != "ag_1" || !strings.Contains(in["inputs"].(string), "pull request #7 of k.renner/homelab") { + if in["agent_id"] != "ag_1" || in["stream"] != true || in["store"] != true || + !strings.Contains(in["inputs"].(string), "pull request #7 of k.renner/homelab") { http.Error(w, "unexpected conversation request", http.StatusBadRequest) return } @@ -125,10 +170,22 @@ func newFixture(t *testing.T) *fixture { if len(f.answers) > 0 { answer, f.answers = f.answers[0], f.answers[1:] } - _, _ = io.WriteString(w, answer) + broken := f.breakStreams > 0 + if broken { + f.breakStreams-- + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sse(f.t, answer, broken, f.streamError)) if f.pushAfter != "" { f.head, f.pushAfter = f.pushAfter, "" } + case r.Method == "GET" && r.URL.Path == "/conversations/conv_1/history": + f.historyCalls++ + if f.history == "" { + http.NotFound(w, r) + return + } + _, _ = io.WriteString(w, f.history) default: http.NotFound(w, r) } @@ -138,7 +195,8 @@ func newFixture(t *testing.T) *fixture { f.cfg = Config{ GiteaURL: f.gitea.URL, BotLogin: "review-bot", ReviewToken: "review-token", WebhookSecret: "hook-secret", - MistralURL: f.mistral.URL, MistralKey: "mistral-key", AgentName: "pr-review", AgentModel: "mistral-large-4-0", + MistralURL: f.mistral.URL, MistralKey: "mistral-key", AgentName: "pr-review", + Models: []Model{{Backend: "mistral", Name: "mistral-large-4-0"}}, ConnectorID: "c1", Repos: []string{"k.renner/homelab"}, Workers: 1, QueueSize: 4, ReviewTimeout: 5 * time.Second, RateLimitRetries: 2, RateLimitWait: time.Millisecond, } @@ -149,15 +207,18 @@ func newFixture(t *testing.T) *fixture { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) go f.status.Run(ctx) + f.mistralBackend = &Mistral{base: f.mistral.URL, key: "mistral-key", agentPrefix: "pr-review", connectorID: "c1", + http: http.DefaultClient, recoverTries: 2, recoverWait: time.Millisecond, + agents: map[string]string{"mistral-large-4-0": "ag_1"}} f.rev = &Reviewer{ - cfg: f.cfg, - gitea: gitea, - mistral: &Mistral{base: f.mistral.URL, key: "mistral-key", http: http.DefaultClient}, - log: log, - metrics: f.metrics, - status: f.status, + cfg: f.cfg, + gitea: gitea, + backends: map[string]Backend{"mistral": f.mistralBackend}, + log: log, + metrics: f.metrics, + status: f.status, } - f.rev.setAgent(Agent{ID: "ag_1", Model: "mistral-large-4-0"}) + f.rev.setReady("mistral", true) return f } @@ -245,7 +306,7 @@ func TestReviewWithoutAReportPostsNothing(t *testing.T) { func TestReviewNeedsTheAgent(t *testing.T) { f := newFixture(t) - f.rev.setAgent(Agent{}) + f.rev.setReady("mistral", false) if err := f.rev.Review(context.Background(), "k.renner/homelab", 7); err == nil { t.Fatal("want an error while the agent is not ready") } @@ -320,7 +381,7 @@ func TestEnsureAgent(t *testing.T) { hash := promptHash(reviewPrompt) f.agents = "[]" - a, err := f.rev.mistral.EnsureAgent(context.Background(), "pr-review", "mistral-large-4-0", "c1", reviewPrompt) + a, err := f.mistralBackend.EnsureAgent(context.Background(), "pr-review", "mistral-large-4-0", "c1", reviewPrompt) if err != nil || a.ID != "ag_new" { t.Fatalf("create: %+v, %v", a, err) } @@ -329,7 +390,7 @@ func TestEnsureAgent(t *testing.T) { f.agentCall = nil f.agents = `[{"id":"ag_9","name":"pr-review","model":"mistral-large-4-0","description":"PR review, read-only Gitea connector. prompt sha256:` + hash + `"}]` f.mu.Unlock() - a, err = f.rev.mistral.EnsureAgent(context.Background(), "pr-review", "mistral-large-4-0", "c1", reviewPrompt) + a, err = f.mistralBackend.EnsureAgent(context.Background(), "pr-review", "mistral-large-4-0", "c1", reviewPrompt) if err != nil || a.ID != "ag_9" || len(f.agentCall) != 0 { t.Fatalf("unchanged agent must not be touched: %+v, %v, calls %v", a, err, f.agentCall) } @@ -337,7 +398,7 @@ func TestEnsureAgent(t *testing.T) { f.mu.Lock() f.agents = `[{"id":"ag_9","name":"pr-review","model":"mistral-large-4-0","description":"PR review, read-only Gitea connector. prompt sha256:old"}]` f.mu.Unlock() - a, err = f.rev.mistral.EnsureAgent(context.Background(), "pr-review", "mistral-large-4-0", "c1", reviewPrompt) + a, err = f.mistralBackend.EnsureAgent(context.Background(), "pr-review", "mistral-large-4-0", "c1", reviewPrompt) if err != nil || a.ID != "ag_9" || len(f.agentCall) != 1 || f.agentCall[0] != "PATCH ag_9" { t.Fatalf("drifted agent must be updated in place: %+v, %v, calls %v", a, err, f.agentCall) } @@ -442,7 +503,7 @@ func TestHealthEndpoints(t *testing.T) { if get("/healthz") != 200 || get("/readyz") != 200 || get("/metrics") != 200 { t.Error("healthy service must answer 200 on healthz, readyz and metrics") } - f.rev.setAgent(Agent{}) + f.rev.setReady("mistral", false) if get("/healthz") != 200 || get("/readyz") != 503 { t.Error("without an agent the service is alive but not ready") } diff --git a/review-bot/selector.go b/review-bot/selector.go new file mode 100644 index 00000000..7b80e2ee --- /dev/null +++ b/review-bot/selector.go @@ -0,0 +1,101 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" +) + +// PRFile is one changed file of a pull request. +type PRFile struct { + Filename string `json:"filename"` + Status string `json:"status"` + Additions int `json:"additions"` + Deletions int `json:"deletions"` +} + +// Choice is the selector's pick with the probability of every model. +type Choice struct { + Model string + Confidence float64 + Probabilities map[string]float64 +} + +// selectInstructions is the question the selector answers for every pull request. +const selectInstructions = "Pick the model that should review the pull request in `pull_request`. " + + "The review reads the repository, checks the change for bugs, security problems and broken behaviour, " + + "and must be thorough. Prefer a cheaper or faster model when the change is small and low-risk " + + "(documentation, version bumps, simple configuration). Pick a stronger model when the change is large, " + + "spans several components, or touches security, authentication, secrets, network policy, concurrency, " + + "data migrations or anything that is hard to undo." + +// TypeSafe picks the review model with a System One Choice question. +type TypeSafe struct { + url string + key string + model string + http *http.Client +} + +func (t *TypeSafe) Select(ctx context.Context, slug string, pr PR, files []PRFile, models []Model) (Choice, error) { + criteria := map[string]any{} + for _, m := range models { + if m.Description != "" { + criteria[m.ID()] = m.Description + } else { + criteria[m.ID()] = nil + } + } + state := map[string]any{"pull_request": map[string]any{ + "repository": slug, "title": pr.Title, "description": pr.Body, + "additions": pr.Additions, "deletions": pr.Deletions, "changed_files": pr.ChangedFiles, + "files": files, + }} + in := map[string]any{ + "state": state, + "model": t.model, + "questions": map[string]any{"model": map[string]any{ + "type": "choice", "instructions": selectInstructions, "criteria": criteria, + }}, + } + b, err := json.Marshal(in) + if err != nil { + return Choice{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, t.url, bytes.NewReader(b)) + if err != nil { + return Choice{}, err + } + req.Header.Set("Authorization", "Bearer "+t.key) + req.Header.Set("Content-Type", "application/json") + resp, err := t.http.Do(req) + if err != nil { + return Choice{}, err + } + defer resp.Body.Close() + data, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if err != nil { + return Choice{}, err + } + if resp.StatusCode != http.StatusOK { + return Choice{}, newAPIError("typesafe", http.MethodPost, "/v1/systemone", resp, data) + } + var out struct { + Answers map[string]struct { + Choice string `json:"choice"` + Confidence float64 `json:"confidence"` + Probabilities map[string]float64 `json:"probabilities"` + } `json:"answers"` + } + if err := json.Unmarshal(data, &out); err != nil { + return Choice{}, err + } + a, ok := out.Answers["model"] + if !ok || a.Choice == "" { + return Choice{}, fmt.Errorf("typesafe answered without a model: %s", snippet(data)) + } + return Choice{Model: a.Choice, Confidence: a.Confidence, Probabilities: a.Probabilities}, nil +} diff --git a/review-bot/status_test.go b/review-bot/status_test.go index 54d5ddaa..232568d6 100644 --- a/review-bot/status_test.go +++ b/review-bot/status_test.go @@ -24,7 +24,7 @@ func TestQueueReportsQueuedRunningAndPosted(t *testing.T) { want := []string{ "abc123 pending: queued", - "abc123 pending: review running", + "abc123 pending: review running: mistral:mistral-large-4-0", "abc123 success: reviewed: request changes", } if got := f.commitStatuses(); !slices.Equal(got, want) {