| 1 | //go:build live |
| 2 | |
| 3 | package agent |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "io" |
| 8 | "net/http" |
| 9 | "net/http/httptest" |
| 10 | "os" |
| 11 | "sync" |
| 12 | "sync/atomic" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "reasonix/internal/provider/responses" |
| 17 | ) |
| 18 | |
| 19 | // TestLiveDeepSeekResponsesAgentToolLoops verifies the official stateless |
| 20 | // endpoint accepts tool turns both with and without a reasoning item. Every |
| 21 | // run must execute the client tool exactly once and reach a visible final. |
| 22 | func TestLiveDeepSeekResponsesAgentToolLoops(t *testing.T) { |
| 23 | key := os.Getenv("DEEPSEEK_API_KEY") |
| 24 | if key == "" { |
| 25 | t.Skip("DEEPSEEK_API_KEY not set") |
| 26 | } |
| 27 | for _, model := range []string{"deepseek-v4-flash", "deepseek-v4-pro"} { |
| 28 | t.Run(model, func(t *testing.T) { |
| 29 | prov := responses.New(responses.Config{ |
| 30 | Name: "deepseek-responses", BaseURL: "https://api.deepseek.com", Model: model, |
| 31 | APIKey: key, KeyEnv: "DEEPSEEK_API_KEY", Effort: "high", Mode: "stateless", MaxOutputTokens: 512, |
| 32 | }) |
| 33 | if closer, ok := prov.(interface{ CloseIdleConnections() }); ok { |
| 34 | t.Cleanup(closer.CloseIdleConnections) |
| 35 | } |
| 36 | retries, recovered := runLiveAgentToolLoops(t, prov, 10) |
| 37 | t.Logf("model=%s runs=10 tool_executions=10 retry_attempts=%d recovered=%d", model, retries, recovered) |
| 38 | }) |
| 39 | } |
| 40 | } |
| 41 | |
| 42 | // TestLiveDeepSeekResponsesMissingReasoningFallback keeps the upstream model |
| 43 | // and stream real while a localhost proxy removes provider reasoning events |
| 44 | // from tool-call responses. This deterministically exercises the compatibility |
| 45 | // fallback without logging model output, tool arguments, or credentials. |
| 46 | func TestLiveDeepSeekResponsesMissingReasoningFallback(t *testing.T) { |
| 47 | key := os.Getenv("DEEPSEEK_API_KEY") |
| 48 | if key == "" { |
| 49 | t.Skip("DEEPSEEK_API_KEY not set") |
| 50 | } |
| 51 | for _, tc := range []struct { |
| 52 | model string |
| 53 | stripResponses int32 |
| 54 | wantRetries int |
| 55 | }{ |
| 56 | {model: "deepseek-v4-flash", stripResponses: 1}, |
| 57 | {model: "deepseek-v4-pro", stripResponses: 1}, |
| 58 | } { |
| 59 | t.Run(tc.model, func(t *testing.T) { |
| 60 | proxy := &liveResponsesReasoningStripProxy{stripResponses: tc.stripResponses} |
| 61 | server := httptest.NewServer(proxy) |
| 62 | defer server.Close() |
| 63 | prov := responses.New(responses.Config{ |
| 64 | Name: "deepseek-responses", BaseURL: "https://api.deepseek.com", RequestURL: server.URL, |
| 65 | Model: tc.model, APIKey: key, KeyEnv: "DEEPSEEK_API_KEY", Effort: "high", Mode: "stateless", MaxOutputTokens: 512, |
| 66 | }) |
| 67 | if closer, ok := prov.(interface{ CloseIdleConnections() }); ok { |
| 68 | t.Cleanup(closer.CloseIdleConnections) |
| 69 | } |
| 70 | retries, recovered := runLiveAgentToolLoops(t, prov, 1) |
| 71 | if retries != tc.wantRetries { |
| 72 | t.Fatalf("reasoning retries = %d, want %d", retries, tc.wantRetries) |
| 73 | } |
| 74 | if proxy.requests.Load() != 2 { |
| 75 | t.Fatalf("upstream requests = %d, want 2", proxy.requests.Load()) |
| 76 | } |
| 77 | if proxy.toolResponses.Load() < tc.stripResponses { |
| 78 | t.Fatalf("tool responses = %d, want at least %d", proxy.toolResponses.Load(), tc.stripResponses) |
| 79 | } |
| 80 | if tc.wantRetries > 0 && !proxy.firstTwoToolRequestsEqual() { |
| 81 | t.Fatal("missing-reasoning retry changed the frozen request") |
| 82 | } |
| 83 | t.Logf("model=%s upstream_requests=%d tool_responses=%d stripped_events=%d retry_attempts=%d recovered=%d exact_retry=%t", |
| 84 | tc.model, proxy.requests.Load(), proxy.toolResponses.Load(), proxy.strippedEvents.Load(), retries, recovered, |
| 85 | proxy.firstTwoToolRequestsEqual()) |
| 86 | }) |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | type liveResponsesReasoningStripProxy struct { |
| 91 | stripResponses int32 |
| 92 | requests atomic.Int32 |
| 93 | toolResponses atomic.Int32 |
| 94 | strippedEvents atomic.Int32 |
| 95 | mu sync.Mutex |
| 96 | toolBodies [][]byte |
| 97 | } |
| 98 | |
| 99 | func (p *liveResponsesReasoningStripProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 100 | body, err := io.ReadAll(r.Body) |
| 101 | if err != nil { |
| 102 | http.Error(w, "read request", http.StatusBadRequest) |
| 103 | return |
| 104 | } |
| 105 | p.requests.Add(1) |
| 106 | upstream, err := http.NewRequestWithContext(r.Context(), http.MethodPost, |
| 107 | "https://api.deepseek.com/responses", bytes.NewReader(body)) |
| 108 | if err != nil { |
| 109 | http.Error(w, "create upstream request", http.StatusInternalServerError) |
| 110 | return |
| 111 | } |
| 112 | upstream.Header.Set("Authorization", r.Header.Get("Authorization")) |
| 113 | upstream.Header.Set("Content-Type", "application/json") |
| 114 | resp, err := (&http.Client{Timeout: 90 * time.Second}).Do(upstream) |
| 115 | if err != nil { |
| 116 | http.Error(w, "upstream request failed", http.StatusBadGateway) |
| 117 | return |
| 118 | } |
| 119 | defer resp.Body.Close() |
| 120 | responseBody, err := io.ReadAll(resp.Body) |
| 121 | if err != nil { |
| 122 | http.Error(w, "read upstream response", http.StatusBadGateway) |
| 123 | return |
| 124 | } |
| 125 | if resp.StatusCode == http.StatusOK && bytes.Contains(responseBody, []byte(`"type":"function_call"`)) { |
| 126 | toolResponse := p.toolResponses.Add(1) |
| 127 | if toolResponse <= p.stripResponses { |
| 128 | p.mu.Lock() |
| 129 | p.toolBodies = append(p.toolBodies, append([]byte(nil), body...)) |
| 130 | p.mu.Unlock() |
| 131 | var stripped int |
| 132 | responseBody, stripped = stripResponsesReasoningEvents(responseBody) |
| 133 | p.strippedEvents.Add(int32(stripped)) |
| 134 | } |
| 135 | } |
| 136 | w.Header().Set("Content-Type", resp.Header.Get("Content-Type")) |
| 137 | w.WriteHeader(resp.StatusCode) |
| 138 | _, _ = w.Write(responseBody) |
| 139 | } |
| 140 | |
| 141 | func (p *liveResponsesReasoningStripProxy) firstTwoToolRequestsEqual() bool { |
| 142 | p.mu.Lock() |
| 143 | defer p.mu.Unlock() |
| 144 | return len(p.toolBodies) >= 2 && bytes.Equal(p.toolBodies[0], p.toolBodies[1]) |
| 145 | } |
| 146 | |
| 147 | func stripResponsesReasoningEvents(body []byte) ([]byte, int) { |
| 148 | lines := bytes.Split(body, []byte("\n")) |
| 149 | out := make([][]byte, 0, len(lines)) |
| 150 | stripped := 0 |
| 151 | for _, line := range lines { |
| 152 | data := bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data:"))) |
| 153 | if bytes.Equal(data, line) || len(data) == 0 { |
| 154 | out = append(out, line) |
| 155 | continue |
| 156 | } |
| 157 | if bytes.Contains(data, []byte(`"type":"response.reasoning`)) || |
| 158 | (bytes.Contains(data, []byte(`"type":"response.output_item`)) && bytes.Contains(data, []byte(`"type":"reasoning"`))) { |
| 159 | if len(out) > 0 && bytes.HasPrefix(bytes.TrimSpace(out[len(out)-1]), []byte("event:")) { |
| 160 | out = out[:len(out)-1] |
| 161 | } |
| 162 | stripped++ |
| 163 | continue |
| 164 | } |
| 165 | out = append(out, line) |
| 166 | } |
| 167 | return bytes.Join(out, []byte("\n")), stripped |
| 168 | } |
| 169 |