Skip to content

fix(EVO-2167): retry the AI Processor call on transient failures #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gomessguii merged 2 commits into from
Jul 18, 2026
Merged

fix(EVO-2167): retry the AI Processor call on transient failures #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4d23878
fix(EVO-2167): retry the AI Processor call on transient failures
Jul 18, 2026
5df6957
fix(EVO-2167): harden retry — total-time cap + per-attempt timeout test
gomessguii Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion cmd/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ func main() {
debounce := debounceService.NewDebounceEngine(pipelineRepo)

// Step 6: AI adapter (URL comes from each event's outgoing_url)
aiAdapter := aiService.NewAIAdapter(cfg.AICallTimeoutSeconds)
aiAdapter := aiService.NewAIAdapter(cfg.AICallTimeoutSeconds, cfg.AICallMaxRetries, cfg.AICallRetryBaseMs)

// Step 7: dispatch engine (sends secret header on postback to CRM)
dispatch := dispatchService.NewDispatchEngine(cfg.BotRuntimeSecret)
Expand Down
15 changes: 15 additions & 0 deletions internal/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,11 @@ type Config struct {
RedisURL string
BotRuntimeSecret string
AICallTimeoutSeconds int
// EVO-2167: retry the AI Processor call on transient failures (5xx/429/network)
// so a momentary blip (deploy, restart, DB hiccup) does not leave the customer
// without a reply. AICallMaxRetries is retries AFTER the first attempt.
AICallMaxRetries int
AICallRetryBaseMs int
}

func Load() (*Config, error) {
Expand All @@ -27,12 +32,22 @@ func Load() (*Config, error) {
if err != nil {
return nil, err
}
aiCallMaxRetries, err := getEnvIntOrDefault("AI_CALL_MAX_RETRIES", 2)
if err != nil {
return nil, err
}
aiCallRetryBaseMs, err := getEnvIntOrDefault("AI_CALL_RETRY_BASE_MS", 200)
if err != nil {
return nil, err
}

return &Config{
ListenAddr: listenAddr,
RedisURL: redisURL,
BotRuntimeSecret: botRuntimeSecret,
AICallTimeoutSeconds: aiCallTimeout,
AICallMaxRetries: aiCallMaxRetries,
AICallRetryBaseMs: aiCallRetryBaseMs,
}, nil
}

Expand Down
152 changes: 136 additions & 16 deletions pkg/ai/service/ai_adapter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"fmt"
"io"
"log/slog"
"math/rand"
"net/http"
"time"

Expand All @@ -18,33 +19,49 @@ import (
// maxResponseBytes caps the AI Processor response body to prevent OOM on oversized payloads.
const maxResponseBytes = 1 << 20 // 1 MiB

// maxBackoff caps the exponential backoff between retries so a large
// AI_CALL_RETRY_BASE_MS or retry count cannot balloon the wait.
const maxBackoff = 5 * time.Second

// AIAdapter calls the AI Processor via A2A protocol (JSON-RPC 2.0).
// Swap the backend by providing a different implementation at main.go wiring.
type AIAdapter interface {
Call(ctx context.Context, req *model.A2ARequest) (*model.NormalizedResponse, error)
}

type aiAdapter struct {
timeoutSecs int
client *http.Client
timeoutSecs int
maxRetries int
retryBaseDelay time.Duration
client *http.Client
}

// NewAIAdapter constructs the adapter. Returns interface (GEAR R03).
// The AI Processor URL comes from each event's outgoing_url field.
func NewAIAdapter(timeoutSecs int) AIAdapter {
//
// EVO-2167: maxRetries is the number of retries AFTER the first attempt; the call
// is retried on transient failures (5xx/429/network errors) with exponential
// backoff + jitter, so a momentary processor blip does not drop the customer's
// reply. Permanent failures (4xx), timeouts and pipeline cancellation are not
// retried. retryBaseMs is the base backoff in milliseconds.
func NewAIAdapter(timeoutSecs, maxRetries, retryBaseMs int) AIAdapter {
if maxRetries < 0 {
maxRetries = 0
}
if retryBaseMs <= 0 {
retryBaseMs = 200
}
return &aiAdapter{
timeoutSecs: timeoutSecs,
client: &http.Client{},
timeoutSecs: timeoutSecs,
maxRetries: maxRetries,
retryBaseDelay: time.Duration(retryBaseMs) * time.Millisecond,
client: &http.Client{},
}
}

func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.NormalizedResponse, error) {
start := time.Now()

// Wrap with timeout — inner timeout, outer ctx for pipeline cancellation.
timeoutCtx, cancel := context.WithTimeout(ctx, time.Duration(a.timeoutSecs)*time.Second)
defer cancel()

// Use the full outgoing_url provided by the CRM (already contains the agent ID)
url := req.OutgoingURL

Expand Down Expand Up @@ -88,37 +105,112 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor
return nil, fmt.Errorf("pipeline.ai.marshal: %w", err)
}

// EVO-2167: retry the send on transient failures (5xx/429/network) with
// exponential backoff + jitter. The body is built once and reused per attempt.
//
// Idempotency caveat (the card flagged this to validate): retrying 502/504/network
// can re-run an attempt the AI Processor already processed but whose response was
// lost in transit → a duplicate agent turn (ADK history, tool calls, LLM cost).
// The customer still gets exactly one reply: dispatch runs once, only after a
// successful attempt. The request body is byte-identical on every retry (built
// once above), so suppressing the duplicate server-side turn is the AI Processor's
// responsibility — tracked in EVO-2166. Keep this path free of extra side effects.
attempts := a.maxRetries + 1
perAttempt := time.Duration(a.timeoutSecs) * time.Second

// AC "teto de tempo total": bound the whole sequence with an overall backstop so a
// growing backoff or attempt count can never run unbounded. The ceiling is the
// natural worst case (attempts × per-attempt timeout + summed max backoff) plus one
// per-attempt of slack, so a genuine per-attempt timeout still surfaces as a timeout
// instead of being swallowed by this backstop. Each attempt keeps its own timeout.
overallCtx := ctx
if perAttempt > 0 {
ceiling := time.Duration(attempts+1)*perAttempt + time.Duration(a.maxRetries)*maxBackoff
var cancelAll context.CancelFunc
overallCtx, cancelAll = context.WithTimeout(ctx, ceiling)
defer cancelAll()
}

var lastErr error
for attempt := 0; attempt < attempts; attempt++ {
if attempt > 0 {
delay := a.backoffDelay(attempt)
slog.Warn("pipeline.ai.retry",
"contact_id", req.ContactID,
"conversation_id", req.ConversationID,
"attempt", attempt,
"max_retries", a.maxRetries,
"delay_ms", delay.Milliseconds(),
)
select {
case <-overallCtx.Done():
return nil, brtErrors.ErrPipelineCancelled
case <-time.After(delay):
}
}

resp, retryable, err := a.doOnce(overallCtx, url, body, req, start)
if err == nil {
return resp, nil
}
lastErr = err
if !retryable {
return nil, err
}
}
return nil, lastErr
}

// doOnce performs a single AI Processor call with a per-attempt timeout. The bool
// return reports whether the error is transient (worth retrying): network errors
// and 5xx/429 statuses are retryable; timeouts, pipeline cancellation, 4xx and
// decode errors are not.
func (a *aiAdapter) doOnce(
ctx context.Context,
url string,
body []byte,
req *model.A2ARequest,
start time.Time,
) (*model.NormalizedResponse, bool, error) {
// Per-attempt timeout — inner timeout, outer ctx for pipeline cancellation.
timeoutCtx, cancel := context.WithTimeout(ctx, time.Duration(a.timeoutSecs)*time.Second)
defer cancel()

httpReq, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost, url, bytes.NewReader(body))
if err != nil {
return nil, fmt.Errorf("pipeline.ai.new_request: %w", err)
return nil, false, fmt.Errorf("pipeline.ai.new_request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("X-API-Key", req.ApiKey)

resp, err := a.client.Do(httpReq)
if err != nil {
if ctx.Err() != nil {
return nil, brtErrors.ErrPipelineCancelled
return nil, false, brtErrors.ErrPipelineCancelled
}
if errors.Is(timeoutCtx.Err(), context.DeadlineExceeded) {
slog.Warn("pipeline.ai.http.timeout",
"contact_id", req.ContactID,
"conversation_id", req.ConversationID,
"timeout_secs", a.timeoutSecs,
)
return nil, brtErrors.ErrAITimeout
return nil, false, brtErrors.ErrAITimeout
}
return nil, fmt.Errorf("pipeline.ai.http: %w", err)
// Transient network error (connection refused/reset, e.g. during a deploy).
return nil, true, fmt.Errorf("pipeline.ai.http: %w", err)
}
defer resp.Body.Close()

if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("pipeline.ai.status: unexpected %d from AI Processor", resp.StatusCode)
retryable := isRetryableStatus(resp.StatusCode)
// Drain a bounded amount so the connection can be reused.
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
return nil, retryable, fmt.Errorf("pipeline.ai.status: unexpected %d from AI Processor", resp.StatusCode)
}

var a2aResp model.A2AResponse
if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBytes)).Decode(&a2aResp); err != nil {
return nil, fmt.Errorf("pipeline.ai.decode: %w", err)
return nil, false, fmt.Errorf("pipeline.ai.decode: %w", err)
}

content := extractResponseText(&a2aResp)
Expand All @@ -129,7 +221,35 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor
"duration_ms", time.Since(start).Milliseconds(),
)

