fix(sessions): preserve complete retry lifecycle
- commit
- d94de73934ef32ee9dc506020a54c28f9c9f454f
- author
- brice · · 2026-08-10 10:13 UTC
- parent
- 387ed88
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +132 -13
M packages/agent/cli.go +9 -0
| @@ -989,6 +989,9 @@ func runPrintMode(ctx context.Context, args Args, version string) (runErr error) | |||
| 989 | 989 | } | |
| 990 | 990 | announceSession(extMgr, sess) | |
| 991 | 991 | wireRetryLifecyclePersistence(ag, sess) | |
| 992 | + | if sess != nil { | |
| 993 | + | defer func() { runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) }() | |
| 994 | + | } | |
| 992 | 995 | ||
| 993 | 996 | prompt := args.Prompt | |
| 994 | 997 | if prompt == "" { | |
| @@ -1071,6 +1074,9 @@ func runStreamMode(ctx context.Context, args Args, version string) (runErr error | |||
| 1071 | 1074 | } | |
| 1072 | 1075 | announceSession(extMgr, sess) | |
| 1073 | 1076 | wireRetryLifecyclePersistence(ag, sess) | |
| 1077 | + | if sess != nil { | |
| 1078 | + | defer func() { runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) }() | |
| 1079 | + | } | |
| 1074 | 1080 | ||
| 1075 | 1081 | prompt := args.Prompt | |
| 1076 | 1082 | if prompt == "" { | |
| @@ -1192,6 +1198,9 @@ func runJSONMode(ctx context.Context, args Args, version string) (runErr error) | |||
| 1192 | 1198 | } | |
| 1193 | 1199 | announceSession(extMgr, sess) | |
| 1194 | 1200 | wireRetryLifecyclePersistence(ag, sess) | |
| 1201 | + | if sess != nil { | |
| 1202 | + | defer func() { runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) }() | |
| 1203 | + | } | |
| 1195 | 1204 | ||
| 1196 | 1205 | prompt := args.Prompt | |
| 1197 | 1206 | if prompt == "" { | |
| @@ -989,6 +989,9 @@ func runPrintMode(ctx context.Context, args Args, version string) (runErr error) | |||||
| 989 | } | 989 | } | ||
| 990 | announceSession(extMgr, sess) | 990 | announceSession(extMgr, sess) | ||
| 991 | wireRetryLifecyclePersistence(ag, sess) | 991 | wireRetryLifecyclePersistence(ag, sess) | ||
| 992 | + | if sess != nil { | |||
| 993 | + | defer func() { runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) }() | |||
| 994 | + | } | |||
| 992 | 995 | ||||
| 993 | prompt := args.Prompt | 996 | prompt := args.Prompt | ||
| 994 | if prompt == "" { | 997 | if prompt == "" { | ||
| @@ -1071,6 +1074,9 @@ func runStreamMode(ctx context.Context, args Args, version string) (runErr error | |||||
| 1071 | } | 1074 | } | ||
| 1072 | announceSession(extMgr, sess) | 1075 | announceSession(extMgr, sess) | ||
| 1073 | wireRetryLifecyclePersistence(ag, sess) | 1076 | wireRetryLifecyclePersistence(ag, sess) | ||
| 1077 | + | if sess != nil { | |||
| 1078 | + | defer func() { runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) }() | |||
| 1079 | + | } | |||
| 1074 | 1080 | ||||
| 1075 | prompt := args.Prompt | 1081 | prompt := args.Prompt | ||
| 1076 | if prompt == "" { | 1082 | if prompt == "" { | ||
| @@ -1192,6 +1198,9 @@ func runJSONMode(ctx context.Context, args Args, version string) (runErr error) | |||||
| 1192 | } | 1198 | } | ||
| 1193 | announceSession(extMgr, sess) | 1199 | announceSession(extMgr, sess) | ||
| 1194 | wireRetryLifecyclePersistence(ag, sess) | 1200 | wireRetryLifecyclePersistence(ag, sess) | ||
| 1201 | + | if sess != nil { | |||
| 1202 | + | defer func() { runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) }() | |||
| 1203 | + | } | |||
| 1195 | 1204 | ||||
| 1196 | prompt := args.Prompt | 1205 | prompt := args.Prompt | ||
| 1197 | if prompt == "" { | 1206 | if prompt == "" { | ||
M packages/agent/cli_session_test.go +27 -6
| @@ -24,10 +24,10 @@ func TestWireRetryLifecyclePersistence(t *testing.T) { | |||
| 24 | 24 | if err != nil { | |
| 25 | 25 | t.Fatal(err) | |
| 26 | 26 | } | |
| 27 | - | if err := session.AppendMessage(provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "hello"}}}); err != nil { | |
| 28 | - | t.Fatal(err) | |
| 29 | - | } | |
| 30 | 27 | agent := core.NewAgent(nil, "model", "system", core.Registry{}) | |
| 28 | + | agent.SetMessages([]provider.Message{{ | |
| 29 | + | Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "hello"}}, | |
| 30 | + | }}) | |
| 31 | 31 | wireRetryLifecyclePersistence(agent, session) | |
| 32 | 32 | if agent.OnRetryLifecycle == nil { | |
| 33 | 33 | t.Fatal("session-backed agent has no retry lifecycle persistence") | |
| @@ -36,7 +36,10 @@ func TestWireRetryLifecyclePersistence(t *testing.T) { | |||
| 36 | 36 | Event: core.RetryLifecycleRequestFailed, Scope: core.RetryScopeAgent, | |
| 37 | 37 | Attempt: 1, MaxAttempts: 1, Reason: core.RetryReasonServer, Terminal: true, | |
| 38 | 38 | }) | |
| 39 | - | if err := session.AppendUsage(provider.Usage{}, provider.Usage{}); err != nil { | |
| 39 | + | // Headless modes call WriteNewTranscript on both success and provider | |
| 40 | + | // failure. Exercise that real finalization path rather than draining the | |
| 41 | + | // session buffer directly. | |
| 42 | + | if err := WriteNewTranscript(agent, session, 0); err != nil { | |
| 40 | 43 | t.Fatal(err) | |
| 41 | 44 | } | |
| 42 | 45 | path := session.Path | |
| @@ -47,8 +50,26 @@ func TestWireRetryLifecyclePersistence(t *testing.T) { | |||
| 47 | 50 | if err != nil { | |
| 48 | 51 | t.Fatal(err) | |
| 49 | 52 | } | |
| 50 | - | if !strings.Contains(string(raw), `"retry_lifecycle"`) { | |
| 51 | - | t.Fatalf("session lacks retry lifecycle: %s", raw) | |
| 53 | + | var lifecycle []core.RetryLifecycleRecord | |
| 54 | + | for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") { | |
| 55 | + | var row struct { | |
| 56 | + | Type string `json:"type"` | |
| 57 | + | RetryLifecycle []core.RetryLifecycleRecord `json:"retry_lifecycle"` | |
| 58 | + | } | |
| 59 | + | if err := json.Unmarshal([]byte(line), &row); err != nil { | |
| 60 | + | t.Fatal(err) | |
| 61 | + | } | |
| 62 | + | if row.Type == "usage" && len(row.RetryLifecycle) > 0 { | |
| 63 | + | lifecycle = row.RetryLifecycle | |
| 64 | + | } | |
| 65 | + | } | |
| 66 | + | if len(lifecycle) != 1 { | |
| 67 | + | t.Fatalf("retry lifecycle = %#v, want one record", lifecycle) | |
| 68 | + | } | |
| 69 | + | record := lifecycle[0] | |
| 70 | + | if record.Event != core.RetryLifecycleRequestFailed || record.Scope != core.RetryScopeAgent || | |
| 71 | + | record.Attempt != 1 || record.MaxAttempts != 1 || record.Reason != core.RetryReasonServer || !record.Terminal { | |
| 72 | + | t.Fatalf("retry lifecycle record = %#v", record) | |
| 52 | 73 | } | |
| 53 | 74 | ||
| 54 | 75 | wireRetryLifecyclePersistence(agent, nil) | |
| @@ -24,10 +24,10 @@ func TestWireRetryLifecyclePersistence(t *testing.T) { | |||||
| 24 | if err != nil { | 24 | if err != nil { | ||
| 25 | t.Fatal(err) | 25 | t.Fatal(err) | ||
| 26 | } | 26 | } | ||
| 27 | - | if err := session.AppendMessage(provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "hello"}}}); err != nil { | |||
| 28 | - | t.Fatal(err) | |||
| 29 | - | } | |||
| 30 | agent := core.NewAgent(nil, "model", "system", core.Registry{}) | 27 | agent := core.NewAgent(nil, "model", "system", core.Registry{}) | ||
| 28 | + | agent.SetMessages([]provider.Message{{ | |||
| 29 | + | Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "hello"}}, | |||
| 30 | + | }}) | |||
| 31 | wireRetryLifecyclePersistence(agent, session) | 31 | wireRetryLifecyclePersistence(agent, session) | ||
| 32 | if agent.OnRetryLifecycle == nil { | 32 | if agent.OnRetryLifecycle == nil { | ||
| 33 | t.Fatal("session-backed agent has no retry lifecycle persistence") | 33 | t.Fatal("session-backed agent has no retry lifecycle persistence") | ||
| @@ -36,7 +36,10 @@ func TestWireRetryLifecyclePersistence(t *testing.T) { | |||||
| 36 | Event: core.RetryLifecycleRequestFailed, Scope: core.RetryScopeAgent, | 36 | Event: core.RetryLifecycleRequestFailed, Scope: core.RetryScopeAgent, | ||
| 37 | Attempt: 1, MaxAttempts: 1, Reason: core.RetryReasonServer, Terminal: true, | 37 | Attempt: 1, MaxAttempts: 1, Reason: core.RetryReasonServer, Terminal: true, | ||
| 38 | }) | 38 | }) | ||
| 39 | - | if err := session.AppendUsage(provider.Usage{}, provider.Usage{}); err != nil { | 39 | + | // Headless modes call WriteNewTranscript on both success and provider |
| 40 | + | // failure. Exercise that real finalization path rather than draining the | |||
| 41 | + | // session buffer directly. | |||
| 42 | + | if err := WriteNewTranscript(agent, session, 0); err != nil { | |||
| 40 | t.Fatal(err) | 43 | t.Fatal(err) | ||
| 41 | } | 44 | } | ||
| 42 | path := session.Path | 45 | path := session.Path | ||
| @@ -47,8 +50,26 @@ func TestWireRetryLifecyclePersistence(t *testing.T) { | |||||
| 47 | if err != nil { | 50 | if err != nil { | ||
| 48 | t.Fatal(err) | 51 | t.Fatal(err) | ||
| 49 | } | 52 | } | ||
| 50 | - | if !strings.Contains(string(raw), `"retry_lifecycle"`) { | 53 | + | var lifecycle []core.RetryLifecycleRecord |
| 51 | - | t.Fatalf("session lacks retry lifecycle: %s", raw) | 54 | + | for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") { |
| 55 | + | var row struct { | |||
| 56 | + | Type string `json:"type"` | |||
| 57 | + | RetryLifecycle []core.RetryLifecycleRecord `json:"retry_lifecycle"` | |||
| 58 | + | } | |||
| 59 | + | if err := json.Unmarshal([]byte(line), &row); err != nil { | |||
| 60 | + | t.Fatal(err) | |||
| 61 | + | } | |||
| 62 | + | if row.Type == "usage" && len(row.RetryLifecycle) > 0 { | |||
| 63 | + | lifecycle = row.RetryLifecycle | |||
| 64 | + | } | |||
| 65 | + | } | |||
| 66 | + | if len(lifecycle) != 1 { | |||
| 67 | + | t.Fatalf("retry lifecycle = %#v, want one record", lifecycle) | |||
| 68 | + | } | |||
| 69 | + | record := lifecycle[0] | |||
| 70 | + | if record.Event != core.RetryLifecycleRequestFailed || record.Scope != core.RetryScopeAgent || | |||
| 71 | + | record.Attempt != 1 || record.MaxAttempts != 1 || record.Reason != core.RetryReasonServer || !record.Terminal { | |||
| 72 | + | t.Fatalf("retry lifecycle record = %#v", record) | |||
| 52 | } | 73 | } | ||
| 53 | 74 | ||||
| 54 | wireRetryLifecyclePersistence(agent, nil) | 75 | wireRetryLifecyclePersistence(agent, nil) | ||
M packages/agent/subagent_worker.go +9 -1
| @@ -162,7 +162,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||
| 162 | 162 | } | |
| 163 | 163 | } | |
| 164 | 164 | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | |
| 165 | - | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 165 | + | defer func() { | |
| 166 | + | runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) | |
| 167 | + | runErr = joinSessionCloseError(runErr, sess) | |
| 168 | + | }() | |
| 166 | 169 | } | |
| 167 | 170 | announceSession(extMgr, sess) | |
| 168 | 171 | ||
| @@ -343,6 +346,11 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||
| 343 | 346 | } | |
| 344 | 347 | } | |
| 345 | 348 | } | |
| 349 | + | if sess != nil { | |
| 350 | + | if persistErr := sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost()); persistErr != nil { | |
| 351 | + | err = errors.Join(err, persistErr) | |
| 352 | + | } | |
| 353 | + | } | |
| 346 | 354 | if sessionErr := sessionPersistence.err(); err == nil && sessionErr != nil { | |
| 347 | 355 | err = sessionErr | |
| 348 | 356 | } | |
| @@ -162,7 +162,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||||
| 162 | } | 162 | } | ||
| 163 | } | 163 | } | ||
| 164 | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | 164 | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | ||
| 165 | - | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 165 | + | defer func() { |
| 166 | + | runErr = errors.Join(runErr, sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost())) | |||
| 167 | + | runErr = joinSessionCloseError(runErr, sess) | |||
| 168 | + | }() | |||
| 166 | } | 169 | } | ||
| 167 | announceSession(extMgr, sess) | 170 | announceSession(extMgr, sess) | ||
| 168 | 171 | ||||
| @@ -343,6 +346,11 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||||
| 343 | } | 346 | } | ||
| 344 | } | 347 | } | ||
| 345 | } | 348 | } | ||
| 349 | + | if sess != nil { | |||
| 350 | + | if persistErr := sess.FlushRetryLifecycle(ag.LastTurnUsage(), ag.Cost()); persistErr != nil { | |||
| 351 | + | err = errors.Join(err, persistErr) | |||
| 352 | + | } | |||
| 353 | + | } | |||
| 346 | if sessionErr := sessionPersistence.err(); err == nil && sessionErr != nil { | 354 | if sessionErr := sessionPersistence.err(); err == nil && sessionErr != nil { | ||
| 347 | err = sessionErr | 355 | err = sessionErr | ||
| 348 | } | 356 | } | ||
M packages/core/compact.go +8 -2
| @@ -78,13 +78,19 @@ func (a *Agent) compact(ctx context.Context, keepTail int, textSink func(delta s | |||
| 78 | 78 | }, | |
| 79 | 79 | }, | |
| 80 | 80 | } | |
| 81 | - | if eventSink != nil { | |
| 81 | + | if eventSink != nil || a.OnRetryLifecycle != nil { | |
| 82 | + | lifecycleEventSink := eventSink | |
| 83 | + | if lifecycleEventSink == nil { | |
| 84 | + | lifecycleEventSink = func(AgentEvent) {} | |
| 85 | + | } | |
| 82 | 86 | req.Lifecycle = &requestLifecycleSink{ | |
| 83 | - | sink: eventSink, | |
| 87 | + | sink: lifecycleEventSink, | |
| 84 | 88 | observe: a.fireRetryLifecycle, | |
| 85 | 89 | provider: a.Client.Name(), | |
| 86 | 90 | model: a.Model, | |
| 87 | 91 | } | |
| 92 | + | } | |
| 93 | + | if eventSink != nil { | |
| 88 | 94 | eventSink(EvRequestStarted{ | |
| 89 | 95 | Provider: a.Client.Name(), | |
| 90 | 96 | Model: a.Model, | |
| @@ -78,13 +78,19 @@ func (a *Agent) compact(ctx context.Context, keepTail int, textSink func(delta s | |||||
| 78 | }, | 78 | }, | ||
| 79 | }, | 79 | }, | ||
| 80 | } | 80 | } | ||
| 81 | - | if eventSink != nil { | 81 | + | if eventSink != nil || a.OnRetryLifecycle != nil { |
| 82 | + | lifecycleEventSink := eventSink | |||
| 83 | + | if lifecycleEventSink == nil { | |||
| 84 | + | lifecycleEventSink = func(AgentEvent) {} | |||
| 85 | + | } | |||
| 82 | req.Lifecycle = &requestLifecycleSink{ | 86 | req.Lifecycle = &requestLifecycleSink{ | ||
| 83 | - | sink: eventSink, | 87 | + | sink: lifecycleEventSink, |
| 84 | observe: a.fireRetryLifecycle, | 88 | observe: a.fireRetryLifecycle, | ||
| 85 | provider: a.Client.Name(), | 89 | provider: a.Client.Name(), | ||
| 86 | model: a.Model, | 90 | model: a.Model, | ||
| 87 | } | 91 | } | ||
| 92 | + | } | |||
| 93 | + | if eventSink != nil { | |||
| 88 | eventSink(EvRequestStarted{ | 94 | eventSink(EvRequestStarted{ | ||
| 89 | Provider: a.Client.Name(), | 95 | Provider: a.Client.Name(), | ||
| 90 | Model: a.Model, | 96 | Model: a.Model, | ||
M packages/core/compact_lifecycle_test.go +44 -1
| @@ -2,14 +2,17 @@ package core | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "reflect" | |
| 5 | 6 | "strings" | |
| 6 | 7 | "testing" | |
| 8 | + | "time" | |
| 7 | 9 | ||
| 8 | 10 | "github.com/bnema/zut/packages/provider" | |
| 9 | 11 | ) | |
| 10 | 12 | ||
| 11 | 13 | type compactLifecycleClient struct { | |
| 12 | 14 | hadLifecycle bool | |
| 15 | + | reportRetry bool | |
| 13 | 16 | req provider.Request | |
| 14 | 17 | summary string | |
| 15 | 18 | usage []provider.Usage | |
| @@ -21,7 +24,17 @@ func (c *compactLifecycleClient) Stream(_ context.Context, req provider.Request) | |||
| 21 | 24 | c.req = req | |
| 22 | 25 | c.hadLifecycle = req.Lifecycle != nil | |
| 23 | 26 | if req.Lifecycle != nil { | |
| 24 | - | req.Lifecycle.RequestAttempt(1, 1) | |
| 27 | + | maxAttempts := 1 | |
| 28 | + | if c.reportRetry { | |
| 29 | + | maxAttempts = 2 | |
| 30 | + | } | |
| 31 | + | req.Lifecycle.RequestAttempt(1, maxAttempts) | |
| 32 | + | if c.reportRetry { | |
| 33 | + | if failures, ok := req.Lifecycle.(provider.RequestFailureLifecycle); ok { | |
| 34 | + | failures.RequestFailed(1, 2, provider.RequestFailureOverload, false) | |
| 35 | + | } | |
| 36 | + | req.Lifecycle.RetryScheduled(2, 2, 250*time.Millisecond) | |
| 37 | + | } | |
| 25 | 38 | } | |
| 26 | 39 | summary := c.summary | |
| 27 | 40 | if summary == "" { | |
| @@ -126,6 +139,36 @@ func TestCompactWithEventsEmitsRequestLifecycle(t *testing.T) { | |||
| 126 | 139 | } | |
| 127 | 140 | } | |
| 128 | 141 | ||
| 142 | + | func TestCompactPersistsProviderRetryLifecycleWithoutEventSink(t *testing.T) { | |
| 143 | + | client := &compactLifecycleClient{reportRetry: true} | |
| 144 | + | agent := NewAgent(client, "compact-model", "system", Registry{}) | |
| 145 | + | agent.SetMessages([]provider.Message{textMessage(provider.RoleUser, "source")}) | |
| 146 | + | ||
| 147 | + | var records []RetryLifecycleRecord | |
| 148 | + | agent.OnRetryLifecycle = func(record RetryLifecycleRecord) { | |
| 149 | + | records = append(records, record) | |
| 150 | + | } | |
| 151 | + | if _, err := agent.Compact(context.Background(), 0, nil); err != nil { | |
| 152 | + | t.Fatalf("Compact: %v", err) | |
| 153 | + | } | |
| 154 | + | if !client.hadLifecycle { | |
| 155 | + | t.Fatal("event-less compaction request did not receive a lifecycle observer") | |
| 156 | + | } | |
| 157 | + | want := []RetryLifecycleRecord{ | |
| 158 | + | { | |
| 159 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeProvider, | |
| 160 | + | Attempt: 1, MaxAttempts: 2, Reason: RetryReasonOverload, | |
| 161 | + | }, | |
| 162 | + | { | |
| 163 | + | Event: RetryLifecycleRetryScheduled, Scope: RetryScopeProvider, | |
| 164 | + | Attempt: 2, MaxAttempts: 2, Reason: RetryReasonOverload, DelayMS: 250, | |
| 165 | + | }, | |
| 166 | + | } | |
| 167 | + | if !reflect.DeepEqual(records, want) { | |
| 168 | + | t.Fatalf("retry lifecycle = %#v, want %#v", records, want) | |
| 169 | + | } | |
| 170 | + | } | |
| 171 | + | ||
| 129 | 172 | func TestCompactPromptPreservesActiveInstructions(t *testing.T) { | |
| 130 | 173 | const activeInstruction = "delegate independent work to subagents" | |
| 131 | 174 | client := &compactLifecycleClient{summary: "Active instruction still in force: " + activeInstruction} | |
| @@ -2,14 +2,17 @@ package core | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "reflect" | |||
| 5 | "strings" | 6 | "strings" | ||
| 6 | "testing" | 7 | "testing" | ||
| 8 | + | "time" | |||
| 7 | 9 | ||||
| 8 | "github.com/bnema/zut/packages/provider" | 10 | "github.com/bnema/zut/packages/provider" | ||
| 9 | ) | 11 | ) | ||
| 10 | 12 | ||||
| 11 | type compactLifecycleClient struct { | 13 | type compactLifecycleClient struct { | ||
| 12 | hadLifecycle bool | 14 | hadLifecycle bool | ||
| 15 | + | reportRetry bool | |||
| 13 | req provider.Request | 16 | req provider.Request | ||
| 14 | summary string | 17 | summary string | ||
| 15 | usage []provider.Usage | 18 | usage []provider.Usage | ||
| @@ -21,7 +24,17 @@ func (c *compactLifecycleClient) Stream(_ context.Context, req provider.Request) | |||||
| 21 | c.req = req | 24 | c.req = req | ||
| 22 | c.hadLifecycle = req.Lifecycle != nil | 25 | c.hadLifecycle = req.Lifecycle != nil | ||
| 23 | if req.Lifecycle != nil { | 26 | if req.Lifecycle != nil { | ||
| 24 | - | req.Lifecycle.RequestAttempt(1, 1) | 27 | + | maxAttempts := 1 |
| 28 | + | if c.reportRetry { | |||
| 29 | + | maxAttempts = 2 | |||
| 30 | + | } | |||
| 31 | + | req.Lifecycle.RequestAttempt(1, maxAttempts) | |||
| 32 | + | if c.reportRetry { | |||
| 33 | + | if failures, ok := req.Lifecycle.(provider.RequestFailureLifecycle); ok { | |||
| 34 | + | failures.RequestFailed(1, 2, provider.RequestFailureOverload, false) | |||
| 35 | + | } | |||
| 36 | + | req.Lifecycle.RetryScheduled(2, 2, 250*time.Millisecond) | |||
| 37 | + | } | |||
| 25 | } | 38 | } | ||
| 26 | summary := c.summary | 39 | summary := c.summary | ||
| 27 | if summary == "" { | 40 | if summary == "" { | ||
| @@ -126,6 +139,36 @@ func TestCompactWithEventsEmitsRequestLifecycle(t *testing.T) { | |||||
| 126 | } | 139 | } | ||
| 127 | } | 140 | } | ||
| 128 | 141 | ||||
| 142 | + | func TestCompactPersistsProviderRetryLifecycleWithoutEventSink(t *testing.T) { | |||
| 143 | + | client := &compactLifecycleClient{reportRetry: true} | |||
| 144 | + | agent := NewAgent(client, "compact-model", "system", Registry{}) | |||
| 145 | + | agent.SetMessages([]provider.Message{textMessage(provider.RoleUser, "source")}) | |||
| 146 | + | ||||
| 147 | + | var records []RetryLifecycleRecord | |||
| 148 | + | agent.OnRetryLifecycle = func(record RetryLifecycleRecord) { | |||
| 149 | + | records = append(records, record) | |||
| 150 | + | } | |||
| 151 | + | if _, err := agent.Compact(context.Background(), 0, nil); err != nil { | |||
| 152 | + | t.Fatalf("Compact: %v", err) | |||
| 153 | + | } | |||
| 154 | + | if !client.hadLifecycle { | |||
| 155 | + | t.Fatal("event-less compaction request did not receive a lifecycle observer") | |||
| 156 | + | } | |||
| 157 | + | want := []RetryLifecycleRecord{ | |||
| 158 | + | { | |||
| 159 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeProvider, | |||
| 160 | + | Attempt: 1, MaxAttempts: 2, Reason: RetryReasonOverload, | |||
| 161 | + | }, | |||
| 162 | + | { | |||
| 163 | + | Event: RetryLifecycleRetryScheduled, Scope: RetryScopeProvider, | |||
| 164 | + | Attempt: 2, MaxAttempts: 2, Reason: RetryReasonOverload, DelayMS: 250, | |||
| 165 | + | }, | |||
| 166 | + | } | |||
| 167 | + | if !reflect.DeepEqual(records, want) { | |||
| 168 | + | t.Fatalf("retry lifecycle = %#v, want %#v", records, want) | |||
| 169 | + | } | |||
| 170 | + | } | |||
| 171 | + | ||||
| 129 | func TestCompactPromptPreservesActiveInstructions(t *testing.T) { | 172 | func TestCompactPromptPreservesActiveInstructions(t *testing.T) { | ||
| 130 | const activeInstruction = "delegate independent work to subagents" | 173 | const activeInstruction = "delegate independent work to subagents" | ||
| 131 | client := &compactLifecycleClient{summary: "Active instruction still in force: " + activeInstruction} | 174 | client := &compactLifecycleClient{summary: "Active instruction still in force: " + activeInstruction} | ||
M packages/provider/retry.go +9 -3
| @@ -147,6 +147,9 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||
| 147 | 147 | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, false) | |
| 148 | 148 | continue | |
| 149 | 149 | } | |
| 150 | + | if resp.StatusCode >= http.StatusBadRequest { | |
| 151 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, requestFailureReasonForHTTP(resp.StatusCode, ""), true) | |
| 152 | + | } | |
| 150 | 153 | return resp, nil | |
| 151 | 154 | } | |
| 152 | 155 | if lastErr == nil { | |
| @@ -174,13 +177,16 @@ func reportRequestFailure(lifecycle RequestLifecycle, attempt, maxAttempts int, | |||
| 174 | 177 | } | |
| 175 | 178 | ||
| 176 | 179 | func requestFailureReasonForHTTP(status int, body string) RequestFailureReason { | |
| 180 | + | if status >= http.StatusBadRequest && status < http.StatusInternalServerError { | |
| 181 | + | if status == http.StatusTooManyRequests { | |
| 182 | + | return RequestFailureRateLimit | |
| 183 | + | } | |
| 184 | + | return RequestFailureClient | |
| 185 | + | } | |
| 177 | 186 | msg := strings.ToLower(body) | |
| 178 | 187 | if status == 529 || strings.Contains(msg, "overload") { | |
| 179 | 188 | return RequestFailureOverload | |
| 180 | 189 | } | |
| 181 | - | if status == http.StatusTooManyRequests { | |
| 182 | - | return RequestFailureRateLimit | |
| 183 | - | } | |
| 184 | 190 | return RequestFailureServer | |
| 185 | 191 | } | |
| 186 | 192 | ||
| @@ -147,6 +147,9 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||||
| 147 | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, false) | 147 | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, false) | ||
| 148 | continue | 148 | continue | ||
| 149 | } | 149 | } | ||
| 150 | + | if resp.StatusCode >= http.StatusBadRequest { | |||
| 151 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, requestFailureReasonForHTTP(resp.StatusCode, ""), true) | |||
| 152 | + | } | |||
| 150 | return resp, nil | 153 | return resp, nil | ||
| 151 | } | 154 | } | ||
| 152 | if lastErr == nil { | 155 | if lastErr == nil { | ||
| @@ -174,13 +177,16 @@ func reportRequestFailure(lifecycle RequestLifecycle, attempt, maxAttempts int, | |||||
| 174 | } | 177 | } | ||
| 175 | 178 | ||||
| 176 | func requestFailureReasonForHTTP(status int, body string) RequestFailureReason { | 179 | func requestFailureReasonForHTTP(status int, body string) RequestFailureReason { | ||
| 180 | + | if status >= http.StatusBadRequest && status < http.StatusInternalServerError { | |||
| 181 | + | if status == http.StatusTooManyRequests { | |||
| 182 | + | return RequestFailureRateLimit | |||
| 183 | + | } | |||
| 184 | + | return RequestFailureClient | |||
| 185 | + | } | |||
| 177 | msg := strings.ToLower(body) | 186 | msg := strings.ToLower(body) | ||
| 178 | if status == 529 || strings.Contains(msg, "overload") { | 187 | if status == 529 || strings.Contains(msg, "overload") { | ||
| 179 | return RequestFailureOverload | 188 | return RequestFailureOverload | ||
| 180 | } | 189 | } | ||
| 181 | - | if status == http.StatusTooManyRequests { | |||
| 182 | - | return RequestFailureRateLimit | |||
| 183 | - | } | |||
| 184 | return RequestFailureServer | 190 | return RequestFailureServer | ||
| 185 | } | 191 | } | ||
| 186 | 192 | ||||
M packages/provider/retry_test.go +26 -0
| @@ -284,6 +284,32 @@ func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | |||
| 284 | 284 | } | |
| 285 | 285 | } | |
| 286 | 286 | ||
| 287 | + | func TestDoStreamWithRetryReportsTerminalClientFailure(t *testing.T) { | |
| 288 | + | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | |
| 289 | + | w.WriteHeader(http.StatusUnauthorized) | |
| 290 | + | _, _ = w.Write([]byte("servers overloaded: private provider detail")) | |
| 291 | + | })) | |
| 292 | + | defer srv.Close() | |
| 293 | + | ||
| 294 | + | lifecycle := &recordingRequestLifecycle{} | |
| 295 | + | resp, err := doStreamWithRetry(context.Background(), srv.Client(), func() (*http.Request, error) { | |
| 296 | + | return http.NewRequestWithContext(context.Background(), http.MethodPost, srv.URL, nil) | |
| 297 | + | }, lifecycle) | |
| 298 | + | if err != nil { | |
| 299 | + | t.Fatalf("doStreamWithRetry: %v", err) | |
| 300 | + | } | |
| 301 | + | defer resp.Body.Close() | |
| 302 | + | if resp.StatusCode != http.StatusUnauthorized { | |
| 303 | + | t.Fatalf("status = %d, want 401", resp.StatusCode) | |
| 304 | + | } | |
| 305 | + | if got := lifecycle.failures; len(got) != 1 || got[0] != (lifecycleFailure{attempt: 1, max: 3, reason: RequestFailureClient, terminal: true}) { | |
| 306 | + | t.Fatalf("failures = %#v, want one terminal client failure", got) | |
| 307 | + | } | |
| 308 | + | if len(lifecycle.retries) != 0 { | |
| 309 | + | t.Fatalf("retries = %#v, want none", lifecycle.retries) | |
| 310 | + | } | |
| 311 | + | } | |
| 312 | + | ||
| 287 | 313 | func TestDoStreamWithRetryReportsTerminalQuota(t *testing.T) { | |
| 288 | 314 | const privateBody = `{"error":{"type":"insufficient_quota","message":"private provider detail"}}` | |
| 289 | 315 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | |
| @@ -284,6 +284,32 @@ func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | |||||
| 284 | } | 284 | } | ||
| 285 | } | 285 | } | ||
| 286 | 286 | ||||
| 287 | + | func TestDoStreamWithRetryReportsTerminalClientFailure(t *testing.T) { | |||
| 288 | + | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | |||
| 289 | + | w.WriteHeader(http.StatusUnauthorized) | |||
| 290 | + | _, _ = w.Write([]byte("servers overloaded: private provider detail")) | |||
| 291 | + | })) | |||
| 292 | + | defer srv.Close() | |||
| 293 | + | ||||
| 294 | + | lifecycle := &recordingRequestLifecycle{} | |||
| 295 | + | resp, err := doStreamWithRetry(context.Background(), srv.Client(), func() (*http.Request, error) { | |||
| 296 | + | return http.NewRequestWithContext(context.Background(), http.MethodPost, srv.URL, nil) | |||
| 297 | + | }, lifecycle) | |||
| 298 | + | if err != nil { | |||
| 299 | + | t.Fatalf("doStreamWithRetry: %v", err) | |||
| 300 | + | } | |||
| 301 | + | defer resp.Body.Close() | |||
| 302 | + | if resp.StatusCode != http.StatusUnauthorized { | |||
| 303 | + | t.Fatalf("status = %d, want 401", resp.StatusCode) | |||
| 304 | + | } | |||
| 305 | + | if got := lifecycle.failures; len(got) != 1 || got[0] != (lifecycleFailure{attempt: 1, max: 3, reason: RequestFailureClient, terminal: true}) { | |||
| 306 | + | t.Fatalf("failures = %#v, want one terminal client failure", got) | |||
| 307 | + | } | |||
| 308 | + | if len(lifecycle.retries) != 0 { | |||
| 309 | + | t.Fatalf("retries = %#v, want none", lifecycle.retries) | |||
| 310 | + | } | |||
| 311 | + | } | |||
| 312 | + | ||||
| 287 | func TestDoStreamWithRetryReportsTerminalQuota(t *testing.T) { | 313 | func TestDoStreamWithRetryReportsTerminalQuota(t *testing.T) { | ||
| 288 | const privateBody = `{"error":{"type":"insufficient_quota","message":"private provider detail"}}` | 314 | const privateBody = `{"error":{"type":"insufficient_quota","message":"private provider detail"}}` | ||
| 289 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | 315 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | ||