feat(sessions): persist sanitized retry lifecycle
- commit
- 387ed887cef709c85b3a4fa957ed65bc1429df64
- author
- brice · · 2026-08-10 09:28 UTC
- parent
- ddca230
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
15 files · +665 -28
M README.md +1 -1
| @@ -458,7 +458,7 @@ This is a guardrail against accidents, not a hard security boundary. If you need | |||
| 458 | 458 | ||
| 459 | 459 | ## Sessions | |
| 460 | 460 | ||
| 461 | - | Every interactive or print/json run (unless `--no-session`) writes a JSONL transcript under `$ZUT_HOME/sessions/<cwd-hash>/`. Resume any of them with `--continue`, `--resume`, `--session <path>`, or interactively via `/sessions` inside the TUI. Session metadata stores the provider and model used by that session; resume honors that pair, rebuilding the active agent when it differs and refusing the switch without discarding the current session if rebuilding fails. Empty sessions (the user exited without prompting) are deleted on close so the list stays tidy. | |
| 461 | + | Every interactive or print/json run (unless `--no-session`) writes a JSONL transcript under `$ZUT_HOME/sessions/<cwd-hash>/`. Resume any of them with `--continue`, `--resume`, `--session <path>`, or interactively via `/sessions` inside the TUI. Session metadata stores the provider and model used by that session; resume honors that pair, rebuilding the active agent when it differs and refusing the switch without discarding the current session if rebuilding fails. Usage checkpoints may include sanitized `retry_lifecycle` metadata—attempts, backoff delays, terminal state, and an allowlisted failure category—but never raw provider errors or response bodies. This metadata is ignored when rebuilding provider context. Empty sessions (the user exited without prompting) are deleted on close so the list stays tidy. | |
| 462 | 462 | ||
| 463 | 463 | ## Providers | |
| 464 | 464 | ||
| @@ -458,7 +458,7 @@ This is a guardrail against accidents, not a hard security boundary. If you need | |||||
| 458 | 458 | ||||
| 459 | ## Sessions | 459 | ## Sessions | ||
| 460 | 460 | ||||
| 461 | - | Every interactive or print/json run (unless `--no-session`) writes a JSONL transcript under `$ZUT_HOME/sessions/<cwd-hash>/`. Resume any of them with `--continue`, `--resume`, `--session <path>`, or interactively via `/sessions` inside the TUI. Session metadata stores the provider and model used by that session; resume honors that pair, rebuilding the active agent when it differs and refusing the switch without discarding the current session if rebuilding fails. Empty sessions (the user exited without prompting) are deleted on close so the list stays tidy. | 461 | + | Every interactive or print/json run (unless `--no-session`) writes a JSONL transcript under `$ZUT_HOME/sessions/<cwd-hash>/`. Resume any of them with `--continue`, `--resume`, `--session <path>`, or interactively via `/sessions` inside the TUI. Session metadata stores the provider and model used by that session; resume honors that pair, rebuilding the active agent when it differs and refusing the switch without discarding the current session if rebuilding fails. Usage checkpoints may include sanitized `retry_lifecycle` metadata—attempts, backoff delays, terminal state, and an allowlisted failure category—but never raw provider errors or response bodies. This metadata is ignored when rebuilding provider context. Empty sessions (the user exited without prompting) are deleted on close so the list stays tidy. |
| 462 | 462 | ||||
| 463 | ## Providers | 463 | ## Providers | ||
| 464 | 464 | ||||
M packages/agent/botcmd.go +5 -1
| @@ -249,10 +249,14 @@ func botRun(spec *botSpec, rawTail []string, version string) (runErr error) { | |||
| 249 | 249 | agent.OnUsage = func(u provider.Usage) { | |
| 250 | 250 | _ = sess.AppendUsage(u, u) | |
| 251 | 251 | } | |
| 252 | + | agent.OnRetryLifecycle = sess.RecordRetryLifecycle | |
| 252 | 253 | agent.OnTranscriptCompacted = func(msgs []provider.Message) { | |
| 253 | 254 | _ = sess.AppendCompaction(msgs) | |
| 254 | 255 | } | |
| 255 | - | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 256 | + | defer func() { | |
| 257 | + | runErr = errors.Join(runErr, sess.FlushRetryLifecycle(agent.LastTurnUsage(), agent.Cost())) | |
| 258 | + | runErr = joinSessionCloseError(runErr, sess) | |
| 259 | + | }() | |
| 256 | 260 | } else { | |
| 257 | 261 | fmt.Fprintln(os.Stderr, "session:", serr) | |
| 258 | 262 | } | |
| @@ -249,10 +249,14 @@ func botRun(spec *botSpec, rawTail []string, version string) (runErr error) { | |||||
| 249 | agent.OnUsage = func(u provider.Usage) { | 249 | agent.OnUsage = func(u provider.Usage) { | ||
| 250 | _ = sess.AppendUsage(u, u) | 250 | _ = sess.AppendUsage(u, u) | ||
| 251 | } | 251 | } | ||
| 252 | + | agent.OnRetryLifecycle = sess.RecordRetryLifecycle | |||
| 252 | agent.OnTranscriptCompacted = func(msgs []provider.Message) { | 253 | agent.OnTranscriptCompacted = func(msgs []provider.Message) { | ||
| 253 | _ = sess.AppendCompaction(msgs) | 254 | _ = sess.AppendCompaction(msgs) | ||
| 254 | } | 255 | } | ||
| 255 | - | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 256 | + | defer func() { |
| 257 | + | runErr = errors.Join(runErr, sess.FlushRetryLifecycle(agent.LastTurnUsage(), agent.Cost())) | |||
| 258 | + | runErr = joinSessionCloseError(runErr, sess) | |||
| 259 | + | }() | |||
| 256 | } else { | 260 | } else { | ||
| 257 | fmt.Fprintln(os.Stderr, "session:", serr) | 261 | fmt.Fprintln(os.Stderr, "session:", serr) | ||
| 258 | } | 262 | } | ||
M packages/agent/cli.go +24 -0
| @@ -988,6 +988,7 @@ func runPrintMode(ctx context.Context, args Args, version string) (runErr error) | |||
| 988 | 988 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 989 | 989 | } | |
| 990 | 990 | announceSession(extMgr, sess) | |
| 991 | + | wireRetryLifecyclePersistence(ag, sess) | |
| 991 | 992 | ||
| 992 | 993 | prompt := args.Prompt | |
| 993 | 994 | if prompt == "" { | |
| @@ -1069,6 +1070,7 @@ func runStreamMode(ctx context.Context, args Args, version string) (runErr error | |||
| 1069 | 1070 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 1070 | 1071 | } | |
| 1071 | 1072 | announceSession(extMgr, sess) | |
| 1073 | + | wireRetryLifecyclePersistence(ag, sess) | |
| 1072 | 1074 | ||
| 1073 | 1075 | prompt := args.Prompt | |
| 1074 | 1076 | if prompt == "" { | |
| @@ -1189,6 +1191,7 @@ func runJSONMode(ctx context.Context, args Args, version string) (runErr error) | |||
| 1189 | 1191 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 1190 | 1192 | } | |
| 1191 | 1193 | announceSession(extMgr, sess) | |
| 1194 | + | wireRetryLifecyclePersistence(ag, sess) | |
| 1192 | 1195 | ||
| 1193 | 1196 | prompt := args.Prompt | |
| 1194 | 1197 | if prompt == "" { | |
| @@ -1785,6 +1788,15 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 1785 | 1788 | } | |
| 1786 | 1789 | _ = sess.AppendUsage(cum, cum) | |
| 1787 | 1790 | } | |
| 1791 | + | persistRetryLifecycle := func(record core.RetryLifecycleRecord) { | |
| 1792 | + | sessionTransitionMu.RLock() | |
| 1793 | + | defer sessionTransitionMu.RUnlock() | |
| 1794 | + | persistMu.Lock() | |
| 1795 | + | defer persistMu.Unlock() | |
| 1796 | + | if sess != nil { | |
| 1797 | + | sess.RecordRetryLifecycle(record) | |
| 1798 | + | } | |
| 1799 | + | } | |
| 1788 | 1800 | persistCompaction := func(messages []provider.Message) { | |
| 1789 | 1801 | sessionTransitionMu.RLock() | |
| 1790 | 1802 | var session *extproto.SessionContext | |
| @@ -1862,6 +1874,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 1862 | 1874 | } | |
| 1863 | 1875 | a.OnMessageAppended = persistMessage | |
| 1864 | 1876 | a.OnUsage = persistUsage | |
| 1877 | + | a.OnRetryLifecycle = persistRetryLifecycle | |
| 1865 | 1878 | a.OnTranscriptCompacted = persistCompaction | |
| 1866 | 1879 | bindAgentToolResultSession(a, currentSession) | |
| 1867 | 1880 | return a | |
| @@ -2796,6 +2809,17 @@ func WriteNewTranscript(ag *core.Agent, sess *core.Session, from int) error { | |||
| 2796 | 2809 | return err | |
| 2797 | 2810 | } | |
| 2798 | 2811 | ||
| 2812 | + | func wireRetryLifecyclePersistence(ag *core.Agent, sess *core.Session) { | |
| 2813 | + | if ag == nil { | |
| 2814 | + | return | |
| 2815 | + | } | |
| 2816 | + | if sess == nil { | |
| 2817 | + | ag.OnRetryLifecycle = nil | |
| 2818 | + | return | |
| 2819 | + | } | |
| 2820 | + | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | |
| 2821 | + | } | |
| 2822 | + | ||
| 2799 | 2823 | // writeNewTranscriptLocked is the same as WriteNewTranscript. The | |
| 2800 | 2824 | // suffix marks that interactive callers must hold persistMu when | |
| 2801 | 2825 | // invoking it so concurrent appends from the agent loop don't race | |
| @@ -988,6 +988,7 @@ func runPrintMode(ctx context.Context, args Args, version string) (runErr error) | |||||
| 988 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 988 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 989 | } | 989 | } | ||
| 990 | announceSession(extMgr, sess) | 990 | announceSession(extMgr, sess) | ||
| 991 | + | wireRetryLifecyclePersistence(ag, sess) | |||
| 991 | 992 | ||||
| 992 | prompt := args.Prompt | 993 | prompt := args.Prompt | ||
| 993 | if prompt == "" { | 994 | if prompt == "" { | ||
| @@ -1069,6 +1070,7 @@ func runStreamMode(ctx context.Context, args Args, version string) (runErr error | |||||
| 1069 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 1070 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 1070 | } | 1071 | } | ||
| 1071 | announceSession(extMgr, sess) | 1072 | announceSession(extMgr, sess) | ||
| 1073 | + | wireRetryLifecyclePersistence(ag, sess) | |||
| 1072 | 1074 | ||||
| 1073 | prompt := args.Prompt | 1075 | prompt := args.Prompt | ||
| 1074 | if prompt == "" { | 1076 | if prompt == "" { | ||
| @@ -1189,6 +1191,7 @@ func runJSONMode(ctx context.Context, args Args, version string) (runErr error) | |||||
| 1189 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 1191 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 1190 | } | 1192 | } | ||
| 1191 | announceSession(extMgr, sess) | 1193 | announceSession(extMgr, sess) | ||
| 1194 | + | wireRetryLifecyclePersistence(ag, sess) | |||
| 1192 | 1195 | ||||
| 1193 | prompt := args.Prompt | 1196 | prompt := args.Prompt | ||
| 1194 | if prompt == "" { | 1197 | if prompt == "" { | ||
| @@ -1785,6 +1788,15 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 1785 | } | 1788 | } | ||
| 1786 | _ = sess.AppendUsage(cum, cum) | 1789 | _ = sess.AppendUsage(cum, cum) | ||
| 1787 | } | 1790 | } | ||
| 1791 | + | persistRetryLifecycle := func(record core.RetryLifecycleRecord) { | |||
| 1792 | + | sessionTransitionMu.RLock() | |||
| 1793 | + | defer sessionTransitionMu.RUnlock() | |||
| 1794 | + | persistMu.Lock() | |||
| 1795 | + | defer persistMu.Unlock() | |||
| 1796 | + | if sess != nil { | |||
| 1797 | + | sess.RecordRetryLifecycle(record) | |||
| 1798 | + | } | |||
| 1799 | + | } | |||
| 1788 | persistCompaction := func(messages []provider.Message) { | 1800 | persistCompaction := func(messages []provider.Message) { | ||
| 1789 | sessionTransitionMu.RLock() | 1801 | sessionTransitionMu.RLock() | ||
| 1790 | var session *extproto.SessionContext | 1802 | var session *extproto.SessionContext | ||
| @@ -1862,6 +1874,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 1862 | } | 1874 | } | ||
| 1863 | a.OnMessageAppended = persistMessage | 1875 | a.OnMessageAppended = persistMessage | ||
| 1864 | a.OnUsage = persistUsage | 1876 | a.OnUsage = persistUsage | ||
| 1877 | + | a.OnRetryLifecycle = persistRetryLifecycle | |||
| 1865 | a.OnTranscriptCompacted = persistCompaction | 1878 | a.OnTranscriptCompacted = persistCompaction | ||
| 1866 | bindAgentToolResultSession(a, currentSession) | 1879 | bindAgentToolResultSession(a, currentSession) | ||
| 1867 | return a | 1880 | return a | ||
| @@ -2796,6 +2809,17 @@ func WriteNewTranscript(ag *core.Agent, sess *core.Session, from int) error { | |||||
| 2796 | return err | 2809 | return err | ||
| 2797 | } | 2810 | } | ||
| 2798 | 2811 | ||||
| 2812 | + | func wireRetryLifecyclePersistence(ag *core.Agent, sess *core.Session) { | |||
| 2813 | + | if ag == nil { | |||
| 2814 | + | return | |||
| 2815 | + | } | |||
| 2816 | + | if sess == nil { | |||
| 2817 | + | ag.OnRetryLifecycle = nil | |||
| 2818 | + | return | |||
| 2819 | + | } | |||
| 2820 | + | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | |||
| 2821 | + | } | |||
| 2822 | + | ||||
| 2799 | // writeNewTranscriptLocked is the same as WriteNewTranscript. The | 2823 | // writeNewTranscriptLocked is the same as WriteNewTranscript. The | ||
| 2800 | // suffix marks that interactive callers must hold persistMu when | 2824 | // suffix marks that interactive callers must hold persistMu when | ||
| 2801 | // invoking it so concurrent appends from the agent loop don't race | 2825 | // invoking it so concurrent appends from the agent loop don't race | ||
M packages/agent/cli_session_test.go +38 -0
| @@ -19,6 +19,44 @@ import ( | |||
| 19 | 19 | "github.com/google/uuid" | |
| 20 | 20 | ) | |
| 21 | 21 | ||
| 22 | + | func TestWireRetryLifecyclePersistence(t *testing.T) { | |
| 23 | + | session, err := core.NewSession(t.TempDir(), "/workspace", "provider", "model", "test") | |
| 24 | + | if err != nil { | |
| 25 | + | t.Fatal(err) | |
| 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{}) | |
| 31 | + | wireRetryLifecyclePersistence(agent, session) | |
| 32 | + | if agent.OnRetryLifecycle == nil { | |
| 33 | + | t.Fatal("session-backed agent has no retry lifecycle persistence") | |
| 34 | + | } | |
| 35 | + | agent.OnRetryLifecycle(core.RetryLifecycleRecord{ | |
| 36 | + | Event: core.RetryLifecycleRequestFailed, Scope: core.RetryScopeAgent, | |
| 37 | + | Attempt: 1, MaxAttempts: 1, Reason: core.RetryReasonServer, Terminal: true, | |
| 38 | + | }) | |
| 39 | + | if err := session.AppendUsage(provider.Usage{}, provider.Usage{}); err != nil { | |
| 40 | + | t.Fatal(err) | |
| 41 | + | } | |
| 42 | + | path := session.Path | |
| 43 | + | if err := session.Close(); err != nil { | |
| 44 | + | t.Fatal(err) | |
| 45 | + | } | |
| 46 | + | raw, err := os.ReadFile(path) | |
| 47 | + | if err != nil { | |
| 48 | + | t.Fatal(err) | |
| 49 | + | } | |
| 50 | + | if !strings.Contains(string(raw), `"retry_lifecycle"`) { | |
| 51 | + | t.Fatalf("session lacks retry lifecycle: %s", raw) | |
| 52 | + | } | |
| 53 | + | ||
| 54 | + | wireRetryLifecyclePersistence(agent, nil) | |
| 55 | + | if agent.OnRetryLifecycle != nil { | |
| 56 | + | t.Fatal("sessionless agent retained retry lifecycle persistence") | |
| 57 | + | } | |
| 58 | + | } | |
| 59 | + | ||
| 22 | 60 | func TestToolResultPersistenceUsesBoundSession(t *testing.T) { | |
| 23 | 61 | root := t.TempDir() | |
| 24 | 62 | original, err := core.NewSession(root, "original", "provider", "model", "test") | |
| @@ -19,6 +19,44 @@ import ( | |||||
| 19 | "github.com/google/uuid" | 19 | "github.com/google/uuid" | ||
| 20 | ) | 20 | ) | ||
| 21 | 21 | ||||
| 22 | + | func TestWireRetryLifecyclePersistence(t *testing.T) { | |||
| 23 | + | session, err := core.NewSession(t.TempDir(), "/workspace", "provider", "model", "test") | |||
| 24 | + | if err != nil { | |||
| 25 | + | t.Fatal(err) | |||
| 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{}) | |||
| 31 | + | wireRetryLifecyclePersistence(agent, session) | |||
| 32 | + | if agent.OnRetryLifecycle == nil { | |||
| 33 | + | t.Fatal("session-backed agent has no retry lifecycle persistence") | |||
| 34 | + | } | |||
| 35 | + | agent.OnRetryLifecycle(core.RetryLifecycleRecord{ | |||
| 36 | + | Event: core.RetryLifecycleRequestFailed, Scope: core.RetryScopeAgent, | |||
| 37 | + | Attempt: 1, MaxAttempts: 1, Reason: core.RetryReasonServer, Terminal: true, | |||
| 38 | + | }) | |||
| 39 | + | if err := session.AppendUsage(provider.Usage{}, provider.Usage{}); err != nil { | |||
| 40 | + | t.Fatal(err) | |||
| 41 | + | } | |||
| 42 | + | path := session.Path | |||
| 43 | + | if err := session.Close(); err != nil { | |||
| 44 | + | t.Fatal(err) | |||
| 45 | + | } | |||
| 46 | + | raw, err := os.ReadFile(path) | |||
| 47 | + | if err != nil { | |||
| 48 | + | t.Fatal(err) | |||
| 49 | + | } | |||
| 50 | + | if !strings.Contains(string(raw), `"retry_lifecycle"`) { | |||
| 51 | + | t.Fatalf("session lacks retry lifecycle: %s", raw) | |||
| 52 | + | } | |||
| 53 | + | ||||
| 54 | + | wireRetryLifecyclePersistence(agent, nil) | |||
| 55 | + | if agent.OnRetryLifecycle != nil { | |||
| 56 | + | t.Fatal("sessionless agent retained retry lifecycle persistence") | |||
| 57 | + | } | |||
| 58 | + | } | |||
| 59 | + | ||||
| 22 | func TestToolResultPersistenceUsesBoundSession(t *testing.T) { | 60 | func TestToolResultPersistenceUsesBoundSession(t *testing.T) { | ||
| 23 | root := t.TempDir() | 61 | root := t.TempDir() | ||
| 24 | original, err := core.NewSession(root, "original", "provider", "model", "test") | 62 | original, err := core.NewSession(root, "original", "provider", "model", "test") | ||
M packages/agent/orchestration_modes.go +1 -0
| @@ -107,6 +107,7 @@ func runOrchestratedMode(parentCtx context.Context, args Args, version string, h | |||
| 107 | 107 | runtime.SetActiveSession(sess.ID) | |
| 108 | 108 | } | |
| 109 | 109 | announceSession(extMgr, sess) | |
| 110 | + | wireRetryLifecyclePersistence(ag, sess) | |
| 110 | 111 | runtime.WireRequiredWorkerGate(ag) | |
| 111 | 112 | ||
| 112 | 113 | start := len(ag.Messages()) | |
| @@ -107,6 +107,7 @@ func runOrchestratedMode(parentCtx context.Context, args Args, version string, h | |||||
| 107 | runtime.SetActiveSession(sess.ID) | 107 | runtime.SetActiveSession(sess.ID) | ||
| 108 | } | 108 | } | ||
| 109 | announceSession(extMgr, sess) | 109 | announceSession(extMgr, sess) | ||
| 110 | + | wireRetryLifecyclePersistence(ag, sess) | |||
| 110 | runtime.WireRequiredWorkerGate(ag) | 111 | runtime.WireRequiredWorkerGate(ag) | ||
| 111 | 112 | ||||
| 112 | start := len(ag.Messages()) | 113 | start := len(ag.Messages()) | ||
M packages/agent/subagent_worker.go +1 -0
| @@ -161,6 +161,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||
| 161 | 161 | sessionPersistence.record(err) | |
| 162 | 162 | } | |
| 163 | 163 | } | |
| 164 | + | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | |
| 164 | 165 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 165 | 166 | } | |
| 166 | 167 | announceSession(extMgr, sess) | |
| @@ -161,6 +161,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||||
| 161 | sessionPersistence.record(err) | 161 | sessionPersistence.record(err) | ||
| 162 | } | 162 | } | ||
| 163 | } | 163 | } | ||
| 164 | + | ag.OnRetryLifecycle = sess.RecordRetryLifecycle | |||
| 164 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 165 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 165 | } | 166 | } | ||
| 166 | announceSession(extMgr, sess) | 167 | announceSession(extMgr, sess) | ||
M packages/core/agent.go +153 -11
| @@ -117,6 +117,10 @@ type Agent struct { | |||
| 117 | 117 | // current and a crash recovers the right cost figure. | |
| 118 | 118 | OnUsage func(cumulative provider.Usage) | |
| 119 | 119 | ||
| 120 | + | // OnRetryLifecycle receives sanitized retry audit records. It never | |
| 121 | + | // receives provider error text or response bodies. | |
| 122 | + | OnRetryLifecycle func(RetryLifecycleRecord) | |
| 123 | + | ||
| 120 | 124 | // OnTranscriptCompacted, if set, fires after Compact replaces the | |
| 121 | 125 | // in-memory transcript with the synthetic summary plus kept tail. | |
| 122 | 126 | // Hosts wire this to append an explicit compaction checkpoint to | |
| @@ -508,7 +512,32 @@ func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error { | |||
| 508 | 512 | for attempt := 0; ; attempt++ { | |
| 509 | 513 | stop, assistantMsg, err = a.oneTurn(ctx, sink, turnContext, attempt, maxAttempts) | |
| 510 | 514 | sink(EvTurnEnd{Stop: stop, Err: err}) | |
| 511 | - | if err == nil || !a.canRetryError(err, attempt) { | |
| 515 | + | if err == nil { | |
| 516 | + | break | |
| 517 | + | } | |
| 518 | + | retryable := a.canRetryError(err, attempt) | |
| 519 | + | var retryCtxErr error | |
| 520 | + | if retryable { | |
| 521 | + | retryCtxErr = ctx.Err() | |
| 522 | + | if retryCtxErr != nil { | |
| 523 | + | retryable = false | |
| 524 | + | } | |
| 525 | + | } | |
| 526 | + | reason := classifyRetryReason(err) | |
| 527 | + | if !errors.Is(err, context.Canceled) { | |
| 528 | + | a.fireRetryLifecycle(RetryLifecycleRecord{ | |
| 529 | + | Event: RetryLifecycleRequestFailed, | |
| 530 | + | Scope: RetryScopeAgent, | |
| 531 | + | Attempt: attempt + 1, | |
| 532 | + | MaxAttempts: maxAttempts, | |
| 533 | + | Reason: reason, | |
| 534 | + | Terminal: !retryable, | |
| 535 | + | }) | |
| 536 | + | } | |
| 537 | + | if retryCtxErr != nil { | |
| 538 | + | return retryCtxErr | |
| 539 | + | } | |
| 540 | + | if !retryable { | |
| 512 | 541 | break | |
| 513 | 542 | } | |
| 514 | 543 | a.dropLastAssistantMessage() | |
| @@ -516,6 +545,14 @@ func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error { | |||
| 516 | 545 | if retryErr := ctx.Err(); retryErr != nil { | |
| 517 | 546 | return retryErr | |
| 518 | 547 | } | |
| 548 | + | a.fireRetryLifecycle(RetryLifecycleRecord{ | |
| 549 | + | Event: RetryLifecycleRetryScheduled, | |
| 550 | + | Scope: RetryScopeAgent, | |
| 551 | + | Attempt: attempt + 2, | |
| 552 | + | MaxAttempts: maxAttempts, | |
| 553 | + | Reason: reason, | |
| 554 | + | DelayMS: delay.Milliseconds(), | |
| 555 | + | }) | |
| 519 | 556 | sink(EvRetryScheduled{ | |
| 520 | 557 | Scope: RetryScopeAgent, | |
| 521 | 558 | Attempt: attempt + 2, | |
| @@ -643,6 +680,57 @@ func isNonRetryableProviderLimit(msg string) bool { | |||
| 643 | 680 | return false | |
| 644 | 681 | } | |
| 645 | 682 | ||
| 683 | + | func classifyRetryReason(err error) RetryReason { | |
| 684 | + | if err == nil { | |
| 685 | + | return RetryReasonUnknown | |
| 686 | + | } | |
| 687 | + | msg := strings.ToLower(err.Error()) | |
| 688 | + | if isNonRetryableProviderLimit(msg) { | |
| 689 | + | return RetryReasonQuota | |
| 690 | + | } | |
| 691 | + | if strings.Contains(msg, "overload") || strings.Contains(msg, "servers are busy") { | |
| 692 | + | return RetryReasonOverload | |
| 693 | + | } | |
| 694 | + | if strings.Contains(msg, "rate limit") || strings.Contains(msg, "ratelimit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "429") { | |
| 695 | + | return RetryReasonRateLimit | |
| 696 | + | } | |
| 697 | + | if errors.Is(err, context.DeadlineExceeded) || strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out") { | |
| 698 | + | return RetryReasonTimeout | |
| 699 | + | } | |
| 700 | + | for _, needle := range []string{"connection", "network", "unexpected eof", "broken pipe", "socket hang up", "transport failure", "upstream connect"} { | |
| 701 | + | if strings.Contains(msg, needle) { | |
| 702 | + | return RetryReasonNetwork | |
| 703 | + | } | |
| 704 | + | } | |
| 705 | + | for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} { | |
| 706 | + | if strings.Contains(msg, needle) { | |
| 707 | + | return RetryReasonContextWindow | |
| 708 | + | } | |
| 709 | + | } | |
| 710 | + | for _, needle := range []string{"unauthorized", "authentication", "invalid api key", "invalid_api_key", "http 401", "http 403"} { | |
| 711 | + | if strings.Contains(msg, needle) { | |
| 712 | + | return RetryReasonAuth | |
| 713 | + | } | |
| 714 | + | } | |
| 715 | + | for _, needle := range []string{"server error", "internal error", "service unavailable", "http 500", "http 502", "http 503", "http 504", "provider returned error"} { | |
| 716 | + | if strings.Contains(msg, needle) { | |
| 717 | + | return RetryReasonServer | |
| 718 | + | } | |
| 719 | + | } | |
| 720 | + | for _, needle := range []string{"bad request", "unsupported", "invalid parameter", "http 400", "http 404", "http 422"} { | |
| 721 | + | if strings.Contains(msg, needle) { | |
| 722 | + | return RetryReasonClient | |
| 723 | + | } | |
| 724 | + | } | |
| 725 | + | return RetryReasonUnknown | |
| 726 | + | } | |
| 727 | + | ||
| 728 | + | func (a *Agent) fireRetryLifecycle(record RetryLifecycleRecord) { | |
| 729 | + | if a.OnRetryLifecycle != nil { | |
| 730 | + | a.OnRetryLifecycle(record) | |
| 731 | + | } | |
| 732 | + | } | |
| 733 | + | ||
| 646 | 734 | func (a *Agent) retryDelay(attempt int) time.Duration { | |
| 647 | 735 | base := a.RetryBaseDelay | |
| 648 | 736 | if base <= 0 { | |
| @@ -677,12 +765,14 @@ func (a *Agent) dropLastAssistantMessage() { | |||
| 677 | 765 | // requestLifecycleSink adapts provider-owned pre-stream retries into core | |
| 678 | 766 | // events without letting provider concerns leak into mode renderers. | |
| 679 | 767 | type requestLifecycleSink struct { | |
| 680 | - | sink func(AgentEvent) | |
| 681 | - | provider string | |
| 682 | - | model string | |
| 768 | + | sink func(AgentEvent) | |
| 769 | + | observe func(RetryLifecycleRecord) | |
| 770 | + | provider string | |
| 771 | + | model string | |
| 772 | + | lastReason RetryReason | |
| 683 | 773 | } | |
| 684 | 774 | ||
| 685 | - | func (s requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) { | |
| 775 | + | func (s *requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) { | |
| 686 | 776 | s.sink(EvRequestStarted{ | |
| 687 | 777 | Provider: s.provider, | |
| 688 | 778 | Model: s.model, | |
| @@ -692,13 +782,63 @@ func (s requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) { | |||
| 692 | 782 | }) | |
| 693 | 783 | } | |
| 694 | 784 | ||
| 695 | - | func (s requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) { | |
| 785 | + | func (s *requestLifecycleSink) RequestFailed(attempt, maxAttempts int, reason provider.RequestFailureReason, terminal bool) { | |
| 786 | + | s.lastReason = retryReasonFromProvider(reason) | |
| 787 | + | if s.observe != nil { | |
| 788 | + | s.observe(RetryLifecycleRecord{ | |
| 789 | + | Event: RetryLifecycleRequestFailed, | |
| 790 | + | Scope: RetryScopeProvider, | |
| 791 | + | Attempt: attempt, | |
| 792 | + | MaxAttempts: maxAttempts, | |
| 793 | + | Reason: s.lastReason, | |
| 794 | + | Terminal: terminal, | |
| 795 | + | }) | |
| 796 | + | } | |
| 797 | + | } | |
| 798 | + | ||
| 799 | + | func (s *requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) { | |
| 696 | 800 | s.sink(EvRetryScheduled{ | |
| 697 | 801 | Scope: RetryScopeProvider, | |
| 698 | 802 | Attempt: attempt, | |
| 699 | 803 | MaxAttempts: maxAttempts, | |
| 700 | 804 | Delay: delay, | |
| 701 | 805 | }) | |
| 806 | + | if s.observe != nil { | |
| 807 | + | reason := s.lastReason | |
| 808 | + | if reason == "" { | |
| 809 | + | reason = RetryReasonUnknown | |
| 810 | + | } | |
| 811 | + | s.observe(RetryLifecycleRecord{ | |
| 812 | + | Event: RetryLifecycleRetryScheduled, | |
| 813 | + | Scope: RetryScopeProvider, | |
| 814 | + | Attempt: attempt, | |
| 815 | + | MaxAttempts: maxAttempts, | |
| 816 | + | Reason: reason, | |
| 817 | + | DelayMS: delay.Milliseconds(), | |
| 818 | + | }) | |
| 819 | + | } | |
| 820 | + | s.lastReason = "" | |
| 821 | + | } | |
| 822 | + | ||
| 823 | + | func retryReasonFromProvider(reason provider.RequestFailureReason) RetryReason { | |
| 824 | + | switch reason { | |
| 825 | + | case provider.RequestFailureOverload: | |
| 826 | + | return RetryReasonOverload | |
| 827 | + | case provider.RequestFailureRateLimit: | |
| 828 | + | return RetryReasonRateLimit | |
| 829 | + | case provider.RequestFailureQuota: | |
| 830 | + | return RetryReasonQuota | |
| 831 | + | case provider.RequestFailureServer: | |
| 832 | + | return RetryReasonServer | |
| 833 | + | case provider.RequestFailureNetwork: | |
| 834 | + | return RetryReasonNetwork | |
| 835 | + | case provider.RequestFailureTimeout: | |
| 836 | + | return RetryReasonTimeout | |
| 837 | + | case provider.RequestFailureClient: | |
| 838 | + | return RetryReasonClient | |
| 839 | + | default: | |
| 840 | + | return RetryReasonUnknown | |
| 841 | + | } | |
| 702 | 842 | } | |
| 703 | 843 | ||
| 704 | 844 | // oneTurn calls the LLM once, forwards events, returns the stop reason | |
| @@ -720,6 +860,12 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext | |||
| 720 | 860 | // repair can add stub results for aborted calls; those results must also | |
| 721 | 861 | // remain in the request so every tool call still has a matching result. | |
| 722 | 862 | messages := repairToolUseResultPairs(a.Messages()) | |
| 863 | + | requestLifecycle := &requestLifecycleSink{ | |
| 864 | + | sink: sink, | |
| 865 | + | observe: a.fireRetryLifecycle, | |
| 866 | + | provider: a.Client.Name(), | |
| 867 | + | model: a.Model, | |
| 868 | + | } | |
| 723 | 869 | req := provider.Request{ | |
| 724 | 870 | Model: a.Model, | |
| 725 | 871 | System: system, | |
| @@ -738,11 +884,7 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext | |||
| 738 | 884 | FastMode: fastMode, | |
| 739 | 885 | MaxTokens: a.MaxTokens, | |
| 740 | 886 | Temperature: a.Temperature, | |
| 741 | - | Lifecycle: requestLifecycleSink{ | |
| 742 | - | sink: sink, | |
| 743 | - | provider: a.Client.Name(), | |
| 744 | - | model: a.Model, | |
| 745 | - | }, | |
| 887 | + | Lifecycle: requestLifecycle, | |
| 746 | 888 | } | |
| 747 | 889 | sink(EvRequestStarted{ | |
| 748 | 890 | Provider: a.Client.Name(), | |
| @@ -117,6 +117,10 @@ type Agent struct { | |||||
| 117 | // current and a crash recovers the right cost figure. | 117 | // current and a crash recovers the right cost figure. | ||
| 118 | OnUsage func(cumulative provider.Usage) | 118 | OnUsage func(cumulative provider.Usage) | ||
| 119 | 119 | ||||
| 120 | + | // OnRetryLifecycle receives sanitized retry audit records. It never | |||
| 121 | + | // receives provider error text or response bodies. | |||
| 122 | + | OnRetryLifecycle func(RetryLifecycleRecord) | |||
| 123 | + | ||||
| 120 | // OnTranscriptCompacted, if set, fires after Compact replaces the | 124 | // OnTranscriptCompacted, if set, fires after Compact replaces the | ||
| 121 | // in-memory transcript with the synthetic summary plus kept tail. | 125 | // in-memory transcript with the synthetic summary plus kept tail. | ||
| 122 | // Hosts wire this to append an explicit compaction checkpoint to | 126 | // Hosts wire this to append an explicit compaction checkpoint to | ||
| @@ -508,7 +512,32 @@ func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error { | |||||
| 508 | for attempt := 0; ; attempt++ { | 512 | for attempt := 0; ; attempt++ { | ||
| 509 | stop, assistantMsg, err = a.oneTurn(ctx, sink, turnContext, attempt, maxAttempts) | 513 | stop, assistantMsg, err = a.oneTurn(ctx, sink, turnContext, attempt, maxAttempts) | ||
| 510 | sink(EvTurnEnd{Stop: stop, Err: err}) | 514 | sink(EvTurnEnd{Stop: stop, Err: err}) | ||
| 511 | - | if err == nil || !a.canRetryError(err, attempt) { | 515 | + | if err == nil { |
| 516 | + | break | |||
| 517 | + | } | |||
| 518 | + | retryable := a.canRetryError(err, attempt) | |||
| 519 | + | var retryCtxErr error | |||
| 520 | + | if retryable { | |||
| 521 | + | retryCtxErr = ctx.Err() | |||
| 522 | + | if retryCtxErr != nil { | |||
| 523 | + | retryable = false | |||
| 524 | + | } | |||
| 525 | + | } | |||
| 526 | + | reason := classifyRetryReason(err) | |||
| 527 | + | if !errors.Is(err, context.Canceled) { | |||
| 528 | + | a.fireRetryLifecycle(RetryLifecycleRecord{ | |||
| 529 | + | Event: RetryLifecycleRequestFailed, | |||
| 530 | + | Scope: RetryScopeAgent, | |||
| 531 | + | Attempt: attempt + 1, | |||
| 532 | + | MaxAttempts: maxAttempts, | |||
| 533 | + | Reason: reason, | |||
| 534 | + | Terminal: !retryable, | |||
| 535 | + | }) | |||
| 536 | + | } | |||
| 537 | + | if retryCtxErr != nil { | |||
| 538 | + | return retryCtxErr | |||
| 539 | + | } | |||
| 540 | + | if !retryable { | |||
| 512 | break | 541 | break | ||
| 513 | } | 542 | } | ||
| 514 | a.dropLastAssistantMessage() | 543 | a.dropLastAssistantMessage() | ||
| @@ -516,6 +545,14 @@ func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error { | |||||
| 516 | if retryErr := ctx.Err(); retryErr != nil { | 545 | if retryErr := ctx.Err(); retryErr != nil { | ||
| 517 | return retryErr | 546 | return retryErr | ||
| 518 | } | 547 | } | ||
| 548 | + | a.fireRetryLifecycle(RetryLifecycleRecord{ | |||
| 549 | + | Event: RetryLifecycleRetryScheduled, | |||
| 550 | + | Scope: RetryScopeAgent, | |||
| 551 | + | Attempt: attempt + 2, | |||
| 552 | + | MaxAttempts: maxAttempts, | |||
| 553 | + | Reason: reason, | |||
| 554 | + | DelayMS: delay.Milliseconds(), | |||
| 555 | + | }) | |||
| 519 | sink(EvRetryScheduled{ | 556 | sink(EvRetryScheduled{ | ||
| 520 | Scope: RetryScopeAgent, | 557 | Scope: RetryScopeAgent, | ||
| 521 | Attempt: attempt + 2, | 558 | Attempt: attempt + 2, | ||
| @@ -643,6 +680,57 @@ func isNonRetryableProviderLimit(msg string) bool { | |||||
| 643 | return false | 680 | return false | ||
| 644 | } | 681 | } | ||
| 645 | 682 | ||||
| 683 | + | func classifyRetryReason(err error) RetryReason { | |||
| 684 | + | if err == nil { | |||
| 685 | + | return RetryReasonUnknown | |||
| 686 | + | } | |||
| 687 | + | msg := strings.ToLower(err.Error()) | |||
| 688 | + | if isNonRetryableProviderLimit(msg) { | |||
| 689 | + | return RetryReasonQuota | |||
| 690 | + | } | |||
| 691 | + | if strings.Contains(msg, "overload") || strings.Contains(msg, "servers are busy") { | |||
| 692 | + | return RetryReasonOverload | |||
| 693 | + | } | |||
| 694 | + | if strings.Contains(msg, "rate limit") || strings.Contains(msg, "ratelimit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "429") { | |||
| 695 | + | return RetryReasonRateLimit | |||
| 696 | + | } | |||
| 697 | + | if errors.Is(err, context.DeadlineExceeded) || strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out") { | |||
| 698 | + | return RetryReasonTimeout | |||
| 699 | + | } | |||
| 700 | + | for _, needle := range []string{"connection", "network", "unexpected eof", "broken pipe", "socket hang up", "transport failure", "upstream connect"} { | |||
| 701 | + | if strings.Contains(msg, needle) { | |||
| 702 | + | return RetryReasonNetwork | |||
| 703 | + | } | |||
| 704 | + | } | |||
| 705 | + | for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} { | |||
| 706 | + | if strings.Contains(msg, needle) { | |||
| 707 | + | return RetryReasonContextWindow | |||
| 708 | + | } | |||
| 709 | + | } | |||
| 710 | + | for _, needle := range []string{"unauthorized", "authentication", "invalid api key", "invalid_api_key", "http 401", "http 403"} { | |||
| 711 | + | if strings.Contains(msg, needle) { | |||
| 712 | + | return RetryReasonAuth | |||
| 713 | + | } | |||
| 714 | + | } | |||
| 715 | + | for _, needle := range []string{"server error", "internal error", "service unavailable", "http 500", "http 502", "http 503", "http 504", "provider returned error"} { | |||
| 716 | + | if strings.Contains(msg, needle) { | |||
| 717 | + | return RetryReasonServer | |||
| 718 | + | } | |||
| 719 | + | } | |||
| 720 | + | for _, needle := range []string{"bad request", "unsupported", "invalid parameter", "http 400", "http 404", "http 422"} { | |||
| 721 | + | if strings.Contains(msg, needle) { | |||
| 722 | + | return RetryReasonClient | |||
| 723 | + | } | |||
| 724 | + | } | |||
| 725 | + | return RetryReasonUnknown | |||
| 726 | + | } | |||
| 727 | + | ||||
| 728 | + | func (a *Agent) fireRetryLifecycle(record RetryLifecycleRecord) { | |||
| 729 | + | if a.OnRetryLifecycle != nil { | |||
| 730 | + | a.OnRetryLifecycle(record) | |||
| 731 | + | } | |||
| 732 | + | } | |||
| 733 | + | ||||
| 646 | func (a *Agent) retryDelay(attempt int) time.Duration { | 734 | func (a *Agent) retryDelay(attempt int) time.Duration { | ||
| 647 | base := a.RetryBaseDelay | 735 | base := a.RetryBaseDelay | ||
| 648 | if base <= 0 { | 736 | if base <= 0 { | ||
| @@ -677,12 +765,14 @@ func (a *Agent) dropLastAssistantMessage() { | |||||
| 677 | // requestLifecycleSink adapts provider-owned pre-stream retries into core | 765 | // requestLifecycleSink adapts provider-owned pre-stream retries into core | ||
| 678 | // events without letting provider concerns leak into mode renderers. | 766 | // events without letting provider concerns leak into mode renderers. | ||
| 679 | type requestLifecycleSink struct { | 767 | type requestLifecycleSink struct { | ||
| 680 | - | sink func(AgentEvent) | 768 | + | sink func(AgentEvent) |
| 681 | - | provider string | 769 | + | observe func(RetryLifecycleRecord) |
| 682 | - | model string | 770 | + | provider string |
| 771 | + | model string | |||
| 772 | + | lastReason RetryReason | |||
| 683 | } | 773 | } | ||
| 684 | 774 | ||||
| 685 | - | func (s requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) { | 775 | + | func (s *requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) { |
| 686 | s.sink(EvRequestStarted{ | 776 | s.sink(EvRequestStarted{ | ||
| 687 | Provider: s.provider, | 777 | Provider: s.provider, | ||
| 688 | Model: s.model, | 778 | Model: s.model, | ||
| @@ -692,13 +782,63 @@ func (s requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) { | |||||
| 692 | }) | 782 | }) | ||
| 693 | } | 783 | } | ||
| 694 | 784 | ||||
| 695 | - | func (s requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) { | 785 | + | func (s *requestLifecycleSink) RequestFailed(attempt, maxAttempts int, reason provider.RequestFailureReason, terminal bool) { |
| 786 | + | s.lastReason = retryReasonFromProvider(reason) | |||
| 787 | + | if s.observe != nil { | |||
| 788 | + | s.observe(RetryLifecycleRecord{ | |||
| 789 | + | Event: RetryLifecycleRequestFailed, | |||
| 790 | + | Scope: RetryScopeProvider, | |||
| 791 | + | Attempt: attempt, | |||
| 792 | + | MaxAttempts: maxAttempts, | |||
| 793 | + | Reason: s.lastReason, | |||
| 794 | + | Terminal: terminal, | |||
| 795 | + | }) | |||
| 796 | + | } | |||
| 797 | + | } | |||
| 798 | + | ||||
| 799 | + | func (s *requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) { | |||
| 696 | s.sink(EvRetryScheduled{ | 800 | s.sink(EvRetryScheduled{ | ||
| 697 | Scope: RetryScopeProvider, | 801 | Scope: RetryScopeProvider, | ||
| 698 | Attempt: attempt, | 802 | Attempt: attempt, | ||
| 699 | MaxAttempts: maxAttempts, | 803 | MaxAttempts: maxAttempts, | ||
| 700 | Delay: delay, | 804 | Delay: delay, | ||
| 701 | }) | 805 | }) | ||
| 806 | + | if s.observe != nil { | |||
| 807 | + | reason := s.lastReason | |||
| 808 | + | if reason == "" { | |||
| 809 | + | reason = RetryReasonUnknown | |||
| 810 | + | } | |||
| 811 | + | s.observe(RetryLifecycleRecord{ | |||
| 812 | + | Event: RetryLifecycleRetryScheduled, | |||
| 813 | + | Scope: RetryScopeProvider, | |||
| 814 | + | Attempt: attempt, | |||
| 815 | + | MaxAttempts: maxAttempts, | |||
| 816 | + | Reason: reason, | |||
| 817 | + | DelayMS: delay.Milliseconds(), | |||
| 818 | + | }) | |||
| 819 | + | } | |||
| 820 | + | s.lastReason = "" | |||
| 821 | + | } | |||
| 822 | + | ||||
| 823 | + | func retryReasonFromProvider(reason provider.RequestFailureReason) RetryReason { | |||
| 824 | + | switch reason { | |||
| 825 | + | case provider.RequestFailureOverload: | |||
| 826 | + | return RetryReasonOverload | |||
| 827 | + | case provider.RequestFailureRateLimit: | |||
| 828 | + | return RetryReasonRateLimit | |||
| 829 | + | case provider.RequestFailureQuota: | |||
| 830 | + | return RetryReasonQuota | |||
| 831 | + | case provider.RequestFailureServer: | |||
| 832 | + | return RetryReasonServer | |||
| 833 | + | case provider.RequestFailureNetwork: | |||
| 834 | + | return RetryReasonNetwork | |||
| 835 | + | case provider.RequestFailureTimeout: | |||
| 836 | + | return RetryReasonTimeout | |||
| 837 | + | case provider.RequestFailureClient: | |||
| 838 | + | return RetryReasonClient | |||
| 839 | + | default: | |||
| 840 | + | return RetryReasonUnknown | |||
| 841 | + | } | |||
| 702 | } | 842 | } | ||
| 703 | 843 | ||||
| 704 | // oneTurn calls the LLM once, forwards events, returns the stop reason | 844 | // oneTurn calls the LLM once, forwards events, returns the stop reason | ||
| @@ -720,6 +860,12 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext | |||||
| 720 | // repair can add stub results for aborted calls; those results must also | 860 | // repair can add stub results for aborted calls; those results must also | ||
| 721 | // remain in the request so every tool call still has a matching result. | 861 | // remain in the request so every tool call still has a matching result. | ||
| 722 | messages := repairToolUseResultPairs(a.Messages()) | 862 | messages := repairToolUseResultPairs(a.Messages()) | ||
| 863 | + | requestLifecycle := &requestLifecycleSink{ | |||
| 864 | + | sink: sink, | |||
| 865 | + | observe: a.fireRetryLifecycle, | |||
| 866 | + | provider: a.Client.Name(), | |||
| 867 | + | model: a.Model, | |||
| 868 | + | } | |||
| 723 | req := provider.Request{ | 869 | req := provider.Request{ | ||
| 724 | Model: a.Model, | 870 | Model: a.Model, | ||
| 725 | System: system, | 871 | System: system, | ||
| @@ -738,11 +884,7 @@ func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext | |||||
| 738 | FastMode: fastMode, | 884 | FastMode: fastMode, | ||
| 739 | MaxTokens: a.MaxTokens, | 885 | MaxTokens: a.MaxTokens, | ||
| 740 | Temperature: a.Temperature, | 886 | Temperature: a.Temperature, | ||
| 741 | - | Lifecycle: requestLifecycleSink{ | 887 | + | Lifecycle: requestLifecycle, |
| 742 | - | sink: sink, | |||
| 743 | - | provider: a.Client.Name(), | |||
| 744 | - | model: a.Model, | |||
| 745 | - | }, | |||
| 746 | } | 888 | } | ||
| 747 | sink(EvRequestStarted{ | 889 | sink(EvRequestStarted{ | ||
| 748 | Provider: a.Client.Name(), | 890 | Provider: a.Client.Name(), | ||
M packages/core/agent_retry_test.go +82 -0
| @@ -5,6 +5,7 @@ import ( | |||
| 5 | 5 | "errors" | |
| 6 | 6 | "fmt" | |
| 7 | 7 | "io" | |
| 8 | + | "reflect" | |
| 8 | 9 | "strings" | |
| 9 | 10 | "sync/atomic" | |
| 10 | 11 | "testing" | |
| @@ -118,6 +119,80 @@ func TestAgentEmitsRetryLifecycleEvents(t *testing.T) { | |||
| 118 | 119 | } | |
| 119 | 120 | } | |
| 120 | 121 | ||
| 122 | + | func 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 | + | ||
| 150 | + | func 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 | + | ||
| 174 | + | func 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 | + | ||
| 121 | 196 | // codexRetryFakeClient reproduces a transient OpenAI Codex backend | |
| 122 | 197 | // failure on the first call and succeeds afterwards. | |
| 123 | 198 | type codexRetryFakeClient struct { | |
| @@ -361,6 +436,10 @@ func TestAgentDoesNotReportRetryAfterCancellation(t *testing.T) { | |||
| 361 | 436 | agent.MaxRetries = 1 | |
| 362 | 437 | ||
| 363 | 438 | var retries []EvRetryScheduled | |
| 439 | + | var lifecycle []RetryLifecycleRecord | |
| 440 | + | agent.OnRetryLifecycle = func(record RetryLifecycleRecord) { | |
| 441 | + | lifecycle = append(lifecycle, record) | |
| 442 | + | } | |
| 364 | 443 | err := agent.Prompt(ctx, "hello", nil, func(event AgentEvent) { | |
| 365 | 444 | if retry, ok := event.(EvRetryScheduled); ok { | |
| 366 | 445 | retries = append(retries, retry) | |
| @@ -372,4 +451,7 @@ func TestAgentDoesNotReportRetryAfterCancellation(t *testing.T) { | |||
| 372 | 451 | if len(retries) != 0 { | |
| 373 | 452 | t.Fatalf("retry events = %#v, want none after cancellation", retries) | |
| 374 | 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 | + | } | |
| 375 | 457 | } | |
| @@ -5,6 +5,7 @@ import ( | |||||
| 5 | "errors" | 5 | "errors" | ||
| 6 | "fmt" | 6 | "fmt" | ||
| 7 | "io" | 7 | "io" | ||
| 8 | + | "reflect" | |||
| 8 | "strings" | 9 | "strings" | ||
| 9 | "sync/atomic" | 10 | "sync/atomic" | ||
| 10 | "testing" | 11 | "testing" | ||
| @@ -118,6 +119,80 @@ func TestAgentEmitsRetryLifecycleEvents(t *testing.T) { | |||||
| 118 | } | 119 | } | ||
| 119 | } | 120 | } | ||
| 120 | 121 | ||||
| 122 | + | func 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 | + | ||||
| 150 | + | func 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 | + | ||||
| 174 | + | func 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 | + | ||||
| 121 | // codexRetryFakeClient reproduces a transient OpenAI Codex backend | 196 | // codexRetryFakeClient reproduces a transient OpenAI Codex backend | ||
| 122 | // failure on the first call and succeeds afterwards. | 197 | // failure on the first call and succeeds afterwards. | ||
| 123 | type codexRetryFakeClient struct { | 198 | type codexRetryFakeClient struct { | ||
| @@ -361,6 +436,10 @@ func TestAgentDoesNotReportRetryAfterCancellation(t *testing.T) { | |||||
| 361 | agent.MaxRetries = 1 | 436 | agent.MaxRetries = 1 | ||
| 362 | 437 | ||||
| 363 | var retries []EvRetryScheduled | 438 | var retries []EvRetryScheduled | ||
| 439 | + | var lifecycle []RetryLifecycleRecord | |||
| 440 | + | agent.OnRetryLifecycle = func(record RetryLifecycleRecord) { | |||
| 441 | + | lifecycle = append(lifecycle, record) | |||
| 442 | + | } | |||
| 364 | err := agent.Prompt(ctx, "hello", nil, func(event AgentEvent) { | 443 | err := agent.Prompt(ctx, "hello", nil, func(event AgentEvent) { | ||
| 365 | if retry, ok := event.(EvRetryScheduled); ok { | 444 | if retry, ok := event.(EvRetryScheduled); ok { | ||
| 366 | retries = append(retries, retry) | 445 | retries = append(retries, retry) | ||
| @@ -372,4 +451,7 @@ func TestAgentDoesNotReportRetryAfterCancellation(t *testing.T) { | |||||
| 372 | if len(retries) != 0 { | 451 | if len(retries) != 0 { | ||
| 373 | t.Fatalf("retry events = %#v, want none after cancellation", retries) | 452 | t.Fatalf("retry events = %#v, want none after cancellation", retries) | ||
| 374 | } | 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 | + | } | |||
| 375 | } | 457 | } | ||
M packages/core/compact.go +2 -1
| @@ -79,8 +79,9 @@ func (a *Agent) compact(ctx context.Context, keepTail int, textSink func(delta s | |||
| 79 | 79 | }, | |
| 80 | 80 | } | |
| 81 | 81 | if eventSink != nil { | |
| 82 | - | req.Lifecycle = requestLifecycleSink{ | |
| 82 | + | req.Lifecycle = &requestLifecycleSink{ | |
| 83 | 83 | sink: eventSink, | |
| 84 | + | observe: a.fireRetryLifecycle, | |
| 84 | 85 | provider: a.Client.Name(), | |
| 85 | 86 | model: a.Model, | |
| 86 | 87 | } | |
| @@ -79,8 +79,9 @@ func (a *Agent) compact(ctx context.Context, keepTail int, textSink func(delta s | |||||
| 79 | }, | 79 | }, | ||
| 80 | } | 80 | } | ||
| 81 | if eventSink != nil { | 81 | if eventSink != nil { | ||
| 82 | - | req.Lifecycle = requestLifecycleSink{ | 82 | + | req.Lifecycle = &requestLifecycleSink{ |
| 83 | sink: eventSink, | 83 | sink: eventSink, | ||
| 84 | + | observe: a.fireRetryLifecycle, | |||
| 84 | provider: a.Client.Name(), | 85 | provider: a.Client.Name(), | ||
| 85 | model: a.Model, | 86 | model: a.Model, | ||
| 86 | } | 87 | } | ||
M packages/core/events.go +36 -0
| @@ -38,6 +38,42 @@ const ( | |||
| 38 | 38 | RetryScopeAgent RetryScope = "agent" | |
| 39 | 39 | ) | |
| 40 | 40 | ||
| 41 | + | // RetryLifecycleEvent identifies a sanitized retry audit record. | |
| 42 | + | type RetryLifecycleEvent string | |
| 43 | + | ||
| 44 | + | const ( | |
| 45 | + | RetryLifecycleRequestFailed RetryLifecycleEvent = "request_failed" | |
| 46 | + | RetryLifecycleRetryScheduled RetryLifecycleEvent = "retry_scheduled" | |
| 47 | + | ) | |
| 48 | + | ||
| 49 | + | // RetryReason is an allowlisted failure category safe for session persistence. | |
| 50 | + | type RetryReason string | |
| 51 | + | ||
| 52 | + | const ( | |
| 53 | + | RetryReasonOverload RetryReason = "overload" | |
| 54 | + | RetryReasonRateLimit RetryReason = "rate_limit" | |
| 55 | + | RetryReasonQuota RetryReason = "quota" | |
| 56 | + | RetryReasonServer RetryReason = "server" | |
| 57 | + | RetryReasonNetwork RetryReason = "network" | |
| 58 | + | RetryReasonTimeout RetryReason = "timeout" | |
| 59 | + | RetryReasonAuth RetryReason = "auth" | |
| 60 | + | RetryReasonContextWindow RetryReason = "context_window" | |
| 61 | + | RetryReasonClient RetryReason = "client" | |
| 62 | + | RetryReasonUnknown RetryReason = "unknown" | |
| 63 | + | ) | |
| 64 | + | ||
| 65 | + | // RetryLifecycleRecord contains only bounded, privacy-safe retry metadata. | |
| 66 | + | // Raw provider errors and response bodies must never be added to this type. | |
| 67 | + | type RetryLifecycleRecord struct { | |
| 68 | + | Event RetryLifecycleEvent `json:"event"` | |
| 69 | + | Scope RetryScope `json:"scope"` | |
| 70 | + | Attempt int `json:"attempt"` | |
| 71 | + | MaxAttempts int `json:"max_attempts"` | |
| 72 | + | Reason RetryReason `json:"reason"` | |
| 73 | + | Terminal bool `json:"terminal,omitempty"` | |
| 74 | + | DelayMS int64 `json:"delay_ms,omitempty"` | |
| 75 | + | } | |
| 76 | + | ||
| 41 | 77 | // EvRequestStarted reports an outbound provider request attempt. | |
| 42 | 78 | type EvRequestStarted struct { | |
| 43 | 79 | Provider string | |
| @@ -38,6 +38,42 @@ const ( | |||||
| 38 | RetryScopeAgent RetryScope = "agent" | 38 | RetryScopeAgent RetryScope = "agent" | ||
| 39 | ) | 39 | ) | ||
| 40 | 40 | ||||
| 41 | + | // RetryLifecycleEvent identifies a sanitized retry audit record. | |||
| 42 | + | type RetryLifecycleEvent string | |||
| 43 | + | ||||
| 44 | + | const ( | |||
| 45 | + | RetryLifecycleRequestFailed RetryLifecycleEvent = "request_failed" | |||
| 46 | + | RetryLifecycleRetryScheduled RetryLifecycleEvent = "retry_scheduled" | |||
| 47 | + | ) | |||
| 48 | + | ||||
| 49 | + | // RetryReason is an allowlisted failure category safe for session persistence. | |||
| 50 | + | type RetryReason string | |||
| 51 | + | ||||
| 52 | + | const ( | |||
| 53 | + | RetryReasonOverload RetryReason = "overload" | |||
| 54 | + | RetryReasonRateLimit RetryReason = "rate_limit" | |||
| 55 | + | RetryReasonQuota RetryReason = "quota" | |||
| 56 | + | RetryReasonServer RetryReason = "server" | |||
| 57 | + | RetryReasonNetwork RetryReason = "network" | |||
| 58 | + | RetryReasonTimeout RetryReason = "timeout" | |||
| 59 | + | RetryReasonAuth RetryReason = "auth" | |||
| 60 | + | RetryReasonContextWindow RetryReason = "context_window" | |||
| 61 | + | RetryReasonClient RetryReason = "client" | |||
| 62 | + | RetryReasonUnknown RetryReason = "unknown" | |||
| 63 | + | ) | |||
| 64 | + | ||||
| 65 | + | // RetryLifecycleRecord contains only bounded, privacy-safe retry metadata. | |||
| 66 | + | // Raw provider errors and response bodies must never be added to this type. | |||
| 67 | + | type RetryLifecycleRecord struct { | |||
| 68 | + | Event RetryLifecycleEvent `json:"event"` | |||
| 69 | + | Scope RetryScope `json:"scope"` | |||
| 70 | + | Attempt int `json:"attempt"` | |||
| 71 | + | MaxAttempts int `json:"max_attempts"` | |||
| 72 | + | Reason RetryReason `json:"reason"` | |||
| 73 | + | Terminal bool `json:"terminal,omitempty"` | |||
| 74 | + | DelayMS int64 `json:"delay_ms,omitempty"` | |||
| 75 | + | } | |||
| 76 | + | ||||
| 41 | // EvRequestStarted reports an outbound provider request attempt. | 77 | // EvRequestStarted reports an outbound provider request attempt. | ||
| 42 | type EvRequestStarted struct { | 78 | type EvRequestStarted struct { | ||
| 43 | Provider string | 79 | Provider string | ||
M packages/core/session.go +98 -13
| @@ -14,6 +14,7 @@ import ( | |||
| 14 | 14 | "path/filepath" | |
| 15 | 15 | "sort" | |
| 16 | 16 | "strings" | |
| 17 | + | "sync" | |
| 17 | 18 | "time" | |
| 18 | 19 | ||
| 19 | 20 | "github.com/bnema/zut/packages/provider" | |
| @@ -45,6 +46,9 @@ type Session struct { | |||
| 45 | 46 | // freshFile it tells Close() whether the session left any content | |
| 46 | 47 | // worth keeping. | |
| 47 | 48 | messagesAppended int | |
| 49 | + | ||
| 50 | + | retryMu sync.Mutex | |
| 51 | + | pendingRetryLifecycle []RetryLifecycleRecord | |
| 48 | 52 | } | |
| 49 | 53 | ||
| 50 | 54 | // GoalStatus is the persisted lifecycle state of an autonomous session goal. | |
| @@ -106,16 +110,17 @@ type SessionMeta struct { | |||
| 106 | 110 | // the default unmarshaler cannot reconstruct); it is written with a | |
| 107 | 111 | // regular provider.Message value. | |
| 108 | 112 | type sessionLine struct { | |
| 109 | - | Type string `json:"type"` | |
| 110 | - | Title string `json:"title,omitempty"` | |
| 111 | - | Generated bool `json:"generated,omitempty"` | |
| 112 | - | Meta *SessionMeta `json:"meta,omitempty"` | |
| 113 | - | Message *provider.Message `json:"message,omitempty"` | |
| 114 | - | Messages *[]provider.Message `json:"messages,omitempty"` | |
| 115 | - | Usage *provider.Usage `json:"usage,omitempty"` | |
| 116 | - | Cumulative *provider.Usage `json:"cumulative,omitempty"` | |
| 117 | - | Extension string `json:"extension,omitempty"` | |
| 118 | - | State json.RawMessage `json:"state,omitempty"` | |
| 113 | + | Type string `json:"type"` | |
| 114 | + | Title string `json:"title,omitempty"` | |
| 115 | + | Generated bool `json:"generated,omitempty"` | |
| 116 | + | Meta *SessionMeta `json:"meta,omitempty"` | |
| 117 | + | Message *provider.Message `json:"message,omitempty"` | |
| 118 | + | Messages *[]provider.Message `json:"messages,omitempty"` | |
| 119 | + | Usage *provider.Usage `json:"usage,omitempty"` | |
| 120 | + | Cumulative *provider.Usage `json:"cumulative,omitempty"` | |
| 121 | + | RetryLifecycle []RetryLifecycleRecord `json:"retry_lifecycle,omitempty"` | |
| 122 | + | Extension string `json:"extension,omitempty"` | |
| 123 | + | State json.RawMessage `json:"state,omitempty"` | |
| 119 | 124 | } | |
| 120 | 125 | ||
| 121 | 126 | type sessionLineHead struct { | |
| @@ -1263,7 +1268,7 @@ func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulat | |||
| 1263 | 1268 | return nil | |
| 1264 | 1269 | } | |
| 1265 | 1270 | compactionMessages := messages | |
| 1266 | - | if err := s.writeLine(sessionLine{ | |
| 1271 | + | if err := s.writeUsageCheckpoint(sessionLine{ | |
| 1267 | 1272 | Type: "compaction", | |
| 1268 | 1273 | Messages: &compactionMessages, | |
| 1269 | 1274 | Usage: &cumulative, | |
| @@ -1352,12 +1357,92 @@ func (s *Session) UpdateTitle(title string) error { | |||
| 1352 | 1357 | return nil | |
| 1353 | 1358 | } | |
| 1354 | 1359 | ||
| 1355 | - | // AppendUsage writes a usage row to the session. | |
| 1360 | + | // RecordRetryLifecycle buffers a sanitized retry record for the next usage | |
| 1361 | + | // checkpoint. Invalid enum values are discarded or reduced to "unknown" so | |
| 1362 | + | // callers cannot use this metadata path to persist provider text. | |
| 1363 | + | func (s *Session) RecordRetryLifecycle(record RetryLifecycleRecord) { | |
| 1364 | + | if s == nil { | |
| 1365 | + | return | |
| 1366 | + | } | |
| 1367 | + | record, ok := sanitizeRetryLifecycleRecord(record) | |
| 1368 | + | if !ok { | |
| 1369 | + | return | |
| 1370 | + | } | |
| 1371 | + | s.retryMu.Lock() | |
| 1372 | + | s.pendingRetryLifecycle = append(s.pendingRetryLifecycle, record) | |
| 1373 | + | s.retryMu.Unlock() | |
| 1374 | + | } | |
| 1375 | + | ||
| 1376 | + | // AppendUsage writes a usage row to the session and includes any retry | |
| 1377 | + | // lifecycle records accumulated since the previous usage checkpoint. | |
| 1356 | 1378 | func (s *Session) AppendUsage(u, cum provider.Usage) error { | |
| 1357 | 1379 | if s == nil { | |
| 1358 | 1380 | return nil | |
| 1359 | 1381 | } | |
| 1360 | - | return s.writeLine(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}) | |
| 1382 | + | return s.writeUsageCheckpoint(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}) | |
| 1383 | + | } | |
| 1384 | + | ||
| 1385 | + | // FlushRetryLifecycle writes a usage checkpoint only when retry records are | |
| 1386 | + | // pending. Long-running hosts use it during shutdown so a terminal failure | |
| 1387 | + | // without a usage event is still diagnosable. | |
| 1388 | + | func (s *Session) FlushRetryLifecycle(u, cum provider.Usage) error { | |
| 1389 | + | if s == nil { | |
| 1390 | + | return nil | |
| 1391 | + | } | |
| 1392 | + | s.retryMu.Lock() | |
| 1393 | + | defer s.retryMu.Unlock() | |
| 1394 | + | if len(s.pendingRetryLifecycle) == 0 { | |
| 1395 | + | return nil | |
| 1396 | + | } | |
| 1397 | + | return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}) | |
| 1398 | + | } | |
| 1399 | + | ||
| 1400 | + | func (s *Session) writeUsageCheckpoint(row sessionLine) error { | |
| 1401 | + | s.retryMu.Lock() | |
| 1402 | + | defer s.retryMu.Unlock() | |
| 1403 | + | return s.writeUsageCheckpointLocked(row) | |
| 1404 | + | } | |
| 1405 | + | ||
| 1406 | + | func (s *Session) writeUsageCheckpointLocked(row sessionLine) error { | |
| 1407 | + | if len(s.pendingRetryLifecycle) > 0 { | |
| 1408 | + | row.RetryLifecycle = append([]RetryLifecycleRecord(nil), s.pendingRetryLifecycle...) | |
| 1409 | + | } | |
| 1410 | + | if err := s.writeLine(row); err != nil { | |
| 1411 | + | return err | |
| 1412 | + | } | |
| 1413 | + | s.pendingRetryLifecycle = nil | |
| 1414 | + | return nil | |
| 1415 | + | } | |
| 1416 | + | ||
| 1417 | + | func sanitizeRetryLifecycleRecord(record RetryLifecycleRecord) (RetryLifecycleRecord, bool) { | |
| 1418 | + | switch record.Event { | |
| 1419 | + | case RetryLifecycleRequestFailed: | |
| 1420 | + | record.DelayMS = 0 | |
| 1421 | + | case RetryLifecycleRetryScheduled: | |
| 1422 | + | record.Terminal = false | |
| 1423 | + | if record.DelayMS < 0 { | |
| 1424 | + | record.DelayMS = 0 | |
| 1425 | + | } | |
| 1426 | + | default: | |
| 1427 | + | return RetryLifecycleRecord{}, false | |
| 1428 | + | } | |
| 1429 | + | switch record.Scope { | |
| 1430 | + | case RetryScopeProvider, RetryScopeAgent: | |
| 1431 | + | default: | |
| 1432 | + | return RetryLifecycleRecord{}, false | |
| 1433 | + | } | |
| 1434 | + | if record.Attempt < 1 || record.MaxAttempts < record.Attempt { | |
| 1435 | + | return RetryLifecycleRecord{}, false | |
| 1436 | + | } | |
| 1437 | + | switch record.Reason { | |
| 1438 | + | case RetryReasonOverload, RetryReasonRateLimit, RetryReasonQuota, | |
| 1439 | + | RetryReasonServer, RetryReasonNetwork, RetryReasonTimeout, | |
| 1440 | + | RetryReasonAuth, RetryReasonContextWindow, RetryReasonClient, | |
| 1441 | + | RetryReasonUnknown: | |
| 1442 | + | default: | |
| 1443 | + | record.Reason = RetryReasonUnknown | |
| 1444 | + | } | |
| 1445 | + | return record, true | |
| 1361 | 1446 | } | |
| 1362 | 1447 | ||
| 1363 | 1448 | const maxExtensionStateBytes = 256 * 1024 | |
| @@ -14,6 +14,7 @@ import ( | |||||
| 14 | "path/filepath" | 14 | "path/filepath" | ||
| 15 | "sort" | 15 | "sort" | ||
| 16 | "strings" | 16 | "strings" | ||
| 17 | + | "sync" | |||
| 17 | "time" | 18 | "time" | ||
| 18 | 19 | ||||
| 19 | "github.com/bnema/zut/packages/provider" | 20 | "github.com/bnema/zut/packages/provider" | ||
| @@ -45,6 +46,9 @@ type Session struct { | |||||
| 45 | // freshFile it tells Close() whether the session left any content | 46 | // freshFile it tells Close() whether the session left any content | ||
| 46 | // worth keeping. | 47 | // worth keeping. | ||
| 47 | messagesAppended int | 48 | messagesAppended int | ||
| 49 | + | ||||
| 50 | + | retryMu sync.Mutex | |||
| 51 | + | pendingRetryLifecycle []RetryLifecycleRecord | |||
| 48 | } | 52 | } | ||
| 49 | 53 | ||||
| 50 | // GoalStatus is the persisted lifecycle state of an autonomous session goal. | 54 | // GoalStatus is the persisted lifecycle state of an autonomous session goal. | ||
| @@ -106,16 +110,17 @@ type SessionMeta struct { | |||||
| 106 | // the default unmarshaler cannot reconstruct); it is written with a | 110 | // the default unmarshaler cannot reconstruct); it is written with a | ||
| 107 | // regular provider.Message value. | 111 | // regular provider.Message value. | ||
| 108 | type sessionLine struct { | 112 | type sessionLine struct { | ||
| 109 | - | Type string `json:"type"` | 113 | + | Type string `json:"type"` |
| 110 | - | Title string `json:"title,omitempty"` | 114 | + | Title string `json:"title,omitempty"` |
| 111 | - | Generated bool `json:"generated,omitempty"` | 115 | + | Generated bool `json:"generated,omitempty"` |
| 112 | - | Meta *SessionMeta `json:"meta,omitempty"` | 116 | + | Meta *SessionMeta `json:"meta,omitempty"` |
| 113 | - | Message *provider.Message `json:"message,omitempty"` | 117 | + | Message *provider.Message `json:"message,omitempty"` |
| 114 | - | Messages *[]provider.Message `json:"messages,omitempty"` | 118 | + | Messages *[]provider.Message `json:"messages,omitempty"` |
| 115 | - | Usage *provider.Usage `json:"usage,omitempty"` | 119 | + | Usage *provider.Usage `json:"usage,omitempty"` |
| 116 | - | Cumulative *provider.Usage `json:"cumulative,omitempty"` | 120 | + | Cumulative *provider.Usage `json:"cumulative,omitempty"` |
| 117 | - | Extension string `json:"extension,omitempty"` | 121 | + | RetryLifecycle []RetryLifecycleRecord `json:"retry_lifecycle,omitempty"` |
| 118 | - | State json.RawMessage `json:"state,omitempty"` | 122 | + | Extension string `json:"extension,omitempty"` |
| 123 | + | State json.RawMessage `json:"state,omitempty"` | |||
| 119 | } | 124 | } | ||
| 120 | 125 | ||||
| 121 | type sessionLineHead struct { | 126 | type sessionLineHead struct { | ||
| @@ -1263,7 +1268,7 @@ func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulat | |||||
| 1263 | return nil | 1268 | return nil | ||
| 1264 | } | 1269 | } | ||
| 1265 | compactionMessages := messages | 1270 | compactionMessages := messages | ||
| 1266 | - | if err := s.writeLine(sessionLine{ | 1271 | + | if err := s.writeUsageCheckpoint(sessionLine{ |
| 1267 | Type: "compaction", | 1272 | Type: "compaction", | ||
| 1268 | Messages: &compactionMessages, | 1273 | Messages: &compactionMessages, | ||
| 1269 | Usage: &cumulative, | 1274 | Usage: &cumulative, | ||
| @@ -1352,12 +1357,92 @@ func (s *Session) UpdateTitle(title string) error { | |||||
| 1352 | return nil | 1357 | return nil | ||
| 1353 | } | 1358 | } | ||
| 1354 | 1359 | ||||
| 1355 | - | // AppendUsage writes a usage row to the session. | 1360 | + | // RecordRetryLifecycle buffers a sanitized retry record for the next usage |
| 1361 | + | // checkpoint. Invalid enum values are discarded or reduced to "unknown" so | |||
| 1362 | + | // callers cannot use this metadata path to persist provider text. | |||
| 1363 | + | func (s *Session) RecordRetryLifecycle(record RetryLifecycleRecord) { | |||
| 1364 | + | if s == nil { | |||
| 1365 | + | return | |||
| 1366 | + | } | |||
| 1367 | + | record, ok := sanitizeRetryLifecycleRecord(record) | |||
| 1368 | + | if !ok { | |||
| 1369 | + | return | |||
| 1370 | + | } | |||
| 1371 | + | s.retryMu.Lock() | |||
| 1372 | + | s.pendingRetryLifecycle = append(s.pendingRetryLifecycle, record) | |||
| 1373 | + | s.retryMu.Unlock() | |||
| 1374 | + | } | |||
| 1375 | + | ||||
| 1376 | + | // AppendUsage writes a usage row to the session and includes any retry | |||
| 1377 | + | // lifecycle records accumulated since the previous usage checkpoint. | |||
| 1356 | func (s *Session) AppendUsage(u, cum provider.Usage) error { | 1378 | func (s *Session) AppendUsage(u, cum provider.Usage) error { | ||
| 1357 | if s == nil { | 1379 | if s == nil { | ||
| 1358 | return nil | 1380 | return nil | ||
| 1359 | } | 1381 | } | ||
| 1360 | - | return s.writeLine(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}) | 1382 | + | return s.writeUsageCheckpoint(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}) |
| 1383 | + | } | |||
| 1384 | + | ||||
| 1385 | + | // FlushRetryLifecycle writes a usage checkpoint only when retry records are | |||
| 1386 | + | // pending. Long-running hosts use it during shutdown so a terminal failure | |||
| 1387 | + | // without a usage event is still diagnosable. | |||
| 1388 | + | func (s *Session) FlushRetryLifecycle(u, cum provider.Usage) error { | |||
| 1389 | + | if s == nil { | |||
| 1390 | + | return nil | |||
| 1391 | + | } | |||
| 1392 | + | s.retryMu.Lock() | |||
| 1393 | + | defer s.retryMu.Unlock() | |||
| 1394 | + | if len(s.pendingRetryLifecycle) == 0 { | |||
| 1395 | + | return nil | |||
| 1396 | + | } | |||
| 1397 | + | return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}) | |||
| 1398 | + | } | |||
| 1399 | + | ||||
| 1400 | + | func (s *Session) writeUsageCheckpoint(row sessionLine) error { | |||
| 1401 | + | s.retryMu.Lock() | |||
| 1402 | + | defer s.retryMu.Unlock() | |||
| 1403 | + | return s.writeUsageCheckpointLocked(row) | |||
| 1404 | + | } | |||
| 1405 | + | ||||
| 1406 | + | func (s *Session) writeUsageCheckpointLocked(row sessionLine) error { | |||
| 1407 | + | if len(s.pendingRetryLifecycle) > 0 { | |||
| 1408 | + | row.RetryLifecycle = append([]RetryLifecycleRecord(nil), s.pendingRetryLifecycle...) | |||
| 1409 | + | } | |||
| 1410 | + | if err := s.writeLine(row); err != nil { | |||
| 1411 | + | return err | |||
| 1412 | + | } | |||
| 1413 | + | s.pendingRetryLifecycle = nil | |||
| 1414 | + | return nil | |||
| 1415 | + | } | |||
| 1416 | + | ||||
| 1417 | + | func sanitizeRetryLifecycleRecord(record RetryLifecycleRecord) (RetryLifecycleRecord, bool) { | |||
| 1418 | + | switch record.Event { | |||
| 1419 | + | case RetryLifecycleRequestFailed: | |||
| 1420 | + | record.DelayMS = 0 | |||
| 1421 | + | case RetryLifecycleRetryScheduled: | |||
| 1422 | + | record.Terminal = false | |||
| 1423 | + | if record.DelayMS < 0 { | |||
| 1424 | + | record.DelayMS = 0 | |||
| 1425 | + | } | |||
| 1426 | + | default: | |||
| 1427 | + | return RetryLifecycleRecord{}, false | |||
| 1428 | + | } | |||
| 1429 | + | switch record.Scope { | |||
| 1430 | + | case RetryScopeProvider, RetryScopeAgent: | |||
| 1431 | + | default: | |||
| 1432 | + | return RetryLifecycleRecord{}, false | |||
| 1433 | + | } | |||
| 1434 | + | if record.Attempt < 1 || record.MaxAttempts < record.Attempt { | |||
| 1435 | + | return RetryLifecycleRecord{}, false | |||
| 1436 | + | } | |||
| 1437 | + | switch record.Reason { | |||
| 1438 | + | case RetryReasonOverload, RetryReasonRateLimit, RetryReasonQuota, | |||
| 1439 | + | RetryReasonServer, RetryReasonNetwork, RetryReasonTimeout, | |||
| 1440 | + | RetryReasonAuth, RetryReasonContextWindow, RetryReasonClient, | |||
| 1441 | + | RetryReasonUnknown: | |||
| 1442 | + | default: | |||
| 1443 | + | record.Reason = RetryReasonUnknown | |||
| 1444 | + | } | |||
| 1445 | + | return record, true | |||
| 1361 | } | 1446 | } | ||
| 1362 | 1447 | ||||
| 1363 | const maxExtensionStateBytes = 256 * 1024 | 1448 | const maxExtensionStateBytes = 256 * 1024 | ||
A packages/core/session_retry_lifecycle_test.go +121 -0
| @@ -0,0 +1,121 @@ | |||
| 1 | + | package core | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "encoding/json" | |
| 5 | + | "os" | |
| 6 | + | "strings" | |
| 7 | + | "testing" | |
| 8 | + | ||
| 9 | + | "github.com/bnema/zut/packages/provider" | |
| 10 | + | ) | |
| 11 | + | ||
| 12 | + | func TestSessionPersistsSanitizedRetryLifecycleOnUsage(t *testing.T) { | |
| 13 | + | const privateDetail = "private provider response body" | |
| 14 | + | ||
| 15 | + | session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test") | |
| 16 | + | if err != nil { | |
| 17 | + | t.Fatal(err) | |
| 18 | + | } | |
| 19 | + | if err := session.AppendMessage(textMessage(provider.RoleUser, "hello")); err != nil { | |
| 20 | + | t.Fatal(err) | |
| 21 | + | } | |
| 22 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |
| 23 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeAgent, | |
| 24 | + | Attempt: 1, MaxAttempts: 4, Reason: RetryReasonOverload, | |
| 25 | + | }) | |
| 26 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |
| 27 | + | Event: RetryLifecycleRetryScheduled, Scope: RetryScopeAgent, | |
| 28 | + | Attempt: 2, MaxAttempts: 4, Reason: RetryReasonOverload, DelayMS: 2000, | |
| 29 | + | }) | |
| 30 | + | // RetryReason has a string representation for JSON, so the session layer | |
| 31 | + | // clamps unknown values rather than trusting callers with persisted text. | |
| 32 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |
| 33 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeProvider, | |
| 34 | + | Attempt: 1, MaxAttempts: 1, Reason: RetryReason(privateDetail), Terminal: true, | |
| 35 | + | }) | |
| 36 | + | if err := session.AppendUsage(provider.Usage{InputTokens: 1}, provider.Usage{InputTokens: 1}); err != nil { | |
| 37 | + | t.Fatal(err) | |
| 38 | + | } | |
| 39 | + | path := session.Path | |
| 40 | + | if err := session.Close(); err != nil { | |
| 41 | + | t.Fatal(err) | |
| 42 | + | } | |
| 43 | + | ||
| 44 | + | raw, err := os.ReadFile(path) | |
| 45 | + | if err != nil { | |
| 46 | + | t.Fatal(err) | |
| 47 | + | } | |
| 48 | + | if strings.Contains(string(raw), privateDetail) { | |
| 49 | + | t.Fatalf("session leaked untrusted lifecycle text: %s", raw) | |
| 50 | + | } | |
| 51 | + | ||
| 52 | + | var lifecycle []RetryLifecycleRecord | |
| 53 | + | for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") { | |
| 54 | + | var row struct { | |
| 55 | + | Type string `json:"type"` | |
| 56 | + | RetryLifecycle []RetryLifecycleRecord `json:"retry_lifecycle"` | |
| 57 | + | } | |
| 58 | + | if err := json.Unmarshal([]byte(line), &row); err != nil { | |
| 59 | + | t.Fatal(err) | |
| 60 | + | } | |
| 61 | + | if row.Type == "usage" { | |
| 62 | + | lifecycle = row.RetryLifecycle | |
| 63 | + | } | |
| 64 | + | } | |
| 65 | + | if len(lifecycle) != 3 { | |
| 66 | + | t.Fatalf("retry lifecycle = %#v, want 3 records", lifecycle) | |
| 67 | + | } | |
| 68 | + | if lifecycle[0].Reason != RetryReasonOverload || lifecycle[1].DelayMS != 2000 || lifecycle[2].Reason != RetryReasonUnknown || !lifecycle[2].Terminal { | |
| 69 | + | t.Fatalf("retry lifecycle = %#v", lifecycle) | |
| 70 | + | } | |
| 71 | + | ||
| 72 | + | snapshot, err := ReadSessionSnapshot(path) | |
| 73 | + | if err != nil { | |
| 74 | + | t.Fatalf("ReadSessionSnapshot with retry metadata: %v", err) | |
| 75 | + | } | |
| 76 | + | if len(snapshot.Messages) != 1 || extractText(snapshot.Messages[0]) != "hello" { | |
| 77 | + | t.Fatalf("snapshot messages changed by retry metadata: %#v", snapshot.Messages) | |
| 78 | + | } | |
| 79 | + | } | |
| 80 | + | ||
| 81 | + | func TestSessionFlushRetryLifecycleWritesOnlyWhenPending(t *testing.T) { | |
| 82 | + | session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test") | |
| 83 | + | if err != nil { | |
| 84 | + | t.Fatal(err) | |
| 85 | + | } | |
| 86 | + | if err := session.AppendMessage(textMessage(provider.RoleUser, "hello")); err != nil { | |
| 87 | + | t.Fatal(err) | |
| 88 | + | } | |
| 89 | + | before, err := os.Stat(session.Path) | |
| 90 | + | if err != nil { | |
| 91 | + | t.Fatal(err) | |
| 92 | + | } | |
| 93 | + | if err := session.FlushRetryLifecycle(provider.Usage{}, provider.Usage{}); err != nil { | |
| 94 | + | t.Fatal(err) | |
| 95 | + | } | |
| 96 | + | after, err := os.Stat(session.Path) | |
| 97 | + | if err != nil { | |
| 98 | + | t.Fatal(err) | |
| 99 | + | } | |
| 100 | + | if after.Size() != before.Size() { | |
| 101 | + | t.Fatalf("empty lifecycle flush changed session size: %d -> %d", before.Size(), after.Size()) | |
| 102 | + | } | |
| 103 | + | ||
| 104 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |
| 105 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeAgent, | |
| 106 | + | Attempt: 1, MaxAttempts: 1, Reason: RetryReasonServer, Terminal: true, | |
| 107 | + | }) | |
| 108 | + | if err := session.FlushRetryLifecycle(provider.Usage{}, provider.Usage{}); err != nil { | |
| 109 | + | t.Fatal(err) | |
| 110 | + | } | |
| 111 | + | if err := session.Close(); err != nil { | |
| 112 | + | t.Fatal(err) | |
| 113 | + | } | |
| 114 | + | raw, err := os.ReadFile(session.Path) | |
| 115 | + | if err != nil { | |
| 116 | + | t.Fatal(err) | |
| 117 | + | } | |
| 118 | + | if !strings.Contains(string(raw), `"retry_lifecycle"`) || !strings.Contains(string(raw), `"terminal":true`) { | |
| 119 | + | t.Fatalf("terminal lifecycle was not flushed: %s", raw) | |
| 120 | + | } | |
| 121 | + | } | |
| @@ -0,0 +1,121 @@ | |||||
| 1 | + | package core | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "encoding/json" | |||
| 5 | + | "os" | |||
| 6 | + | "strings" | |||
| 7 | + | "testing" | |||
| 8 | + | ||||
| 9 | + | "github.com/bnema/zut/packages/provider" | |||
| 10 | + | ) | |||
| 11 | + | ||||
| 12 | + | func TestSessionPersistsSanitizedRetryLifecycleOnUsage(t *testing.T) { | |||
| 13 | + | const privateDetail = "private provider response body" | |||
| 14 | + | ||||
| 15 | + | session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test") | |||
| 16 | + | if err != nil { | |||
| 17 | + | t.Fatal(err) | |||
| 18 | + | } | |||
| 19 | + | if err := session.AppendMessage(textMessage(provider.RoleUser, "hello")); err != nil { | |||
| 20 | + | t.Fatal(err) | |||
| 21 | + | } | |||
| 22 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |||
| 23 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeAgent, | |||
| 24 | + | Attempt: 1, MaxAttempts: 4, Reason: RetryReasonOverload, | |||
| 25 | + | }) | |||
| 26 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |||
| 27 | + | Event: RetryLifecycleRetryScheduled, Scope: RetryScopeAgent, | |||
| 28 | + | Attempt: 2, MaxAttempts: 4, Reason: RetryReasonOverload, DelayMS: 2000, | |||
| 29 | + | }) | |||
| 30 | + | // RetryReason has a string representation for JSON, so the session layer | |||
| 31 | + | // clamps unknown values rather than trusting callers with persisted text. | |||
| 32 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |||
| 33 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeProvider, | |||
| 34 | + | Attempt: 1, MaxAttempts: 1, Reason: RetryReason(privateDetail), Terminal: true, | |||
| 35 | + | }) | |||
| 36 | + | if err := session.AppendUsage(provider.Usage{InputTokens: 1}, provider.Usage{InputTokens: 1}); err != nil { | |||
| 37 | + | t.Fatal(err) | |||
| 38 | + | } | |||
| 39 | + | path := session.Path | |||
| 40 | + | if err := session.Close(); err != nil { | |||
| 41 | + | t.Fatal(err) | |||
| 42 | + | } | |||
| 43 | + | ||||
| 44 | + | raw, err := os.ReadFile(path) | |||
| 45 | + | if err != nil { | |||
| 46 | + | t.Fatal(err) | |||
| 47 | + | } | |||
| 48 | + | if strings.Contains(string(raw), privateDetail) { | |||
| 49 | + | t.Fatalf("session leaked untrusted lifecycle text: %s", raw) | |||
| 50 | + | } | |||
| 51 | + | ||||
| 52 | + | var lifecycle []RetryLifecycleRecord | |||
| 53 | + | for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") { | |||
| 54 | + | var row struct { | |||
| 55 | + | Type string `json:"type"` | |||
| 56 | + | RetryLifecycle []RetryLifecycleRecord `json:"retry_lifecycle"` | |||
| 57 | + | } | |||
| 58 | + | if err := json.Unmarshal([]byte(line), &row); err != nil { | |||
| 59 | + | t.Fatal(err) | |||
| 60 | + | } | |||
| 61 | + | if row.Type == "usage" { | |||
| 62 | + | lifecycle = row.RetryLifecycle | |||
| 63 | + | } | |||
| 64 | + | } | |||
| 65 | + | if len(lifecycle) != 3 { | |||
| 66 | + | t.Fatalf("retry lifecycle = %#v, want 3 records", lifecycle) | |||
| 67 | + | } | |||
| 68 | + | if lifecycle[0].Reason != RetryReasonOverload || lifecycle[1].DelayMS != 2000 || lifecycle[2].Reason != RetryReasonUnknown || !lifecycle[2].Terminal { | |||
| 69 | + | t.Fatalf("retry lifecycle = %#v", lifecycle) | |||
| 70 | + | } | |||
| 71 | + | ||||
| 72 | + | snapshot, err := ReadSessionSnapshot(path) | |||
| 73 | + | if err != nil { | |||
| 74 | + | t.Fatalf("ReadSessionSnapshot with retry metadata: %v", err) | |||
| 75 | + | } | |||
| 76 | + | if len(snapshot.Messages) != 1 || extractText(snapshot.Messages[0]) != "hello" { | |||
| 77 | + | t.Fatalf("snapshot messages changed by retry metadata: %#v", snapshot.Messages) | |||
| 78 | + | } | |||
| 79 | + | } | |||
| 80 | + | ||||
| 81 | + | func TestSessionFlushRetryLifecycleWritesOnlyWhenPending(t *testing.T) { | |||
| 82 | + | session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test") | |||
| 83 | + | if err != nil { | |||
| 84 | + | t.Fatal(err) | |||
| 85 | + | } | |||
| 86 | + | if err := session.AppendMessage(textMessage(provider.RoleUser, "hello")); err != nil { | |||
| 87 | + | t.Fatal(err) | |||
| 88 | + | } | |||
| 89 | + | before, err := os.Stat(session.Path) | |||
| 90 | + | if err != nil { | |||
| 91 | + | t.Fatal(err) | |||
| 92 | + | } | |||
| 93 | + | if err := session.FlushRetryLifecycle(provider.Usage{}, provider.Usage{}); err != nil { | |||
| 94 | + | t.Fatal(err) | |||
| 95 | + | } | |||
| 96 | + | after, err := os.Stat(session.Path) | |||
| 97 | + | if err != nil { | |||
| 98 | + | t.Fatal(err) | |||
| 99 | + | } | |||
| 100 | + | if after.Size() != before.Size() { | |||
| 101 | + | t.Fatalf("empty lifecycle flush changed session size: %d -> %d", before.Size(), after.Size()) | |||
| 102 | + | } | |||
| 103 | + | ||||
| 104 | + | session.RecordRetryLifecycle(RetryLifecycleRecord{ | |||
| 105 | + | Event: RetryLifecycleRequestFailed, Scope: RetryScopeAgent, | |||
| 106 | + | Attempt: 1, MaxAttempts: 1, Reason: RetryReasonServer, Terminal: true, | |||
| 107 | + | }) | |||
| 108 | + | if err := session.FlushRetryLifecycle(provider.Usage{}, provider.Usage{}); err != nil { | |||
| 109 | + | t.Fatal(err) | |||
| 110 | + | } | |||
| 111 | + | if err := session.Close(); err != nil { | |||
| 112 | + | t.Fatal(err) | |||
| 113 | + | } | |||
| 114 | + | raw, err := os.ReadFile(session.Path) | |||
| 115 | + | if err != nil { | |||
| 116 | + | t.Fatal(err) | |||
| 117 | + | } | |||
| 118 | + | if !strings.Contains(string(raw), `"retry_lifecycle"`) || !strings.Contains(string(raw), `"terminal":true`) { | |||
| 119 | + | t.Fatalf("terminal lifecycle was not flushed: %s", raw) | |||
| 120 | + | } | |||
| 121 | + | } | |||
M packages/provider/provider.go +22 -0
| @@ -266,6 +266,28 @@ type RequestLifecycle interface { | |||
| 266 | 266 | RetryScheduled(attempt, maxAttempts int, delay time.Duration) | |
| 267 | 267 | } | |
| 268 | 268 | ||
| 269 | + | // RequestFailureReason is an allowlisted request-failure category. It is safe | |
| 270 | + | // to persist; raw provider errors and response bodies are not. | |
| 271 | + | type RequestFailureReason string | |
| 272 | + | ||
| 273 | + | const ( | |
| 274 | + | RequestFailureOverload RequestFailureReason = "overload" | |
| 275 | + | RequestFailureRateLimit RequestFailureReason = "rate_limit" | |
| 276 | + | RequestFailureQuota RequestFailureReason = "quota" | |
| 277 | + | RequestFailureServer RequestFailureReason = "server" | |
| 278 | + | RequestFailureNetwork RequestFailureReason = "network" | |
| 279 | + | RequestFailureTimeout RequestFailureReason = "timeout" | |
| 280 | + | RequestFailureClient RequestFailureReason = "client" | |
| 281 | + | RequestFailureUnknown RequestFailureReason = "unknown" | |
| 282 | + | ) | |
| 283 | + | ||
| 284 | + | // RequestFailureLifecycle optionally augments RequestLifecycle with sanitized | |
| 285 | + | // failed-attempt metadata. Keeping it separate preserves existing lifecycle | |
| 286 | + | // implementations. | |
| 287 | + | type RequestFailureLifecycle interface { | |
| 288 | + | RequestFailed(attempt, maxAttempts int, reason RequestFailureReason, terminal bool) | |
| 289 | + | } | |
| 290 | + | ||
| 269 | 291 | // Request is a single LLM call. | |
| 270 | 292 | type Request struct { | |
| 271 | 293 | Model string | |
| @@ -266,6 +266,28 @@ type RequestLifecycle interface { | |||||
| 266 | RetryScheduled(attempt, maxAttempts int, delay time.Duration) | 266 | RetryScheduled(attempt, maxAttempts int, delay time.Duration) | ||
| 267 | } | 267 | } | ||
| 268 | 268 | ||||
| 269 | + | // RequestFailureReason is an allowlisted request-failure category. It is safe | |||
| 270 | + | // to persist; raw provider errors and response bodies are not. | |||
| 271 | + | type RequestFailureReason string | |||
| 272 | + | ||||
| 273 | + | const ( | |||
| 274 | + | RequestFailureOverload RequestFailureReason = "overload" | |||
| 275 | + | RequestFailureRateLimit RequestFailureReason = "rate_limit" | |||
| 276 | + | RequestFailureQuota RequestFailureReason = "quota" | |||
| 277 | + | RequestFailureServer RequestFailureReason = "server" | |||
| 278 | + | RequestFailureNetwork RequestFailureReason = "network" | |||
| 279 | + | RequestFailureTimeout RequestFailureReason = "timeout" | |||
| 280 | + | RequestFailureClient RequestFailureReason = "client" | |||
| 281 | + | RequestFailureUnknown RequestFailureReason = "unknown" | |||
| 282 | + | ) | |||
| 283 | + | ||||
| 284 | + | // RequestFailureLifecycle optionally augments RequestLifecycle with sanitized | |||
| 285 | + | // failed-attempt metadata. Keeping it separate preserves existing lifecycle | |||
| 286 | + | // implementations. | |||
| 287 | + | type RequestFailureLifecycle interface { | |||
| 288 | + | RequestFailed(attempt, maxAttempts int, reason RequestFailureReason, terminal bool) | |||
| 289 | + | } | |||
| 290 | + | ||||
| 269 | // Request is a single LLM call. | 291 | // Request is a single LLM call. | ||
| 270 | type Request struct { | 292 | type Request struct { | ||
| 271 | Model string | 293 | Model string | ||
M packages/provider/retry.go +41 -1
| @@ -102,7 +102,12 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||
| 102 | 102 | resp, err := client.Do(req) | |
| 103 | 103 | if err != nil { | |
| 104 | 104 | lastErr = err | |
| 105 | - | if !isTransientConnectError(err) || ctx.Err() != nil { | |
| 105 | + | if ctx.Err() != nil { | |
| 106 | + | return nil, err | |
| 107 | + | } | |
| 108 | + | transient := isTransientConnectError(err) | |
| 109 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, requestFailureReasonForError(err), !transient || attempt == streamRetryAttempts) | |
| 110 | + | if !transient { | |
| 106 | 111 | return nil, err | |
| 107 | 112 | } | |
| 108 | 113 | continue | |
| @@ -114,15 +119,18 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||
| 114 | 119 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) | |
| 115 | 120 | resp.Body.Close() | |
| 116 | 121 | trimmed := strings.TrimSpace(string(body)) | |
| 122 | + | reason := requestFailureReasonForHTTP(resp.StatusCode, trimmed) | |
| 117 | 123 | if resp.StatusCode == http.StatusTooManyRequests && isTerminalRateLimitBody(trimmed) { | |
| 118 | 124 | // Subscription/quota/billing limits are not transient | |
| 119 | 125 | // throttles; retrying only burns time. Surface the | |
| 120 | 126 | // response immediately so the caller formats the | |
| 121 | 127 | // normal "http 429: body" error. | |
| 128 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, RequestFailureQuota, true) | |
| 122 | 129 | return synthesizeResponse(resp.StatusCode, body), nil | |
| 123 | 130 | } | |
| 124 | 131 | lastErr = &transientHTTPError{Status: resp.StatusCode, Body: trimmed} | |
| 125 | 132 | if attempt == streamRetryAttempts { | |
| 133 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, true) | |
| 126 | 134 | // Wrap as a real *http.Response shape the caller | |
| 127 | 135 | // expects so it formats the error identically to a | |
| 128 | 136 | // non-retried failure. | |
| @@ -130,11 +138,13 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||
| 130 | 138 | } | |
| 131 | 139 | if d, ok := retryAfterDelay(resp.Header); ok { | |
| 132 | 140 | if d > maxServerRetryDelay { | |
| 141 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, true) | |
| 133 | 142 | return nil, fmt.Errorf("server requested %s retry delay (max %s): http %d: %s", | |
| 134 | 143 | d.Round(time.Second), maxServerRetryDelay, resp.StatusCode, trimmed) | |
| 135 | 144 | } | |
| 136 | 145 | serverDelay = d | |
| 137 | 146 | } | |
| 147 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, false) | |
| 138 | 148 | continue | |
| 139 | 149 | } | |
| 140 | 150 | return resp, nil | |
| @@ -157,6 +167,36 @@ type transientHTTPError struct { | |||
| 157 | 167 | ||
| 158 | 168 | func (e *transientHTTPError) Error() string { return "transient http error" } | |
| 159 | 169 | ||
| 170 | + | func reportRequestFailure(lifecycle RequestLifecycle, attempt, maxAttempts int, reason RequestFailureReason, terminal bool) { | |
| 171 | + | if observer, ok := lifecycle.(RequestFailureLifecycle); ok { | |
| 172 | + | observer.RequestFailed(attempt, maxAttempts, reason, terminal) | |
| 173 | + | } | |
| 174 | + | } | |
| 175 | + | ||
| 176 | + | func requestFailureReasonForHTTP(status int, body string) RequestFailureReason { | |
| 177 | + | msg := strings.ToLower(body) | |
| 178 | + | if status == 529 || strings.Contains(msg, "overload") { | |
| 179 | + | return RequestFailureOverload | |
| 180 | + | } | |
| 181 | + | if status == http.StatusTooManyRequests { | |
| 182 | + | return RequestFailureRateLimit | |
| 183 | + | } | |
| 184 | + | return RequestFailureServer | |
| 185 | + | } | |
| 186 | + | ||
| 187 | + | func requestFailureReasonForError(err error) RequestFailureReason { | |
| 188 | + | if err == nil { | |
| 189 | + | return RequestFailureUnknown | |
| 190 | + | } | |
| 191 | + | if errors.Is(err, context.DeadlineExceeded) || strings.Contains(strings.ToLower(err.Error()), "timeout") { | |
| 192 | + | return RequestFailureTimeout | |
| 193 | + | } | |
| 194 | + | if isTransientConnectError(err) { | |
| 195 | + | return RequestFailureNetwork | |
| 196 | + | } | |
| 197 | + | return RequestFailureUnknown | |
| 198 | + | } | |
| 199 | + | ||
| 160 | 200 | // synthesizeResponse rebuilds a closable response wrapping the captured | |
| 161 | 201 | // body bytes so the caller's existing non-200 handling path keeps | |
| 162 | 202 | // working without needing to know retries happened. | |
| @@ -102,7 +102,12 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||||
| 102 | resp, err := client.Do(req) | 102 | resp, err := client.Do(req) | ||
| 103 | if err != nil { | 103 | if err != nil { | ||
| 104 | lastErr = err | 104 | lastErr = err | ||
| 105 | - | if !isTransientConnectError(err) || ctx.Err() != nil { | 105 | + | if ctx.Err() != nil { |
| 106 | + | return nil, err | |||
| 107 | + | } | |||
| 108 | + | transient := isTransientConnectError(err) | |||
| 109 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, requestFailureReasonForError(err), !transient || attempt == streamRetryAttempts) | |||
| 110 | + | if !transient { | |||
| 106 | return nil, err | 111 | return nil, err | ||
| 107 | } | 112 | } | ||
| 108 | continue | 113 | continue | ||
| @@ -114,15 +119,18 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||||
| 114 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) | 119 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) | ||
| 115 | resp.Body.Close() | 120 | resp.Body.Close() | ||
| 116 | trimmed := strings.TrimSpace(string(body)) | 121 | trimmed := strings.TrimSpace(string(body)) | ||
| 122 | + | reason := requestFailureReasonForHTTP(resp.StatusCode, trimmed) | |||
| 117 | if resp.StatusCode == http.StatusTooManyRequests && isTerminalRateLimitBody(trimmed) { | 123 | if resp.StatusCode == http.StatusTooManyRequests && isTerminalRateLimitBody(trimmed) { | ||
| 118 | // Subscription/quota/billing limits are not transient | 124 | // Subscription/quota/billing limits are not transient | ||
| 119 | // throttles; retrying only burns time. Surface the | 125 | // throttles; retrying only burns time. Surface the | ||
| 120 | // response immediately so the caller formats the | 126 | // response immediately so the caller formats the | ||
| 121 | // normal "http 429: body" error. | 127 | // normal "http 429: body" error. | ||
| 128 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, RequestFailureQuota, true) | |||
| 122 | return synthesizeResponse(resp.StatusCode, body), nil | 129 | return synthesizeResponse(resp.StatusCode, body), nil | ||
| 123 | } | 130 | } | ||
| 124 | lastErr = &transientHTTPError{Status: resp.StatusCode, Body: trimmed} | 131 | lastErr = &transientHTTPError{Status: resp.StatusCode, Body: trimmed} | ||
| 125 | if attempt == streamRetryAttempts { | 132 | if attempt == streamRetryAttempts { | ||
| 133 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, true) | |||
| 126 | // Wrap as a real *http.Response shape the caller | 134 | // Wrap as a real *http.Response shape the caller | ||
| 127 | // expects so it formats the error identically to a | 135 | // expects so it formats the error identically to a | ||
| 128 | // non-retried failure. | 136 | // non-retried failure. | ||
| @@ -130,11 +138,13 @@ func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() ( | |||||
| 130 | } | 138 | } | ||
| 131 | if d, ok := retryAfterDelay(resp.Header); ok { | 139 | if d, ok := retryAfterDelay(resp.Header); ok { | ||
| 132 | if d > maxServerRetryDelay { | 140 | if d > maxServerRetryDelay { | ||
| 141 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, true) | |||
| 133 | return nil, fmt.Errorf("server requested %s retry delay (max %s): http %d: %s", | 142 | return nil, fmt.Errorf("server requested %s retry delay (max %s): http %d: %s", | ||
| 134 | d.Round(time.Second), maxServerRetryDelay, resp.StatusCode, trimmed) | 143 | d.Round(time.Second), maxServerRetryDelay, resp.StatusCode, trimmed) | ||
| 135 | } | 144 | } | ||
| 136 | serverDelay = d | 145 | serverDelay = d | ||
| 137 | } | 146 | } | ||
| 147 | + | reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, false) | |||
| 138 | continue | 148 | continue | ||
| 139 | } | 149 | } | ||
| 140 | return resp, nil | 150 | return resp, nil | ||
| @@ -157,6 +167,36 @@ type transientHTTPError struct { | |||||
| 157 | 167 | ||||
| 158 | func (e *transientHTTPError) Error() string { return "transient http error" } | 168 | func (e *transientHTTPError) Error() string { return "transient http error" } | ||
| 159 | 169 | ||||
| 170 | + | func reportRequestFailure(lifecycle RequestLifecycle, attempt, maxAttempts int, reason RequestFailureReason, terminal bool) { | |||
| 171 | + | if observer, ok := lifecycle.(RequestFailureLifecycle); ok { | |||
| 172 | + | observer.RequestFailed(attempt, maxAttempts, reason, terminal) | |||
| 173 | + | } | |||
| 174 | + | } | |||
| 175 | + | ||||
| 176 | + | func requestFailureReasonForHTTP(status int, body string) RequestFailureReason { | |||
| 177 | + | msg := strings.ToLower(body) | |||
| 178 | + | if status == 529 || strings.Contains(msg, "overload") { | |||
| 179 | + | return RequestFailureOverload | |||
| 180 | + | } | |||
| 181 | + | if status == http.StatusTooManyRequests { | |||
| 182 | + | return RequestFailureRateLimit | |||
| 183 | + | } | |||
| 184 | + | return RequestFailureServer | |||
| 185 | + | } | |||
| 186 | + | ||||
| 187 | + | func requestFailureReasonForError(err error) RequestFailureReason { | |||
| 188 | + | if err == nil { | |||
| 189 | + | return RequestFailureUnknown | |||
| 190 | + | } | |||
| 191 | + | if errors.Is(err, context.DeadlineExceeded) || strings.Contains(strings.ToLower(err.Error()), "timeout") { | |||
| 192 | + | return RequestFailureTimeout | |||
| 193 | + | } | |||
| 194 | + | if isTransientConnectError(err) { | |||
| 195 | + | return RequestFailureNetwork | |||
| 196 | + | } | |||
| 197 | + | return RequestFailureUnknown | |||
| 198 | + | } | |||
| 199 | + | ||||
| 160 | // synthesizeResponse rebuilds a closable response wrapping the captured | 200 | // synthesizeResponse rebuilds a closable response wrapping the captured | ||
| 161 | // body bytes so the caller's existing non-200 handling path keeps | 201 | // body bytes so the caller's existing non-200 handling path keeps | ||
| 162 | // working without needing to know retries happened. | 202 | // working without needing to know retries happened. | ||
M packages/provider/retry_test.go +40 -0
| @@ -227,9 +227,17 @@ type lifecycleRetry struct { | |||
| 227 | 227 | delay time.Duration | |
| 228 | 228 | } | |
| 229 | 229 | ||
| 230 | + | type lifecycleFailure struct { | |
| 231 | + | attempt int | |
| 232 | + | max int | |
| 233 | + | reason RequestFailureReason | |
| 234 | + | terminal bool | |
| 235 | + | } | |
| 236 | + | ||
| 230 | 237 | type recordingRequestLifecycle struct { | |
| 231 | 238 | attempts []lifecycleAttempt | |
| 232 | 239 | retries []lifecycleRetry | |
| 240 | + | failures []lifecycleFailure | |
| 233 | 241 | } | |
| 234 | 242 | ||
| 235 | 243 | func (r *recordingRequestLifecycle) RequestAttempt(attempt, maxAttempts int) { | |
| @@ -240,11 +248,16 @@ func (r *recordingRequestLifecycle) RetryScheduled(attempt, maxAttempts int, del | |||
| 240 | 248 | r.retries = append(r.retries, lifecycleRetry{attempt: attempt, max: maxAttempts, delay: delay}) | |
| 241 | 249 | } | |
| 242 | 250 | ||
| 251 | + | func (r *recordingRequestLifecycle) RequestFailed(attempt, maxAttempts int, reason RequestFailureReason, terminal bool) { | |
| 252 | + | r.failures = append(r.failures, lifecycleFailure{attempt: attempt, max: maxAttempts, reason: reason, terminal: terminal}) | |
| 253 | + | } | |
| 254 | + | ||
| 243 | 255 | func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | |
| 244 | 256 | var attempts atomic.Int32 | |
| 245 | 257 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | |
| 246 | 258 | if attempts.Add(1) == 1 { | |
| 247 | 259 | w.WriteHeader(http.StatusServiceUnavailable) | |
| 260 | + | _, _ = w.Write([]byte("servers overloaded: private provider detail")) | |
| 248 | 261 | return | |
| 249 | 262 | } | |
| 250 | 263 | w.WriteHeader(http.StatusOK) | |
| @@ -266,6 +279,33 @@ func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | |||
| 266 | 279 | if got := lifecycle.retries; len(got) != 1 || got[0].attempt != 2 || got[0].max != 3 || got[0].delay != 250*time.Millisecond { | |
| 267 | 280 | t.Fatalf("retries = %#v, want attempt 2 of 3 after 250ms", got) | |
| 268 | 281 | } | |
| 282 | + | if got := lifecycle.failures; len(got) != 1 || got[0] != (lifecycleFailure{attempt: 1, max: 3, reason: RequestFailureOverload}) { | |
| 283 | + | t.Fatalf("failures = %#v, want one non-terminal overload", got) | |
| 284 | + | } | |
| 285 | + | } | |
| 286 | + | ||
| 287 | + | func TestDoStreamWithRetryReportsTerminalQuota(t *testing.T) { | |
| 288 | + | const privateBody = `{"error":{"type":"insufficient_quota","message":"private provider detail"}}` | |
| 289 | + | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | |
| 290 | + | w.WriteHeader(http.StatusTooManyRequests) | |
| 291 | + | _, _ = w.Write([]byte(privateBody)) | |
| 292 | + | })) | |
| 293 | + | defer srv.Close() | |
| 294 | + | ||
| 295 | + | lifecycle := &recordingRequestLifecycle{} | |
| 296 | + | resp, err := doStreamWithRetry(context.Background(), srv.Client(), func() (*http.Request, error) { | |
| 297 | + | return http.NewRequestWithContext(context.Background(), http.MethodPost, srv.URL, nil) | |
| 298 | + | }, lifecycle) | |
| 299 | + | if err != nil { | |
| 300 | + | t.Fatalf("doStreamWithRetry: %v", err) | |
| 301 | + | } | |
| 302 | + | defer resp.Body.Close() | |
| 303 | + | if resp.StatusCode != http.StatusTooManyRequests { | |
| 304 | + | t.Fatalf("status = %d, want 429", resp.StatusCode) | |
| 305 | + | } | |
| 306 | + | if got := lifecycle.failures; len(got) != 1 || got[0] != (lifecycleFailure{attempt: 1, max: 3, reason: RequestFailureQuota, terminal: true}) { | |
| 307 | + | t.Fatalf("failures = %#v, want one terminal quota failure", got) | |
| 308 | + | } | |
| 269 | 309 | } | |
| 270 | 310 | ||
| 271 | 311 | func TestDoStreamWithRetryStopsLifecycleAfterCancellation(t *testing.T) { | |
| @@ -227,9 +227,17 @@ type lifecycleRetry struct { | |||||
| 227 | delay time.Duration | 227 | delay time.Duration | ||
| 228 | } | 228 | } | ||
| 229 | 229 | ||||
| 230 | + | type lifecycleFailure struct { | |||
| 231 | + | attempt int | |||
| 232 | + | max int | |||
| 233 | + | reason RequestFailureReason | |||
| 234 | + | terminal bool | |||
| 235 | + | } | |||
| 236 | + | ||||
| 230 | type recordingRequestLifecycle struct { | 237 | type recordingRequestLifecycle struct { | ||
| 231 | attempts []lifecycleAttempt | 238 | attempts []lifecycleAttempt | ||
| 232 | retries []lifecycleRetry | 239 | retries []lifecycleRetry | ||
| 240 | + | failures []lifecycleFailure | |||
| 233 | } | 241 | } | ||
| 234 | 242 | ||||
| 235 | func (r *recordingRequestLifecycle) RequestAttempt(attempt, maxAttempts int) { | 243 | func (r *recordingRequestLifecycle) RequestAttempt(attempt, maxAttempts int) { | ||
| @@ -240,11 +248,16 @@ func (r *recordingRequestLifecycle) RetryScheduled(attempt, maxAttempts int, del | |||||
| 240 | r.retries = append(r.retries, lifecycleRetry{attempt: attempt, max: maxAttempts, delay: delay}) | 248 | r.retries = append(r.retries, lifecycleRetry{attempt: attempt, max: maxAttempts, delay: delay}) | ||
| 241 | } | 249 | } | ||
| 242 | 250 | ||||
| 251 | + | func (r *recordingRequestLifecycle) RequestFailed(attempt, maxAttempts int, reason RequestFailureReason, terminal bool) { | |||
| 252 | + | r.failures = append(r.failures, lifecycleFailure{attempt: attempt, max: maxAttempts, reason: reason, terminal: terminal}) | |||
| 253 | + | } | |||
| 254 | + | ||||
| 243 | func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | 255 | func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | ||
| 244 | var attempts atomic.Int32 | 256 | var attempts atomic.Int32 | ||
| 245 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | 257 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | ||
| 246 | if attempts.Add(1) == 1 { | 258 | if attempts.Add(1) == 1 { | ||
| 247 | w.WriteHeader(http.StatusServiceUnavailable) | 259 | w.WriteHeader(http.StatusServiceUnavailable) | ||
| 260 | + | _, _ = w.Write([]byte("servers overloaded: private provider detail")) | |||
| 248 | return | 261 | return | ||
| 249 | } | 262 | } | ||
| 250 | w.WriteHeader(http.StatusOK) | 263 | w.WriteHeader(http.StatusOK) | ||
| @@ -266,6 +279,33 @@ func TestDoStreamWithRetryReportsLifecycle(t *testing.T) { | |||||
| 266 | if got := lifecycle.retries; len(got) != 1 || got[0].attempt != 2 || got[0].max != 3 || got[0].delay != 250*time.Millisecond { | 279 | if got := lifecycle.retries; len(got) != 1 || got[0].attempt != 2 || got[0].max != 3 || got[0].delay != 250*time.Millisecond { | ||
| 267 | t.Fatalf("retries = %#v, want attempt 2 of 3 after 250ms", got) | 280 | t.Fatalf("retries = %#v, want attempt 2 of 3 after 250ms", got) | ||
| 268 | } | 281 | } | ||
| 282 | + | if got := lifecycle.failures; len(got) != 1 || got[0] != (lifecycleFailure{attempt: 1, max: 3, reason: RequestFailureOverload}) { | |||
| 283 | + | t.Fatalf("failures = %#v, want one non-terminal overload", got) | |||
| 284 | + | } | |||
| 285 | + | } | |||
| 286 | + | ||||
| 287 | + | func TestDoStreamWithRetryReportsTerminalQuota(t *testing.T) { | |||
| 288 | + | const privateBody = `{"error":{"type":"insufficient_quota","message":"private provider detail"}}` | |||
| 289 | + | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | |||
| 290 | + | w.WriteHeader(http.StatusTooManyRequests) | |||
| 291 | + | _, _ = w.Write([]byte(privateBody)) | |||
| 292 | + | })) | |||
| 293 | + | defer srv.Close() | |||
| 294 | + | ||||
| 295 | + | lifecycle := &recordingRequestLifecycle{} | |||
| 296 | + | resp, err := doStreamWithRetry(context.Background(), srv.Client(), func() (*http.Request, error) { | |||
| 297 | + | return http.NewRequestWithContext(context.Background(), http.MethodPost, srv.URL, nil) | |||
| 298 | + | }, lifecycle) | |||
| 299 | + | if err != nil { | |||
| 300 | + | t.Fatalf("doStreamWithRetry: %v", err) | |||
| 301 | + | } | |||
| 302 | + | defer resp.Body.Close() | |||
| 303 | + | if resp.StatusCode != http.StatusTooManyRequests { | |||
| 304 | + | t.Fatalf("status = %d, want 429", resp.StatusCode) | |||
| 305 | + | } | |||
| 306 | + | if got := lifecycle.failures; len(got) != 1 || got[0] != (lifecycleFailure{attempt: 1, max: 3, reason: RequestFailureQuota, terminal: true}) { | |||
| 307 | + | t.Fatalf("failures = %#v, want one terminal quota failure", got) | |||
| 308 | + | } | |||
| 269 | } | 309 | } | ||
| 270 | 310 | ||||
| 271 | func TestDoStreamWithRetryStopsLifecycleAfterCancellation(t *testing.T) { | 311 | func TestDoStreamWithRetryStopsLifecycleAfterCancellation(t *testing.T) { | ||