return &model.NormalizedResponse{Content: content}, nil
return &model.NormalizedResponse{Content: content}, false, nil
}

// isRetryableStatus reports whether an HTTP status from the AI Processor is a
// transient failure worth retrying. 4xx (bad key/request) are permanent — after
// EVO-2166 an infra error surfaces as 503, not 401, so retrying 5xx covers the
// transient auth/infra case without retrying genuine 401s.
func isRetryableStatus(code int) bool {
switch code {
case http.StatusTooManyRequests, // 429
http.StatusInternalServerError, // 500
http.StatusBadGateway, // 502
http.StatusServiceUnavailable, // 503
http.StatusGatewayTimeout: // 504
return true
}
return false
}

// backoffDelay returns the exponential backoff for the given attempt (1-based)
// with up to 50% jitter, capped at maxBackoff.
func (a *aiAdapter) backoffDelay(attempt int) time.Duration {
d := a.retryBaseDelay << (attempt - 1) // base * 2^(attempt-1)
if d <= 0 || d > maxBackoff {
d = maxBackoff
}
// Keep at least half the delay, add up to half more as jitter.
jitter := time.Duration(rand.Int63n(int64(d/2) + 1))
Comment on lines +250 to +251

@sourcery-ai sourcery-ai Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Consider seeding or isolating the RNG source for jittered backoff

Using the package-level math/rand default source means the jitter sequence is deterministic (seeded to 1) unless something else seeds it, so all processes will see the same jitter pattern after startup. For more effective jitter, seed the RNG once at startup or use a dedicated rand.Source/rand.Rand for backoff (e.g., on aiAdapter).

Suggested implementation:

import (
	"math/rand"
	// existing imports...
)

// backoffDelay returns the exponential backoff for the given attempt (1-based)
// with up to 50% jitter, capped at maxBackoff.
func (a *aiAdapter) backoffDelay(attempt int) time.Duration {
	d := a.retryBaseDelay << (attempt - 1) // base * 2^(attempt-1)
	if d <= 0 || d > maxBackoff {
		d = maxBackoff
	}

	// Lazily initialize a jitter RNG specific to this adapter instance to
	// avoid using the global math/rand source and to get per-process jitter.
	if a.rng == nil {
		a.rng = rand.New(rand.NewSource(time.Now().UnixNano()))
	}

	// Keep at least half the delay, add up to half more as jitter.
	maxJitter := d / 2
	jitter := time.Duration(a.rng.Int63n(int64(maxJitter) + 1))

	return d/2 + jitter
type aiAdapter struct {
	timeoutSecs    int
	maxRetries     int
	retryBaseDelay time.Duration
	client         *http.Client

	// rng is a per-adapter RNG used for jittered backoff.
	rng *rand.Rand
}
  1. The import edit above assumes a simple import block; you may need to merge the math/rand import into the existing import (...) block and remove the placeholder comment // existing imports....
  2. If you already have a constructor or initializer for aiAdapter, you could prefer initializing rng there instead of lazily inside backoffDelay, e.g. rng: rand.New(rand.NewSource(time.Now().UnixNano())), to avoid the nil check on every call.
  3. If your project uses a central randomness source (for testability), you may want to inject a rand.Source or *rand.Rand into aiAdapter instead of creating it with time.Now().UnixNano() directly.

return d/2 + jitter
}

// extractResponseText extracts the text content from the A2A JSON-RPC response.
Expand Down
Loading
Loading