| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "bytes" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "net/http" |
| 11 | "os" |
| 12 | "strings" |
| 13 | "time" |
| 14 | ) |
| 15 | |
| 16 | // A minimal Anthropic Messages API client — just the slice the bot brain |
| 17 | // needs (system blocks with cache markers, adaptive thinking, tools), on |
| 18 | // net/http so the static binary takes no new dependency. Base URL and key |
| 19 | // come from the same env the SDKs read (ANTHROPIC_API_KEY, |
| 20 | // ANTHROPIC_BASE_URL), so tests point it at an httptest server. |
| 21 | |
| 22 | const anthropicVersion = "2023-06-01" |
| 23 | |
| 24 | type anthropicClient struct { |
| 25 | key string |
| 26 | base string |
| 27 | http *http.Client |
| 28 | // record, when set, receives every successful call's usage block — the |
| 29 | // usage meter's tap (#857). It sits here, not at the call sites, so every |
| 30 | // model path (reply, classifier, task, research, app build) is counted |
| 31 | // without each one remembering to. |
| 32 | record func(model string, u anthUsage) |
| 33 | } |
| 34 | |
| 35 | func newAnthropicClient(key string, record func(model string, u anthUsage)) *anthropicClient { |
| 36 | base := os.Getenv("ANTHROPIC_BASE_URL") |
| 37 | if base == "" { |
| 38 | base = "https://api.anthropic.com" |
| 39 | } |
| 40 | // Long timeout: a thinking run over a large cached system prompt can |
| 41 | // legitimately take a couple of minutes. |
| 42 | return &anthropicClient{key: key, base: base, http: &http.Client{Timeout: 5 * time.Minute}, record: record} |
| 43 | } |
| 44 | |
| 45 | type anthBlock struct { |
| 46 | Type string `json:"type"` |
| 47 | Text string `json:"text,omitempty"` |
| 48 | // CacheControl marks the prefix up to this block cacheable (the big system |
| 49 | // prompt rides every call — see systemBlocks in brain.go). |
| 50 | CacheControl *anthCache `json:"cache_control,omitempty"` |
| 51 | // tool_use response fields |
| 52 | ID string `json:"id,omitempty"` |
| 53 | Name string `json:"name,omitempty"` |
| 54 | Input map[string]any `json:"input,omitempty"` |
| 55 | // tool_result request fields (the task loop's round-trip, task.go) |
| 56 | ToolUseID string `json:"tool_use_id,omitempty"` |
| 57 | Content any `json:"content,omitempty"` |
| 58 | // Extended-thinking blocks. A tool loop echoes the model's content back |
| 59 | // verbatim as the assistant turn, and the API REQUIRES thinking blocks to |
| 60 | // come back intact (signature and all) — dropping these fields turns |
| 61 | // every live thinking+tools run into a hard 400 ("thinking.thinking: |
| 62 | // Field required"), which scripted test fakes can never surface. |
| 63 | Thinking string `json:"thinking,omitempty"` |
| 64 | Signature string `json:"signature,omitempty"` |
| 65 | Data string `json:"data,omitempty"` // redacted_thinking payload |
| 66 | } |
| 67 | |
| 68 | // MarshalJSON closes two omitempty traps on the tool-loop echo — both the |
| 69 | // same shape: the API REQUIRES a field the model sometimes returns EMPTY, |
| 70 | // and Go's omitempty drops empty values entirely, turning the echo into a |
| 71 | // hard 400 on the loop's next call. |
| 72 | // |
| 73 | // - tool_use.input: a tool called with NO arguments (a bare issues.list) |
| 74 | // echoed with no input at all → "tool_use.input: Field required". |
| 75 | // - thinking.thinking: adaptive thinking can emit a ZERO-LENGTH thinking |
| 76 | // block on a continuation turn (seen live from the read_app_source |
| 77 | // loop) → "thinking.thinking: Field required". |
| 78 | // |
| 79 | // Each marshals with the field explicitly present; every other block shape |
| 80 | // is untouched. |
| 81 | func (b anthBlock) MarshalJSON() ([]byte, error) { |
| 82 | type alias anthBlock // method-less: plain marshalling, no recursion |
| 83 | switch { |
| 84 | case b.Type == "tool_use" && len(b.Input) == 0: |
| 85 | return json.Marshal(struct { |
| 86 | alias |
| 87 | Input map[string]any `json:"input"` // shallower field wins: {} rides |
| 88 | }{alias(b), map[string]any{}}) |
| 89 | case b.Type == "thinking" && b.Thinking == "": |
| 90 | return json.Marshal(struct { |
| 91 | alias |
| 92 | Thinking string `json:"thinking"` // shallower field wins: "" rides |
| 93 | }{alias(b), ""}) |
| 94 | } |
| 95 | return json.Marshal(alias(b)) |
| 96 | } |
| 97 | |
| 98 | type anthCache struct { |
| 99 | Type string `json:"type"` // "ephemeral" |
| 100 | } |
| 101 | |
| 102 | // anthMessage's content is either a plain string (the conversational path) or |
| 103 | // a []anthBlock (the task loop's tool_use / tool_result turns). |
| 104 | type anthMessage struct { |
| 105 | Role string `json:"role"` |
| 106 | Content any `json:"content"` |
| 107 | } |
| 108 | |
| 109 | type anthTool struct { |
| 110 | Name string `json:"name"` |
| 111 | Description string `json:"description"` |
| 112 | InputSchema map[string]any `json:"input_schema"` |
| 113 | } |
| 114 | |
| 115 | type anthRequest struct { |
| 116 | Model string `json:"model"` |
| 117 | MaxTokens int `json:"max_tokens"` |
| 118 | System []anthBlock `json:"system,omitempty"` |
| 119 | Thinking map[string]any `json:"thinking,omitempty"` |
| 120 | Tools []anthTool `json:"tools,omitempty"` |
| 121 | ToolChoice map[string]any `json:"tool_choice,omitempty"` |
| 122 | Messages []anthMessage `json:"messages"` |
| 123 | Stream bool `json:"stream,omitempty"` |
| 124 | } |
| 125 | |
| 126 | type anthResponse struct { |
| 127 | Content []anthBlock `json:"content"` |
| 128 | StopReason string `json:"stop_reason"` |
| 129 | Usage anthUsage `json:"usage"` |
| 130 | } |
| 131 | |
| 132 | // anthUsage is the token accounting block every Messages response carries. |
| 133 | // All four counters are kept distinct — cache reads and writes price |
| 134 | // differently from fresh input, so the usage meter (#857) records each as its |
| 135 | // own dimension. |
| 136 | type anthUsage struct { |
| 137 | InputTokens int64 `json:"input_tokens"` |
| 138 | OutputTokens int64 `json:"output_tokens"` |
| 139 | CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"` |
| 140 | CacheReadInputTokens int64 `json:"cache_read_input_tokens"` |
| 141 | } |
| 142 | |
| 143 | // anthHTTPError is a non-200 response — carried as an error so create's retry |
| 144 | // loop can tell a deterministic client error (4xx: bad request/auth) from a |
| 145 | // transient one (5xx/429) worth retrying. |
| 146 | type anthHTTPError struct { |
| 147 | status int |
| 148 | body string |
| 149 | } |
| 150 | |
| 151 | func (e *anthHTTPError) Error() string { return fmt.Sprintf("anthropic: %d: %s", e.status, e.body) } |
| 152 | |
| 153 | // createAttempts is how many times create tries a request. The long build call |
| 154 | // (large max_tokens + thinking) is a non-streamed request that intermediaries |
| 155 | // sometimes cut mid-response ("unexpected EOF") — almost always transient, so a |
| 156 | // couple of retries makes the builder resilient without masking real failures. |
| 157 | const createAttempts = 3 |
| 158 | |
| 159 | func (c *anthropicClient) create(req anthRequest) (*anthResponse, error) { |
| 160 | return c.do(req, false) |
| 161 | } |
| 162 | |
| 163 | // createStream makes the request as a server-sent-event stream and assembles the |
| 164 | // same anthResponse from the deltas. Streaming keeps the connection continuously |
| 165 | // active with incremental events, so a long generation isn't cut by an |
| 166 | // idle-connection timeout the way a non-streamed request is — the fix for the |
| 167 | // app builder's recurring "unexpected EOF". Use it for the long build call. |
| 168 | func (c *anthropicClient) createStream(req anthRequest) (*anthResponse, error) { |
| 169 | return c.do(req, true) |
| 170 | } |
| 171 | |
| 172 | func (c *anthropicClient) do(req anthRequest, stream bool) (*anthResponse, error) { |
| 173 | req.Stream = stream |
| 174 | body, err := json.Marshal(req) |
| 175 | if err != nil { |
| 176 | return nil, err |
| 177 | } |
| 178 | var lastErr error |
| 179 | for attempt := 1; attempt <= createAttempts; attempt++ { |
| 180 | if attempt > 1 { |
| 181 | time.Sleep(time.Duration(attempt-1) * time.Second) // 1s, 2s |
| 182 | } |
| 183 | var resp *anthResponse |
| 184 | if stream { |
| 185 | resp, err = c.doStream(body) |
| 186 | } else { |
| 187 | resp, err = c.doRequest(body) |
| 188 | } |
| 189 | if err == nil { |
| 190 | if c.record != nil { |
| 191 | c.record(req.Model, resp.Usage) |
| 192 | } |
| 193 | return resp, nil |
| 194 | } |
| 195 | lastErr = err |
| 196 | // Deterministic client errors (4xx other than 429) won't change on a |
| 197 | // retry — fail fast. Everything else (transport drops like unexpected |
| 198 | // EOF, 5xx, 429, a cut stream) is retried. |
| 199 | var he *anthHTTPError |
| 200 | if errors.As(err, &he) && he.status < 500 && he.status != http.StatusTooManyRequests { |
| 201 | return nil, err |
| 202 | } |
| 203 | } |
| 204 | return nil, lastErr |
| 205 | } |
| 206 | |
| 207 | func (c *anthropicClient) doRequest(body []byte) (*anthResponse, error) { |
| 208 | hr, err := http.NewRequest("POST", c.base+"/v1/messages", bytes.NewReader(body)) |
| 209 | if err != nil { |
| 210 | return nil, err |
| 211 | } |
| 212 | hr.Header.Set("Content-Type", "application/json") |
| 213 | hr.Header.Set("x-api-key", c.key) |
| 214 | hr.Header.Set("anthropic-version", anthropicVersion) |
| 215 | resp, err := c.http.Do(hr) |
| 216 | if err != nil { |
| 217 | return nil, err // transport error (DNS, TLS, connection reset/EOF before headers) |
| 218 | } |
| 219 | defer resp.Body.Close() |
| 220 | if resp.StatusCode != http.StatusOK { |
| 221 | msg, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) |
| 222 | return nil, &anthHTTPError{status: resp.StatusCode, body: string(msg)} |
| 223 | } |
| 224 | var out anthResponse |
| 225 | if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { |
| 226 | return nil, err // e.g. unexpected EOF: connection cut mid-response |
| 227 | } |
| 228 | return &out, nil |
| 229 | } |
| 230 | |
| 231 | // doStream issues the request with stream:true and reassembles the response from |
| 232 | // the SSE deltas. It accumulates text and tool_use blocks (the only kinds the |
| 233 | // bot reads back — thinking is streamed too but ignored). A stream cut before |
| 234 | // message_stop returns an error, which the retry loop treats as transient. |
| 235 | func (c *anthropicClient) doStream(body []byte) (*anthResponse, error) { |
| 236 | hr, err := http.NewRequest("POST", c.base+"/v1/messages", bytes.NewReader(body)) |
| 237 | if err != nil { |
| 238 | return nil, err |
| 239 | } |
| 240 | hr.Header.Set("Content-Type", "application/json") |
| 241 | hr.Header.Set("x-api-key", c.key) |
| 242 | hr.Header.Set("anthropic-version", anthropicVersion) |
| 243 | hr.Header.Set("Accept", "text/event-stream") |
| 244 | resp, err := c.http.Do(hr) |
| 245 | if err != nil { |
| 246 | return nil, err |
| 247 | } |
| 248 | defer resp.Body.Close() |
| 249 | if resp.StatusCode != http.StatusOK { |
| 250 | msg, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) |
| 251 | return nil, &anthHTTPError{status: resp.StatusCode, body: string(msg)} |
| 252 | } |
| 253 | |
| 254 | // One accumulator per content block, indexed as the API indexes them. |
| 255 | type block struct { |
| 256 | typ string |
| 257 | id string |
| 258 | name string |
| 259 | text strings.Builder |
| 260 | js strings.Builder // partial tool_use input JSON |
| 261 | } |
| 262 | var blocks []*block |
| 263 | out := &anthResponse{} |
| 264 | |
| 265 | sc := bufio.NewScanner(resp.Body) |
| 266 | sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024) // deltas are small; be generous anyway |
| 267 | done := false |
| 268 | for sc.Scan() { |
| 269 | line := sc.Text() |
| 270 | if !strings.HasPrefix(line, "data:") { |
| 271 | continue // skip "event:" lines and blank separators — the data JSON self-describes |
| 272 | } |
| 273 | data := strings.TrimSpace(line[len("data:"):]) |
| 274 | if data == "" { |
| 275 | continue |
| 276 | } |
| 277 | var ev struct { |
| 278 | Type string `json:"type"` |
| 279 | Index int `json:"index"` |
| 280 | ContentBlock *struct { |
| 281 | Type string `json:"type"` |
| 282 | ID string `json:"id"` |
| 283 | Name string `json:"name"` |
| 284 | } `json:"content_block"` |
| 285 | Delta *struct { |
| 286 | Type string `json:"type"` |
| 287 | Text string `json:"text"` |
| 288 | PartialJSON string `json:"partial_json"` |
| 289 | StopReason string `json:"stop_reason"` |
| 290 | } `json:"delta"` |
| 291 | // Usage rides message_start (input + cache counters, inside the |
| 292 | // message envelope) and message_delta (cumulative output tokens, |
| 293 | // as a sibling of delta) — both are folded into out.Usage. |
| 294 | Message *struct { |
| 295 | Usage anthUsage `json:"usage"` |
| 296 | } `json:"message"` |
| 297 | Usage *anthUsage `json:"usage"` |
| 298 | Error *struct { |
| 299 | Type string `json:"type"` |
| 300 | Message string `json:"message"` |
| 301 | } `json:"error"` |
| 302 | } |
| 303 | if json.Unmarshal([]byte(data), &ev) != nil { |
| 304 | continue |
| 305 | } |
| 306 | switch ev.Type { |
| 307 | case "message_start": |
| 308 | if ev.Message != nil { |
| 309 | out.Usage = ev.Message.Usage |
| 310 | } |
| 311 | case "content_block_start": |
| 312 | for len(blocks) <= ev.Index { |
| 313 | blocks = append(blocks, nil) |
| 314 | } |
| 315 | b := &block{} |
| 316 | if ev.ContentBlock != nil { |
| 317 | b.typ = ev.ContentBlock.Type |
| 318 | b.id = ev.ContentBlock.ID |
| 319 | b.name = ev.ContentBlock.Name |
| 320 | } |
| 321 | blocks[ev.Index] = b |
| 322 | case "content_block_delta": |
| 323 | if ev.Index >= 0 && ev.Index < len(blocks) && blocks[ev.Index] != nil && ev.Delta != nil { |
| 324 | switch ev.Delta.Type { |
| 325 | case "text_delta": |
| 326 | blocks[ev.Index].text.WriteString(ev.Delta.Text) |
| 327 | case "input_json_delta": |
| 328 | blocks[ev.Index].js.WriteString(ev.Delta.PartialJSON) |
| 329 | } |
| 330 | } |
| 331 | case "message_delta": |
| 332 | if ev.Delta != nil && ev.Delta.StopReason != "" { |
| 333 | out.StopReason = ev.Delta.StopReason |
| 334 | } |
| 335 | // message_delta usage counters are cumulative totals — take the |
| 336 | // latest non-zero values rather than summing. |
| 337 | if ev.Usage != nil { |
| 338 | if ev.Usage.OutputTokens > 0 { |
| 339 | out.Usage.OutputTokens = ev.Usage.OutputTokens |
| 340 | } |
| 341 | if ev.Usage.InputTokens > 0 { |
| 342 | out.Usage.InputTokens = ev.Usage.InputTokens |
| 343 | } |
| 344 | if ev.Usage.CacheCreationInputTokens > 0 { |
| 345 | out.Usage.CacheCreationInputTokens = ev.Usage.CacheCreationInputTokens |
| 346 | } |
| 347 | if ev.Usage.CacheReadInputTokens > 0 { |
| 348 | out.Usage.CacheReadInputTokens = ev.Usage.CacheReadInputTokens |
| 349 | } |
| 350 | } |
| 351 | case "error": |
| 352 | if ev.Error != nil { |
| 353 | return nil, fmt.Errorf("anthropic: stream error: %s: %s", ev.Error.Type, ev.Error.Message) |
| 354 | } |
| 355 | return nil, fmt.Errorf("anthropic: stream error") |
| 356 | case "message_stop": |
| 357 | done = true |
| 358 | } |
| 359 | } |
| 360 | if err := sc.Err(); err != nil { |
| 361 | return nil, err // stream cut mid-response — retryable |
| 362 | } |
| 363 | if !done { |
| 364 | return nil, fmt.Errorf("anthropic: stream ended without message_stop") // retryable |
| 365 | } |
| 366 | for _, b := range blocks { |
| 367 | if b == nil { |
| 368 | continue |
| 369 | } |
| 370 | switch b.typ { |
| 371 | case "text": |
| 372 | out.Content = append(out.Content, anthBlock{Type: "text", Text: b.text.String()}) |
| 373 | case "tool_use": |
| 374 | blk := anthBlock{Type: "tool_use", ID: b.id, Name: b.name} |
| 375 | if b.js.Len() > 0 { |
| 376 | _ = json.Unmarshal([]byte(b.js.String()), &blk.Input) |
| 377 | } |
| 378 | out.Content = append(out.Content, blk) |
| 379 | } |
| 380 | } |
| 381 | return out, nil |
| 382 | } |
| 383 | |
| 384 | // text joins the response's text blocks. |
| 385 | func (r *anthResponse) text() string { |
| 386 | var b bytes.Buffer |
| 387 | for _, blk := range r.Content { |
| 388 | if blk.Type == "text" { |
| 389 | b.WriteString(blk.Text) |
| 390 | } |
| 391 | } |
| 392 | return b.String() |
| 393 | } |
| 394 | |
amadan