package llm import ( "context" "encoding/json" "io" "net/http" "net/http/httptest" "strings" "testing" "time" ) // sse formats events the way the API does. func sse(events ...string) string { var b strings.Builder for _, e := range events { b.WriteString("event: x\ndata: " + e + "\n\n") } return b.String() } func streamServer(t *testing.T, payload string) *Client { t.Helper() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Header.Get("accept") != "text/event-stream" { t.Errorf("accept = %q", r.Header.Get("accept")) } var req map[string]any body, _ := io.ReadAll(r.Body) json.Unmarshal(body, &req) if req["stream"] != true { t.Errorf("stream flag not set in body: %s", body) } w.Header().Set("content-type", "text/event-stream") io.WriteString(w, payload) })) t.Cleanup(srv.Close) c := New(WithBaseURL(srv.URL), WithAPIKey("k")) c.RetryDelay = time.Millisecond return c } // The fiddly part: tool arguments arrive as fragments that are individually // invalid JSON and only parseable once concatenated. func TestStreamAccumulatesToolInputFragments(t *testing.T) { c := streamServer(t, sse( `{"type":"message_start","message":{"id":"msg_1","model":"claude-opus-5","usage":{"input_tokens":100,"cache_read_input_tokens":40}}}`, `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me "}}`, `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"look."}}`, `{"type":"content_block_stop","index":0}`, `{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"tu_1","name":"edit_file","input":{}}}`, `{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}}`, `{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"\"main.go\",\"old_str\""}}`, `{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":":\"a\",\"new_str\":\"b\"}"}}`, `{"type":"content_block_stop","index":1}`, `{"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":55}}`, `{"type":"message_stop"}`, )) var textSeen strings.Builder var toolStarts int resp, err := c.StreamMessage(context.Background(), Request{}, func(ev StreamEvent) { switch ev.Kind { case StreamText: textSeen.WriteString(ev.Text) case StreamToolStart: toolStarts++ if ev.ToolName != "edit_file" || ev.ToolID != "tu_1" { t.Errorf("tool start = %+v", ev) } } }) if err != nil { t.Fatalf("StreamMessage: %v", err) } if textSeen.String() != "Let me look." { t.Errorf("streamed text = %q", textSeen.String()) } if toolStarts != 1 { t.Errorf("%d tool starts, want 1", toolStarts) } if resp.StopReason != StopToolUse { t.Errorf("stop_reason = %q", resp.StopReason) } if got := resp.Text(); got != "Let me look." { t.Errorf("assembled text = %q", got) } uses := resp.ToolUses() if len(uses) != 1 { t.Fatalf("%d tool uses, want 1", len(uses)) } var args struct{ Path, OldStr, NewStr string } if err := json.Unmarshal(uses[0].Input, &struct { Path *string `json:"path"` OldStr *string `json:"old_str"` NewStr *string `json:"new_str"` }{&args.Path, &args.OldStr, &args.NewStr}); err != nil { t.Fatalf("reassembled input is not valid JSON: %v (%s)", err, uses[0].Input) } if args.Path != "main.go" || args.OldStr != "a" || args.NewStr != "b" { t.Errorf("reassembled args = %+v", args) } if resp.Usage.InputTokens != 100 || resp.Usage.CacheReadInputTokens != 40 || resp.Usage.OutputTokens != 55 { t.Errorf("usage = %+v: input counts come from message_start, output from message_delta", resp.Usage) } } func TestStreamPreservesThinkingSignature(t *testing.T) { c := streamServer(t, sse( `{"type":"message_start","message":{"id":"m","usage":{"input_tokens":1}}}`, `{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}`, `{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-"}}`, `{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"abc"}}`, `{"type":"content_block_stop","index":0}`, `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}`, )) resp, err := c.StreamMessage(context.Background(), Request{}, nil) if err != nil { t.Fatalf("StreamMessage: %v", err) } if len(resp.Content) != 1 || resp.Content[0].Signature != "sig-abc" { t.Errorf("signature = %q, want the fragments joined", resp.Content[0].Signature) } out, _ := json.Marshal(resp.Content[0]) if !strings.Contains(string(out), `"signature":"sig-abc"`) { t.Errorf("signature must survive re-marshalling: %s", out) } } func TestStreamErrors(t *testing.T) { tests := []struct { name string payload string want string }{ { name: "mid-stream error event", payload: sse( `{"type":"message_start","message":{"id":"m"}}`, `{"type":"error","error":{"type":"overloaded_error","message":"try later"}}`), want: "try later", }, { name: "truncated tool arguments", payload: sse( `{"type":"message_start","message":{"id":"m"}}`, `{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t","name":"bash"}}`, `{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"command\":"}}`, `{"type":"content_block_stop","index":0}`, `{"type":"message_delta","delta":{"stop_reason":"tool_use"}}`), want: "invalid JSON arguments", }, { name: "stream ends without a stop reason", payload: sse(`{"type":"message_start","message":{"id":"m"}}`), want: "without a stop_reason", }, { name: "malformed event", payload: "data: {not json\n\n", want: "malformed stream event", }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { _, err := streamServer(t, tc.payload).StreamMessage(context.Background(), Request{}, nil) if err == nil || !strings.Contains(err.Error(), tc.want) { t.Errorf("error = %v, want it to mention %q", err, tc.want) } }) } } func TestStreamToolUseWithNoArguments(t *testing.T) { c := streamServer(t, sse( `{"type":"message_start","message":{"id":"m"}}`, `{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t","name":"pwd"}}`, `{"type":"content_block_stop","index":0}`, `{"type":"message_delta","delta":{"stop_reason":"tool_use"}}`, )) resp, err := c.StreamMessage(context.Background(), Request{}, nil) if err != nil { t.Fatalf("StreamMessage: %v", err) } if got := string(resp.ToolUses()[0].Input); got != "{}" { t.Errorf("input = %s, want an empty object", got) } } func TestStreamHTTPErrorIsAnAPIError(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusBadRequest) io.WriteString(w, `{"error":{"type":"invalid_request_error","message":"bad model"}}`) })) defer srv.Close() c := New(WithBaseURL(srv.URL), WithAPIKey("k")) _, err := c.StreamMessage(context.Background(), Request{}, nil) apiErr, ok := err.(*APIError) if !ok { t.Fatalf("error type = %T, want *APIError", err) } if apiErr.StatusCode != 400 || apiErr.Message != "bad model" { t.Errorf("APIError = %+v", apiErr) } }