git.iko.soy/crush
on main help sign in
cleanup retry
0d8d324Kujtim Hoxha
internal/llm/provider/anthropic.go+44−73
compare with main
unifiedsplit
@@ -135,40 +135,9 @@ func (a *anthropicProvider) StreamResponse(ctx context.Context, messages []messa
135 attempts := 0
135 attempts := 0
136
136
137 for {
137 for {
138− // If this isn't the first attempt, we're retrying
139− if attempts > 0 {
140− if attempts > maxRetries {
141− eventChan <- ProviderEvent{
142− Type: EventError,
143− Error: errors.New("maximum retry attempts reached for rate limit (429)"),
144− }
145− return
146− }
147−
148− // Inform user we're retrying with attempt number
149− eventChan <- ProviderEvent{
150− Type: EventWarning,
151− Info: fmt.Sprintf("[Retrying due to rate limit... attempt %d of %d]", attempts, maxRetries),
152− }
153−
154− // Calculate backoff with exponential backoff and jitter
155− backoffMs := 2000 * (1 << (attempts - 1)) // 2s, 4s, 8s, 16s, 32s
156− jitterMs := int(float64(backoffMs) * 0.2)
157− totalBackoffMs := backoffMs + jitterMs
158−
159− // Sleep with backoff, respecting context cancellation
160− select {
161− case <-ctx.Done():
162− eventChan <- ProviderEvent{Type: EventError, Error: ctx.Err()}
163− return
164− case <-time.After(time.Duration(totalBackoffMs) * time.Millisecond):
165− // Continue with retry
166− }
167− }
168
138
169 attempts++
139 attempts++
170
140
171− // Create new streaming request
172 stream := a.client.Messages.NewStreaming(
141 stream := a.client.Messages.NewStreaming(
173 ctx,
142 ctx,
174 anthropic.MessageNewParams{
143 anthropic.MessageNewParams{
@@ -189,11 +158,8 @@ func (a *anthropicProvider) StreamResponse(ctx context.Context, messages []messa
189 },
158 },
190 )
159 )
191
160
192− // Process stream events
193 accumulatedMessage := anthropic.Message{}
161 accumulatedMessage := anthropic.Message{}
194− streamSuccess := false
195
162
196− // Process the stream until completion or error
197 for stream.Next() {
163 for stream.Next() {
198 event := stream.Current()
164 event := stream.Current()
199 err := accumulatedMessage.Accumulate(event)
165 err := accumulatedMessage.Accumulate(event)
@@ -223,7 +189,6 @@ func (a *anthropicProvider) StreamResponse(ctx context.Context, messages []messa
223 eventChan <- ProviderEvent{Type: EventContentStop}
189 eventChan <- ProviderEvent{Type: EventContentStop}
224
190
225 case anthropic.MessageStopEvent:
191 case anthropic.MessageStopEvent:
226− streamSuccess = true
227 content := ""
192 content := ""
228 for _, block := range accumulatedMessage.Content {
193 for _, block := range accumulatedMessage.Content {
229 if text, ok := block.AsAny().(anthropic.TextBlock); ok {
194 if text, ok := block.AsAny().(anthropic.TextBlock); ok {
@@ -246,51 +211,59 @@ func (a *anthropicProvider) StreamResponse(ctx context.Context, messages []messa
246 }
211 }
247 }
212 }
248
213
249− // If the stream completed successfully, we're done
214+ err := stream.Err()
250− if streamSuccess {
215+ if err == nil {
251 return
216 return
252 }
217 }
253
218
254− // Check for stream errors
255− err := stream.Err()
256− if err != nil {
257− var apierr *anthropic.Error
219+ var apierr *anthropic.Error
258− if errors.As(err, &apierr) {
220+ if !errors.As(err, &apierr) {
259− if apierr.StatusCode == 429 || apierr.StatusCode == 529 {
260− // Check for Retry-After header
261− if retryAfterValues := apierr.Response.Header.Values("Retry-After"); len(retryAfterValues) > 0 {
262− // Parse the retry after value (seconds)
263− var retryAfterSec int
264− if _, err := fmt.Sscanf(retryAfterValues[0], "%d", &retryAfterSec); err == nil {
265− retryMs := retryAfterSec * 1000
266−
267− // Inform user of retry with specific wait time
268− eventChan <- ProviderEvent{
269− Type: EventWarning,
270− Info: fmt.Sprintf("[Rate limited: waiting %d seconds as specified by API]", retryAfterSec),
271− }
272−
273− // Sleep respecting context cancellation
274− select {
275− case <-ctx.Done():
276− eventChan <- ProviderEvent{Type: EventError, Error: ctx.Err()}
221+ eventChan <- ProviderEvent{Type: EventError, Error: err}
277− return
222+ return
278− case <-time.After(time.Duration(retryMs) * time.Millisecond):
279− // Continue with retry after specified delay
280− continue
281− }
282− }
283− }
223+ }
284
224
285− // Fall back to exponential backoff if Retry-After parsing failed
225+ if apierr.StatusCode != 429 && apierr.StatusCode != 529 {
286− continue
226+ eventChan <- ProviderEvent{Type: EventError, Error: err}
227+ return
228+ }
229+
230+ if attempts > maxRetries {
231+ eventChan <- ProviderEvent{
232+ Type: EventError,
233+ Error: errors.New("maximum retry attempts reached for rate limit (429)"),
234+ }
235+ return
236+ }
237+
238+ retryMs := 0
239+ retryAfterValues := apierr.Response.Header.Values("Retry-After")
240+ if len(retryAfterValues) > 0 {
241+ var retryAfterSec int
242+ if _, err := fmt.Sscanf(retryAfterValues[0], "%d", &retryAfterSec); err == nil {
243+ retryMs = retryAfterSec * 1000
244+ eventChan <- ProviderEvent{
245+ Type: EventWarning,
246+ Info: fmt.Sprintf("[Rate limited: waiting %d seconds as specified by API]", retryAfterSec),
287 }
247 }
288 }
248 }
249+ } else {
250+ eventChan <- ProviderEvent{
251+ Type: EventWarning,
252+ Info: fmt.Sprintf("[Retrying due to rate limit... attempt %d of %d]", attempts, maxRetries),
253+ }
289
254
290− // For non-rate limit errors, report and exit
255+ backoffMs := 2000 * (1 << (attempts - 1))
256+ jitterMs := int(float64(backoffMs) * 0.2)
257+ retryMs = backoffMs + jitterMs
258+ }
259+ select {
260+ case <-ctx.Done():
291− eventChan <- ProviderEvent{Type: EventError, Error: err}
261+ eventChan <- ProviderEvent{Type: EventError, Error: ctx.Err()}
292 return
262 return
263+ case <-time.After(time.Duration(retryMs) * time.Millisecond):
264+ continue
293 }
265 }
266+
294 }
267 }
295 }()
268 }()
296
269
@@ -388,7 +361,6 @@ func (a *anthropicProvider) convertToAnthropicMessages(messages []message.Messag
388 blocks = append(blocks, anthropic.ContentBlockParamOfRequestToolUseBlock(toolCall.ID, inputMap, toolCall.Name))
361 blocks = append(blocks, anthropic.ContentBlockParamOfRequestToolUseBlock(toolCall.ID, inputMap, toolCall.Name))
389 }
362 }
390
363
391− // Skip empty assistant messages completely
392 if len(blocks) > 0 {
364 if len(blocks) > 0 {
393 anthropicMessages = append(anthropicMessages, anthropic.NewAssistantMessage(blocks...))
365 anthropicMessages = append(anthropicMessages, anthropic.NewAssistantMessage(blocks...))
394 }
366 }
@@ -404,4 +376,3 @@ func (a *anthropicProvider) convertToAnthropicMessages(messages []message.Messag
404
376
405 return anthropicMessages
377 return anthropicMessages
406}
378}
407−
internal/llm/tools/fetch.go+1−5
@@ -121,11 +121,7 @@ func (t *fetchTool) Run(ctx context.Context, call ToolCall) (ToolResponse, error
121 ToolName: FetchToolName,
121 ToolName: FetchToolName,
122 Action: "fetch",
122 Action: "fetch",
123 Description: fmt.Sprintf("Fetch content from URL: %s", params.URL),
123 Description: fmt.Sprintf("Fetch content from URL: %s", params.URL),
124− Params: FetchPermissionsParams{
124+ Params: FetchPermissionsParams(params),
125− URL: params.URL,
126− Format: params.Format,
127− Timeout: params.Timeout,
128− },
129 },
125 },
130 )
126 )
131
127