1package core
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "reflect"
9 "strings"
10 "sync/atomic"
11 "testing"
12 "time"
13 "unicode/utf8"
14
15 "github.com/bnema/zut/packages/provider"
16)
17
18type retryFakeClient struct {
19 calls int32
20 firstErr error
21}
22
23func (c *retryFakeClient) Name() string { return "retry-fake" }
24
25func (c *retryFakeClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
26 if req.Lifecycle != nil {
27 req.Lifecycle.RequestAttempt(1, 1)
28 }
29 call := atomic.AddInt32(&c.calls, 1)
30 out := make(chan provider.Event, 4)
31 go func() {
32 defer close(out)
33 out <- provider.EventStart{Provider: "retry-fake", Model: req.Model}
34 if call == 1 {
35 err := c.firstErr
36 if err == nil {
37 err = fmt.Errorf("anthropic overloaded_error: Overloaded")
38 }
39 out <- provider.EventDone{Stop: provider.StopError, Err: err}
40 return
41 }
42 out <- provider.EventTextDelta{Delta: "ok"}
43 out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
44 Role: provider.RoleAssistant,
45 Content: []provider.Content{provider.TextBlock{Text: "ok"}},
46 }}
47 }()
48 return out, nil
49}
50
51func TestAgentRetriesOverloadedStreamError(t *testing.T) {
52 client := &retryFakeClient{}
53 a := NewAgent(client, "fake-model", "system", Registry{})
54 a.RetryBaseDelay = time.Millisecond
55
56 var turnErrs []string
57 err := a.Prompt(context.Background(), "hello", nil, func(ev AgentEvent) {
58 if e, ok := ev.(EvTurnEnd); ok && e.Err != nil {
59 turnErrs = append(turnErrs, e.Err.Error())
60 }
61 })
62 if err != nil {
63 t.Fatalf("Prompt returned %v", err)
64 }
65 if got := atomic.LoadInt32(&client.calls); got != 2 {
66 t.Fatalf("Stream calls = %d; want 2", got)
67 }
68 if len(turnErrs) != 1 || !strings.Contains(turnErrs[0], "overloaded_error") {
69 t.Fatalf("turn errors = %v; want one overloaded error before retry", turnErrs)
70 }
71 msgs := a.Messages()
72 if len(msgs) != 2 {
73 t.Fatalf("message count = %d; want user + final assistant", len(msgs))
74 }
75 if got := extractText(msgs[1]); got != "ok" {
76 t.Fatalf("final assistant text = %q; want ok", got)
77 }
78}
79
80func TestAgentRetriesUnexpectedEOFStreamError(t *testing.T) {
81 client := &retryFakeClient{firstErr: fmt.Errorf("read SSE: %w", io.ErrUnexpectedEOF)}
82 a := NewAgent(client, "fake-model", "system", Registry{})
83 a.RetryBaseDelay = time.Millisecond
84
85 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
86 t.Fatalf("Prompt returned %v", err)
87 }
88 if got := atomic.LoadInt32(&client.calls); got != 2 {
89 t.Fatalf("Stream calls = %d; want 2", got)
90 }
91}
92
93func TestAgentEmitsRetryLifecycleEvents(t *testing.T) {
94 client := &retryFakeClient{}
95 a := NewAgent(client, "fake-model", "system", Registry{})
96 a.RetryBaseDelay = time.Millisecond
97
98 var lifecycle []string
99 if err := a.Prompt(context.Background(), "hello", nil, func(ev AgentEvent) {
100 switch e := ev.(type) {
101 case EvRequestStarted:
102 lifecycle = append(lifecycle, fmt.Sprintf("request:%s:%d/%d", e.Scope, e.Attempt, e.MaxAttempts))
103 case EvRetryScheduled:
104 lifecycle = append(lifecycle, fmt.Sprintf("retry:%s:%d/%d", e.Scope, e.Attempt, e.MaxAttempts))
105 }
106 }); err != nil {
107 t.Fatalf("Prompt returned %v", err)
108 }
109
110 want := []string{
111 "request:agent:1/4",
112 "request:provider:1/1",
113 "retry:agent:2/4",
114 "request:agent:2/4",
115 "request:provider:1/1",
116 }
117 if strings.Join(lifecycle, ",") != strings.Join(want, ",") {
118 t.Fatalf("lifecycle = %v, want %v", lifecycle, want)
119 }
120}
121
122func TestAgentReportsSanitizedRetryLifecycle(t *testing.T) {
123 client := &retryFakeClient{}
124 a := NewAgent(client, "fake-model", "system", Registry{})
125 a.RetryBaseDelay = time.Millisecond
126
127 var records []RetryLifecycleRecord
128 a.OnRetryLifecycle = func(record RetryLifecycleRecord) {
129 records = append(records, record)
130 }
131 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
132 t.Fatalf("Prompt returned %v", err)
133 }
134
135 want := []RetryLifecycleRecord{
136 {
137 Event: RetryLifecycleRequestFailed, Scope: RetryScopeAgent,
138 Attempt: 1, MaxAttempts: 4, Reason: RetryReasonOverload,
139 },
140 {
141 Event: RetryLifecycleRetryScheduled, Scope: RetryScopeAgent,
142 Attempt: 2, MaxAttempts: 4, Reason: RetryReasonOverload, DelayMS: 1,
143 },
144 }
145 if !reflect.DeepEqual(records, want) {
146 t.Fatalf("retry lifecycle = %#v, want %#v", records, want)
147 }
148}
149
150func TestRequestLifecycleSinkForwardsProviderFailureReason(t *testing.T) {
151 var records []RetryLifecycleRecord
152 sink := &requestLifecycleSink{
153 sink: func(AgentEvent) {},
154 observe: func(record RetryLifecycleRecord) { records = append(records, record) },
155 }
156 sink.RequestFailed(1, 3, provider.RequestFailureOverload, false)
157 sink.RetryScheduled(2, 3, 250*time.Millisecond)
158
159 want := []RetryLifecycleRecord{
160 {
161 Event: RetryLifecycleRequestFailed, Scope: RetryScopeProvider,
162 Attempt: 1, MaxAttempts: 3, Reason: RetryReasonOverload,
163 },
164 {
165 Event: RetryLifecycleRetryScheduled, Scope: RetryScopeProvider,
166 Attempt: 2, MaxAttempts: 3, Reason: RetryReasonOverload, DelayMS: 250,
167 },
168 }
169 if !reflect.DeepEqual(records, want) {
170 t.Fatalf("provider retry lifecycle = %#v, want %#v", records, want)
171 }
172}
173
174func TestAgentReportsTerminalRetryFailure(t *testing.T) {
175 client := &retryFakeClient{}
176 a := NewAgent(client, "fake-model", "system", Registry{})
177 a.MaxRetries = 0
178
179 var records []RetryLifecycleRecord
180 a.OnRetryLifecycle = func(record RetryLifecycleRecord) {
181 records = append(records, record)
182 }
183 if err := a.Prompt(context.Background(), "hello", nil, nil); err == nil {
184 t.Fatal("Prompt succeeded; want terminal provider error")
185 }
186 if len(records) != 1 {
187 t.Fatalf("retry lifecycle = %#v, want one terminal record", records)
188 }
189 record := records[0]
190 if record.Event != RetryLifecycleRequestFailed || record.Scope != RetryScopeAgent ||
191 record.Attempt != 1 || record.MaxAttempts != 1 || record.Reason != RetryReasonOverload || !record.Terminal {
192 t.Fatalf("terminal retry record = %#v", record)
193 }
194}
195
196// codexRetryFakeClient reproduces a transient OpenAI Codex backend
197// failure on the first call and succeeds afterwards.
198type codexRetryFakeClient struct {
199 calls int32
200 firstErr string
201}
202
203func (c *codexRetryFakeClient) Name() string { return "openai-codex" }
204
205func (c *codexRetryFakeClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
206 call := atomic.AddInt32(&c.calls, 1)
207 out := make(chan provider.Event, 4)
208 go func() {
209 defer close(out)
210 out <- provider.EventStart{Provider: "openai-codex", Model: req.Model}
211 if call == 1 {
212 out <- provider.EventDone{Stop: provider.StopError, Err: fmt.Errorf("%s", c.firstErr)}
213 return
214 }
215 out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
216 Role: provider.RoleAssistant,
217 Content: []provider.Content{provider.TextBlock{Text: "ok"}},
218 }}
219 }()
220 return out, nil
221}
222
223func TestAgentRetriesCodexProcessingError(t *testing.T) {
224 cases := []struct {
225 name string
226 err string
227 }{
228 {
229 name: "processing error",
230 err: "codex error: An error occurred while processing your request. You can retry your request, or contact us through our help center at help.openai.com if the error persists. Please include the request ID 60c8ebbd-20bd-42e4-b756-6e844041cfc0 in your message.",
231 },
232 {
233 name: "servers overloaded",
234 err: "codex error: Our servers are currently overloaded. Please try again later.",
235 },
236 {
237 name: "try again later only",
238 err: "codex error: Something went wrong. Please try again later.",
239 },
240 }
241 for _, tc := range cases {
242 t.Run(tc.name, func(t *testing.T) {
243 client := &codexRetryFakeClient{firstErr: tc.err}
244 a := NewAgent(client, "gpt-5.6-sol", "system", Registry{})
245 a.RetryBaseDelay = time.Millisecond
246
247 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
248 t.Fatalf("Prompt returned %v; want retry to succeed", err)
249 }
250 if got := atomic.LoadInt32(&client.calls); got != 2 {
251 t.Fatalf("Stream calls = %d; want 2", got)
252 }
253 msgs := a.Messages()
254 if len(msgs) != 2 || extractText(msgs[1]) != "ok" {
255 t.Fatalf("messages = %v; want user + ok assistant", msgs)
256 }
257 })
258 }
259}
260
261// TestCanRetryErrorCapacityMessages pins the classification of Codex
262// capacity wording and makes sure quota/usage limits stay terminal even
263// when they carry "try again later" style advice.
264func TestCanRetryErrorCapacityMessages(t *testing.T) {
265 a := NewAgent(nil, "gpt-5.6-sol", "system", Registry{})
266 cases := []struct {
267 msg string
268 want bool
269 }{
270 {"codex error: Our servers are currently overloaded. Please try again later.", true},
271 {"codex error: Our servers are busy right now.", true},
272 {"codex error: Please try again later.", true},
273 {"codex error: You have hit your monthly usage limit. Try again later.", false},
274 {"codex error: quota exceeded, try again later", false},
275 {"codex error: unsupported parameter: reasoning", false},
276 }
277 for _, tc := range cases {
278 if got := a.canRetryError(errors.New(tc.msg), 0); got != tc.want {
279 t.Errorf("canRetryError(%q) = %v; want %v", tc.msg, got, tc.want)
280 }
281 }
282}
283
284type partialRetryFakeClient struct {
285 calls int32
286}
287
288func (c *partialRetryFakeClient) Name() string { return "partial-retry-fake" }
289
290func (c *partialRetryFakeClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
291 call := atomic.AddInt32(&c.calls, 1)
292 out := make(chan provider.Event, 4)
293 go func() {
294 defer close(out)
295 out <- provider.EventStart{Provider: "partial-retry-fake", Model: req.Model}
296 if call == 1 {
297 out <- provider.EventTextDelta{Delta: "partial"}
298 out <- provider.EventDone{Stop: provider.StopError, Err: fmt.Errorf("provider returned error: 503"), Message: provider.Message{
299 Role: provider.RoleAssistant,
300 Content: []provider.Content{provider.TextBlock{Text: "partial"}},
301 }}
302 return
303 }
304 out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
305 Role: provider.RoleAssistant,
306 Content: []provider.Content{provider.TextBlock{Text: "recovered"}},
307 }}
308 }()
309 return out, nil
310}
311
312func TestAgentDropsPartialAssistantBeforeRetry(t *testing.T) {
313 client := &partialRetryFakeClient{}
314 a := NewAgent(client, "fake-model", "system", Registry{})
315 a.RetryBaseDelay = time.Millisecond
316
317 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
318 t.Fatalf("Prompt returned %v", err)
319 }
320 msgs := a.Messages()
321 if len(msgs) != 2 {
322 t.Fatalf("message count = %d; want user + recovered assistant", len(msgs))
323 }
324 if got := extractText(msgs[1]); got != "recovered" {
325 t.Fatalf("final assistant text = %q; want recovered", got)
326 }
327}
328
329// captureClient records the last Request it received so tests can
330// assert what the agent put on the wire.
331type captureClient struct {
332 lastReq provider.Request
333}
334
335func (c *captureClient) Name() string { return "capture" }
336
337func (c *captureClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
338 c.lastReq = req
339 out := make(chan provider.Event, 3)
340 go func() {
341 defer close(out)
342 out <- provider.EventStart{Provider: "capture", Model: req.Model}
343 out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
344 Role: provider.RoleAssistant,
345 Content: []provider.Content{provider.TextBlock{Text: "ok"}},
346 }}
347 }()
348 return out, nil
349}
350
351func TestAgentInjectsHiddenTurnContextIntoRequestOnly(t *testing.T) {
352 client := &captureClient{}
353 a := NewAgent(client, "fake-model", "system", Registry{})
354 a.BeforeTurnContext = func(context.Context, int) (bool, string, string) {
355 return true, "", "current phase: parse files"
356 }
357 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
358 t.Fatalf("Prompt returned %v", err)
359 }
360 if !strings.Contains(client.lastReq.System, "current phase: parse files") {
361 t.Fatalf("request system prompt = %q; missing hidden context", client.lastReq.System)
362 }
363 if len(a.Messages()) != 2 {
364 t.Fatalf("transcript message count = %d; hidden context must not be persisted", len(a.Messages()))
365 }
366 for _, msg := range a.Messages() {
367 if strings.Contains(extractText(msg), "current phase: parse files") {
368 t.Fatal("hidden turn context leaked into transcript")
369 }
370 }
371}
372
373func TestBoundedTurnContextPreservesUTF8AndLimit(t *testing.T) {
374 contextText := strings.Repeat("界", maxTurnContextBytes)
375 got := boundedTurnContext(contextText)
376 if len(got) > maxTurnContextBytes {
377 t.Fatalf("bounded context bytes = %d, want <= %d", len(got), maxTurnContextBytes)
378 }
379 if !utf8.ValidString(got) {
380 t.Fatal("bounded context is not valid UTF-8")
381 }
382 if !strings.HasSuffix(got, turnContextTruncatedMarker) {
383 tail := got
384 if len(tail) > 64 {
385 tail = tail[len(tail)-64:]
386 }
387 t.Fatalf("bounded context missing truncation marker: %q", tail)
388 }
389}
390
391func TestAgentPropagatesMaxTokens(t *testing.T) {
392 client := &captureClient{}
393 a := NewAgent(client, "fake-model", "system", Registry{})
394 a.MaxTokens = 64000
395
396 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
397 t.Fatalf("Prompt returned %v", err)
398 }
399 if client.lastReq.MaxTokens != 64000 {
400 t.Fatalf("request MaxTokens = %d; want 64000 (Agent.MaxTokens not propagated)", client.lastReq.MaxTokens)
401 }
402}
403
404func TestAgentPropagatesTemperature(t *testing.T) {
405 client := &captureClient{}
406 a := NewAgent(client, "fake-model", "system", Registry{})
407 temp := float32(0)
408 a.Temperature = &temp
409
410 if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
411 t.Fatalf("Prompt returned %v", err)
412 }
413 if client.lastReq.Temperature == nil || *client.lastReq.Temperature != temp {
414 t.Fatalf("request Temperature = %v; want %v", client.lastReq.Temperature, temp)
415 }
416}
417
418type cancelledRetryClient struct {
419 cancel context.CancelFunc
420}
421
422func (c *cancelledRetryClient) Name() string { return "cancelled-retry" }
423
424func (c *cancelledRetryClient) Stream(_ context.Context, _ provider.Request) (<-chan provider.Event, error) {
425 out := make(chan provider.Event, 1)
426 c.cancel()
427 out <- provider.EventDone{Stop: provider.StopError, Err: errors.New("provider overloaded")}
428 close(out)
429 return out, nil
430}
431
432func TestAgentDoesNotReportRetryAfterCancellation(t *testing.T) {
433 ctx, cancel := context.WithCancel(context.Background())
434 client := &cancelledRetryClient{cancel: cancel}
435 agent := NewAgent(client, "fake-model", "system", Registry{})
436 agent.MaxRetries = 1
437
438 var retries []EvRetryScheduled
439 var lifecycle []RetryLifecycleRecord
440 agent.OnRetryLifecycle = func(record RetryLifecycleRecord) {
441 lifecycle = append(lifecycle, record)
442 }
443 err := agent.Prompt(ctx, "hello", nil, func(event AgentEvent) {
444 if retry, ok := event.(EvRetryScheduled); ok {
445 retries = append(retries, retry)
446 }
447 })
448 if !errors.Is(err, context.Canceled) {
449 t.Fatalf("Prompt error = %v, want context.Canceled", err)
450 }
451 if len(retries) != 0 {
452 t.Fatalf("retry events = %#v, want none after cancellation", retries)
453 }
454 if len(lifecycle) != 1 || lifecycle[0].Event != RetryLifecycleRequestFailed || !lifecycle[0].Terminal {
455 t.Fatalf("retry lifecycle = %#v, want one terminal failure after cancellation", lifecycle)
456 }
457}