| 1 | //go:build live |
| 2 | |
| 3 | package agent |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "context" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "net/http/httptest" |
| 12 | "os" |
| 13 | "path/filepath" |
| 14 | "strings" |
| 15 | "sync/atomic" |
| 16 | "testing" |
| 17 | "time" |
| 18 | |
| 19 | "reasonix/internal/event" |
| 20 | "reasonix/internal/provider" |
| 21 | "reasonix/internal/provider/anthropic" |
| 22 | "reasonix/internal/provider/openai" |
| 23 | "reasonix/internal/provider/responses" |
| 24 | "reasonix/internal/tool" |
| 25 | ) |
| 26 | |
| 27 | type multiProviderCase struct{ vendor, keyEnv, model, protocol, base, thinking, effort, reasoning string } |
| 28 | |
| 29 | // Every credential is restricted to one documented vendor endpoint. Only echo |
| 30 | // is exposed to the model, and no request/response body or credential is logged. |
| 31 | func multiProviderCases() []multiProviderCase { |
| 32 | var out []multiProviderCase |
| 33 | add := func(vendor, env, base, protocol, thinking, effort, reasoning string, models ...string) { |
| 34 | for _, model := range models { |
| 35 | out = append(out, multiProviderCase{vendor, env, model, protocol, base, thinking, effort, reasoning}) |
| 36 | } |
| 37 | } |
| 38 | add("longcat", "LONGCAT_API_KEY", "https://api.longcat.chat/openai/v1", "chat", "enabled", "enabled", "", "LongCat-2.0") |
| 39 | add("longcat", "LONGCAT_API_KEY", "https://api.longcat.chat/anthropic", "anthropic", "enabled", "enabled", "", "LongCat-2.0") |
| 40 | glmModels := []string{"glm-5.3-flash", "glm-5.3", "glm-5.2", "glm-5.1", "glm-5", "glm-4.7", "glm-4.5-air"} |
| 41 | add("glm", "GLM_PLAN_API_KEY", "https://open.bigmodel.cn/api/coding/paas/v4", "chat", "", "", "glm", glmModels...) |
| 42 | add("glm", "GLM_PLAN_API_KEY", "https://open.bigmodel.cn/api/anthropic", "anthropic", "adaptive", "", "", glmModels...) |
| 43 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go/v1", "chat", "", "low", "openai", "glm-5.3-flash", "glm-5.3", "glm-5.1", "kimi-k3", "kimi-k2.7-code", "kimi-k2.6", "hy4-preview", "hy3") |
| 44 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go/v1", "chat", "", "high", "openai", "glm-5.2") |
| 45 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go/v1", "chat", "", "", "", "longcat-2.0", "mimo-v2.5", "mimo-v2.5-pro", "omen-alpha") |
| 46 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go/v1", "chat", "enabled", "high", "deepseek", "deepseek-v4-flash", "deepseek-v4-pro", "deepseek-v4-flash-vision-exp") |
| 47 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go", "anthropic", "adaptive", "", "", "minimax-m3", "minimax-m2.7", "qwen3.8-max", "qwen3.8-flash", "qwen3.7-max", "qwen3.7-plus", "qwen3.6-plus") |
| 48 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go/v1", "responses", "", "low", "", "gpt-5.6-luna", "grok-4.6", "muse-spark-1.3-contributor", "muse-spark-1.2-contributor") |
| 49 | // Custom DeepSeek protocol entries discussed in #9808, distinct from Go's |
| 50 | // recommended Chat route. Availability is measured, never assumed. |
| 51 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go", "anthropic", "adaptive", "high", "deepseek", "deepseek-v4-flash", "deepseek-v4-pro") |
| 52 | add("go", "OPENCODE_GO_API_KEY", "https://opencode.ai/zen/go/v1", "responses", "", "high", "deepseek", "deepseek-v4-flash", "deepseek-v4-pro") |
| 53 | // Flash/Pro already have full official recovery coverage; add vision model's |
| 54 | // text/tool path here without claiming image understanding was tested. |
| 55 | for _, proto := range []string{"chat", "anthropic", "responses"} { |
| 56 | base := "https://api.deepseek.com" |
| 57 | if proto == "anthropic" { |
| 58 | base += "/anthropic" |
| 59 | } |
| 60 | add("deepseek", "DEEPSEEK_API_KEY", base, proto, "enabled", "high", "deepseek", "deepseek-v4-flash-vision-exp") |
| 61 | } |
| 62 | return out |
| 63 | } |
| 64 | |
| 65 | func (tc multiProviderCase) upstream() string { |
| 66 | suffix := map[string]string{"chat": "/chat/completions", "anthropic": "/v1/messages", "responses": "/responses"}[tc.protocol] |
| 67 | return tc.base + suffix |
| 68 | } |
| 69 | func (tc multiProviderCase) new(t *testing.T, url, scenario string) provider.Provider { |
| 70 | t.Helper() |
| 71 | key := os.Getenv(tc.keyEnv) |
| 72 | if effort := os.Getenv("REASONIX_LIVE_EFFORT"); effort != "" { |
| 73 | tc.effort = effort |
| 74 | } |
| 75 | extra := map[string]any{"api_key_env": tc.keyEnv, "request_url": url, "reject_redirects": true} |
| 76 | if scenario == "search" { |
| 77 | extra["web_search"] = true |
| 78 | } |
| 79 | if scenario == "vision" { |
| 80 | extra["vision"] = true |
| 81 | } |
| 82 | if tc.vendor == "longcat" || tc.vendor == "glm" { |
| 83 | extra["auth_header"] = true |
| 84 | } |
| 85 | if tc.reasoning != "" { |
| 86 | extra["reasoning_protocol"] = tc.reasoning |
| 87 | } |
| 88 | if tc.thinking != "" { |
| 89 | extra["thinking"] = tc.thinking |
| 90 | } |
| 91 | if tc.effort != "" { |
| 92 | extra["effort"] = tc.effort |
| 93 | } |
| 94 | if scenario == "disabled" { |
| 95 | extra["thinking"] = "disabled" |
| 96 | extra["effort"] = "disabled" |
| 97 | } |
| 98 | var p provider.Provider |
| 99 | var err error |
| 100 | switch tc.protocol { |
| 101 | case "chat": |
| 102 | p, err = openai.New(provider.Config{Name: "live-" + tc.vendor, BaseURL: tc.base, Model: tc.model, APIKey: key, Extra: extra}) |
| 103 | case "anthropic": |
| 104 | p, err = anthropic.New(provider.Config{Name: "live-" + tc.vendor, BaseURL: tc.base, Model: tc.model, APIKey: key, Extra: extra}) |
| 105 | case "responses": |
| 106 | effort := tc.effort |
| 107 | if scenario == "disabled" { |
| 108 | effort = "none" |
| 109 | } |
| 110 | p = responses.New(responses.Config{Name: "live-" + tc.vendor, BaseURL: tc.base, Model: tc.model, APIKey: key, KeyEnv: tc.keyEnv, Effort: effort, Mode: "stateless", MaxOutputTokens: 4096, RequestURL: url, Extra: extra, WebSearch: scenario == "search"}) |
| 111 | } |
| 112 | if err != nil { |
| 113 | t.Fatal(err) |
| 114 | } |
| 115 | if c, ok := p.(interface{ CloseIdleConnections() }); ok { |
| 116 | t.Cleanup(c.CloseIdleConnections) |
| 117 | } |
| 118 | return p |
| 119 | } |
| 120 | |
| 121 | func TestLiveMultiProviderMatrix(t *testing.T) { |
| 122 | scenarios := strings.Split(os.Getenv("REASONIX_LIVE_SCENARIOS"), ",") |
| 123 | if len(scenarios) == 1 && scenarios[0] == "" { |
| 124 | scenarios = []string{"baseline"} |
| 125 | } |
| 126 | blocked := map[string]string{} |
| 127 | for _, tc := range multiProviderCases() { |
| 128 | if os.Getenv(tc.keyEnv) == "" { |
| 129 | continue |
| 130 | } |
| 131 | for _, scenario := range scenarios { |
| 132 | t.Run(tc.vendor+"/"+tc.model+"/"+tc.protocol+"/"+scenario, func(t *testing.T) { |
| 133 | if reason := blocked[tc.vendor+"/"+tc.protocol]; reason != "" { |
| 134 | t.Skip("earlier credential/quota gate: " + reason) |
| 135 | } |
| 136 | runMultiProviderCase(t, tc, scenario, blocked) |
| 137 | }) |
| 138 | } |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | func runMultiProviderCase(t *testing.T, tc multiProviderCase, scenario string, blocked map[string]string) { |
| 143 | ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) |
| 144 | defer cancel() |
| 145 | proxy := &officialRecoveryProxy{protocol: tc.protocol, scenario: scenario, cancel: cancel, upstreamURL: tc.upstream()} |
| 146 | |
| 147 | srv := httptest.NewServer(proxy) |
| 148 | defer srv.Close() |
| 149 | p := tc.new(t, srv.URL, scenario) |
| 150 | var executions atomic.Int32 |
| 151 | reg := tool.NewRegistry() |
| 152 | reg.Add(liveRecoveryEchoTool{executions: &executions}) |
| 153 | sink := &recordSink{} |
| 154 | system := "Call echo exactly once for each new user request, then report its fixed marker. Do not repeat completed work. Be concise." |
| 155 | if os.Getenv("REASONIX_LIVE_PROMPT_PROFILE") == "action-evidence" { |
| 156 | // A prompt-only experiment inspired by OpenCode's Kimi-specific action |
| 157 | // instructions. This does not run OpenCode or change Reasonix defaults. |
| 158 | system += " When the user requests a tool action, perform it using the provided tool instead of describing or simulating it. You cannot know this tool's result before executing it. Report only the actual returned result, and never invent a successful execution." |
| 159 | } |
| 160 | sess := NewSession(system) |
| 161 | opts := Options{MaxSteps: 4, MaxOutputTokens: 4096, MissingReasoningWarnStateDir: t.TempDir()} |
| 162 | a := New(p, reg, sess, opts, sink) |
| 163 | sessionPath := filepath.Join(t.TempDir(), "session.jsonl") |
| 164 | a.SetSessionPath(sessionPath) |
| 165 | start := time.Now() |
| 166 | err := a.Run(ctx, "Call echo exactly once, then report its result.") |
| 167 | rounds := 1 |
| 168 | if scenario == "continuity" && err == nil { |
| 169 | path := sessionPath |
| 170 | lease, e := TryAcquireSessionLease(path) |
| 171 | if e != nil { |
| 172 | t.Fatal(e) |
| 173 | } |
| 174 | defer lease.Release() |
| 175 | if e = sess.Save(path); e != nil { |
| 176 | t.Fatal(e) |
| 177 | } |
| 178 | sess, e = LoadSession(path) |
| 179 | if e != nil { |
| 180 | t.Fatal(e) |
| 181 | } |
| 182 | a = New(p, reg, sess, opts, sink) |
| 183 | a.SetSessionPath(path) |
| 184 | for i := 2; i <= 3; i++ { |
| 185 | rounds = i |
| 186 | err = a.Run(ctx, fmt.Sprintf("New request %d: call echo once and report its result. Earlier requests are complete.", i)) |
| 187 | if err != nil { |
| 188 | break |
| 189 | } |
| 190 | } |
| 191 | } |
| 192 | proxy.mu.Lock() |
| 193 | requests, upstream, mutations := proxy.requests, proxy.upstream, proxy.mutations |
| 194 | bodies := append([][]byte(nil), proxy.bodies...) |
| 195 | statuses := append([]int(nil), proxy.statuses...) |
| 196 | wireTools := append([]string(nil), proxy.wireTools...) |
| 197 | wireStops := append([]string(nil), proxy.wireStops...) |
| 198 | proxy.mu.Unlock() |
| 199 | prompt, completion, cached, accounted := 0, 0, 0, 0 |
| 200 | unknown := false |
| 201 | for _, e := range sink.kinds(event.Usage) { |
| 202 | if u := e.Usage; u != nil { |
| 203 | prompt += u.PromptTokens |
| 204 | completion += u.CompletionTokens |
| 205 | cached += u.CacheHitTokens |
| 206 | accounted += u.RequestCount |
| 207 | unknown = unknown || u.Unknown |
| 208 | } |
| 209 | } |
| 210 | errorText := "" |
| 211 | if err != nil { |
| 212 | errorText = err.Error() |
| 213 | failure := provider.ClassifyRecovery(err) |
| 214 | if failure.Phase == "quota" { |
| 215 | blocked[tc.vendor+"/"+tc.protocol] = failure.Phase |
| 216 | } |
| 217 | } |
| 218 | reasoningBytes, thinkingBlocks, responseItems := 0, 0, 0 |
| 219 | for _, m := range sess.Snapshot() { |
| 220 | reasoningBytes += len(m.ReasoningContent) |
| 221 | thinkingBlocks += len(m.ThinkingBlocks) |
| 222 | responseItems += len(m.ResponsesItems) |
| 223 | } |
| 224 | metric := map[string]any{"wire_tools": wireTools, "wire_stops": wireStops, "reasoning_bytes": reasoningBytes, "thinking_blocks": thinkingBlocks, "response_items": responseItems, "provider": tc.vendor, "model": tc.model, "protocol": tc.protocol, "scenario": scenario, "requests": requests, "upstream": upstream, "statuses": statuses, "mutations": mutations, "tools": executions.Load(), "rounds": rounds, "retries": len(sink.kinds(event.Retrying)), "prompt": prompt, "completion": completion, "cache_hit": cached, "accounted": accounted, "unknown_usage": unknown, "elapsed_ms": time.Since(start).Milliseconds(), "error": errorText} |
| 225 | b, _ := json.Marshal(metric) |
| 226 | t.Logf("METRIC %s", b) |
| 227 | if scenario == "cancel_before_commit" { |
| 228 | if !errors.Is(err, context.Canceled) || executions.Load() != 0 { |
| 229 | t.Errorf("cancellation boundary: err=%v executions=%d", err, executions.Load()) |
| 230 | } |
| 231 | return |
| 232 | } |
| 233 | if scenario == "cut_once" { |
| 234 | if err == nil || requests != 1 || mutations != 1 || executions.Load() != 0 || accounted != 1 || len(sink.kinds(event.Retrying)) != 0 { |
| 235 | t.Fatalf("cut stream must stop without retry: requests=%d mutations=%d executions=%d accounted=%d err=%v", requests, mutations, executions.Load(), accounted, err) |
| 236 | } |
| 237 | return |
| 238 | } |
| 239 | if err != nil { |
| 240 | t.Fatalf("live provider run: %v", err) |
| 241 | } |
| 242 | if executions.Load() != int32(rounds) { |
| 243 | for _, m := range sess.Snapshot() { |
| 244 | if m.Role == provider.RoleAssistant && m.Content != "" { |
| 245 | text := m.Content |
| 246 | if len(text) > 1024 { |
| 247 | text = text[:1024] |
| 248 | } |
| 249 | t.Logf("visible_assistant=%q", text) |
| 250 | } |
| 251 | } |
| 252 | t.Errorf("tool executions=%d want=%d", executions.Load(), rounds) |
| 253 | } |
| 254 | messages := sess.Snapshot() |
| 255 | if len(messages) == 0 || strings.TrimSpace(messages[len(messages)-1].Content) == "" { |
| 256 | t.Error("missing final content") |
| 257 | } |
| 258 | if accounted != requests { |
| 259 | t.Errorf("usage request count=%d want=%d", accounted, requests) |
| 260 | } |
| 261 | if scenario == "server_replay_rejection" { |
| 262 | rejected := false |
| 263 | for _, status := range statuses { |
| 264 | rejected = rejected || status == 400 |
| 265 | } |
| 266 | if !rejected { |
| 267 | t.Skip("upstream accepted modified replay; no rejection recovery exercised") |
| 268 | } |
| 269 | } |
| 270 | if strings.HasPrefix(scenario, "missing") && mutations == 0 { |
| 271 | t.Skip("endpoint produced no reasoning: missing-reasoning fault was not exercised") |
| 272 | } |
| 273 | if scenario == "continuity" { |
| 274 | checkMultiProviderPrefix(t, tc.protocol, bodies) |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | func checkMultiProviderPrefix(t *testing.T, protocol string, bodies [][]byte) { |
| 279 | t.Helper() |
| 280 | var previous map[string]json.RawMessage |
| 281 | for n, body := range bodies { |
| 282 | var current map[string]json.RawMessage |
| 283 | if err := json.Unmarshal(body, ¤t); err != nil { |
| 284 | t.Fatal(err) |
| 285 | } |
| 286 | if n > 0 { |
| 287 | for _, field := range []string{"tools", "system", "model", "thinking", "reasoning", "output_config"} { |
| 288 | if !bytes.Equal(previous[field], current[field]) { |
| 289 | t.Errorf("request %d changed %s", n+1, field) |
| 290 | } |
| 291 | } |
| 292 | field := "messages" |
| 293 | if protocol == "responses" { |
| 294 | field = "input" |
| 295 | } |
| 296 | var before, after []json.RawMessage |
| 297 | if err := json.Unmarshal(previous[field], &before); err != nil { |
| 298 | t.Fatal(err) |
| 299 | } |
| 300 | if err := json.Unmarshal(current[field], &after); err != nil { |
| 301 | t.Fatal(err) |
| 302 | } |
| 303 | if len(after) < len(before) { |
| 304 | t.Errorf("request %d lost prefix", n+1) |
| 305 | } else { |
| 306 | for j := range before { |
| 307 | if !bytes.Equal(before[j], after[j]) && !(protocol == "anthropic" && j == len(before)-1 && equalAfterMovingTailCacheMarker(before[j], after[j])) { |
| 308 | t.Errorf("request %d changed history %d", n+1, j) |
| 309 | } |
| 310 | } |
| 311 | } |
| 312 | } |
| 313 | previous = current |
| 314 | } |
| 315 | } |
| 316 | |
| 317 | func TestLiveMultiProviderWriteResume(t *testing.T) { |
| 318 | for _, tc := range multiProviderCases() { |
| 319 | if os.Getenv(tc.keyEnv) == "" { |
| 320 | continue |
| 321 | } |
| 322 | t.Run(tc.vendor+"/"+tc.model+"/"+tc.protocol, func(t *testing.T) { |
| 323 | p := tc.new(t, "", "baseline") |
| 324 | runLiveWriteAfterEffectResume(t, p, tc.vendor+"/"+tc.model+"/"+tc.protocol) |
| 325 | }) |
| 326 | } |
| 327 | } |
| 328 | |
| 329 | // Anthropic moves the ephemeral breakpoint from the old request tail to the |
| 330 | // newly appended tail. Only that exact field on the old final content block |
| 331 | // may differ; tool output, reasoning, signatures and other blocks stay exact. |
| 332 | func equalAfterMovingTailCacheMarker(before, after json.RawMessage) bool { |
| 333 | normalize := func(raw json.RawMessage) ([]byte, bool) { |
| 334 | var message map[string]json.RawMessage |
| 335 | if json.Unmarshal(raw, &message) != nil { |
| 336 | return nil, false |
| 337 | } |
| 338 | var blocks []map[string]json.RawMessage |
| 339 | if json.Unmarshal(message["content"], &blocks) != nil || len(blocks) == 0 { |
| 340 | return nil, false |
| 341 | } |
| 342 | last := blocks[len(blocks)-1] |
| 343 | if marker, ok := last["cache_control"]; ok { |
| 344 | var control map[string]string |
| 345 | if json.Unmarshal(marker, &control) != nil || len(control) != 1 || control["type"] != "ephemeral" { |
| 346 | return nil, false |
| 347 | } |
| 348 | delete(last, "cache_control") |
| 349 | } |
| 350 | content, err := json.Marshal(blocks) |
| 351 | if err != nil { |
| 352 | return nil, false |
| 353 | } |
| 354 | message["content"] = content |
| 355 | out, err := json.Marshal(message) |
| 356 | return out, err == nil |
| 357 | } |
| 358 | a, ok := normalize(before) |
| 359 | b, ok2 := normalize(after) |
| 360 | return ok && ok2 && bytes.Equal(a, b) |
| 361 | } |
| 362 | |
| 363 | func TestLiveTailCacheComparisonPreservesProofAndToolBytes(t *testing.T) { |
| 364 | before := json.RawMessage(`{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-1","content":"marker-alpha","cache_control":{"type":"ephemeral"}}]}`) |
| 365 | after := json.RawMessage(`{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-1","content":"marker-alpha"}]}`) |
| 366 | if !equalAfterMovingTailCacheMarker(before, after) { |
| 367 | t.Fatal("valid tail marker move rejected") |
| 368 | } |
| 369 | for _, bad := range []string{ |
| 370 | `{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-2","content":"marker-alpha"}]}`, |
| 371 | `{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-1","content":"different"}]}`, |
| 372 | `{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-1","content":"marker-alpha","cache_control":{"type":"ephemeral","ttl":"1h"}}]}`, |
| 373 | } { |
| 374 | if equalAfterMovingTailCacheMarker(before, json.RawMessage(bad)) { |
| 375 | t.Fatal("changed tool data or nonstandard marker accepted") |
| 376 | } |
| 377 | } |
| 378 | thinking := json.RawMessage(`{"role":"assistant","content":[{"type":"thinking","thinking":"proof","signature":"sig","cache_control":{"type":"ephemeral"}}]}`) |
| 379 | altered := json.RawMessage(`{"role":"assistant","content":[{"type":"thinking","thinking":"proof","signature":"changed"}]}`) |
| 380 | if equalAfterMovingTailCacheMarker(thinking, altered) { |
| 381 | t.Fatal("signature mutation hidden") |
| 382 | } |
| 383 | } |
| 384 |