diff --git a/cmd/server/main.go b/cmd/server/main.go index 5452ad5..85f80de 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -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) diff --git a/internal/config/config.go b/internal/config/config.go index 6c2a2d5..76592f4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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) { @@ -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 } diff --git a/pkg/ai/service/ai_adapter.go b/pkg/ai/service/ai_adapter.go index 30a94b1..6ef82dc 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "log/slog" + "math/rand" "net/http" "time" @@ -18,6 +19,10 @@ 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 { @@ -25,26 +30,38 @@ type AIAdapter interface { } 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 @@ -88,9 +105,80 @@ 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) @@ -98,7 +186,7 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor 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", @@ -106,19 +194,23 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor "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) @@ -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)) + return d/2 + jitter } // extractResponseText extracts the text content from the A2A JSON-RPC response. diff --git a/pkg/ai/service/ai_adapter_retry_test.go b/pkg/ai/service/ai_adapter_retry_test.go new file mode 100644 index 0000000..3b0ca28 --- /dev/null +++ b/pkg/ai/service/ai_adapter_retry_test.go @@ -0,0 +1,160 @@ +package service_test + +// EVO-2167: the AI Processor call must retry transient failures (5xx/429/network) +// so a momentary blip does not leave the customer without a reply, while NOT +// retrying permanent failures (4xx). + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" + aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" + aiService "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service" +) + +func writeOK(w http.ResponseWriter) { + _ = json.NewEncoder(w).Encode(aiModel.A2AResponse{ + Result: &aiModel.A2AResult{ + Artifacts: []aiModel.A2AArtifact{ + {Parts: []aiModel.A2APart{{Type: "text", Text: "ok"}}}, + }, + }, + }) +} + +func retryReq(url string) *aiModel.A2ARequest { + return &aiModel.A2ARequest{ + OutgoingURL: url + "/api/v1/a2a/agent-123", + Message: "hi", + ContactID: 42, + ConversationID: 7, + ApiKey: "k", + } +} + +func TestCall_RetriesOn503ThenSucceeds(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) == 1 { + w.WriteHeader(http.StatusServiceUnavailable) // 503 -> transient + return + } + writeOK(w) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 2, 1) // 2 retries, 1ms base + resp, err := adapter.Call(context.Background(), retryReq(server.URL)) + if err != nil { + t.Fatalf("expected success after retry, got error: %v", err) + } + if resp.Content != "ok" { + t.Errorf("content = %q, want ok", resp.Content) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("server calls = %d, want 2 (503 then 200)", got) + } +} + +func TestCall_ExhaustsRetriesOnPersistent500(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 2, 1) + if _, err := adapter.Call(context.Background(), retryReq(server.URL)); err == nil { + t.Fatal("expected error after exhausting retries, got nil") + } + if got := atomic.LoadInt32(&calls); got != 3 { + t.Errorf("server calls = %d, want 3 (1 + 2 retries)", got) + } +} + +func TestCall_DoesNotRetryOn400(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusBadRequest) // 400 -> permanent + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 3, 1) + if _, err := adapter.Call(context.Background(), retryReq(server.URL)); err == nil { + t.Fatal("expected error for 400, got nil") + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("server calls = %d, want 1 (400 is permanent, no retry)", got) + } +} + +func TestCall_RetriesOnNetworkErrorThenSucceeds(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) == 1 { + panic(http.ErrAbortHandler) // abruptly close the connection -> client network error + } + writeOK(w) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 2, 1) + resp, err := adapter.Call(context.Background(), retryReq(server.URL)) + if err != nil { + t.Fatalf("expected success after network retry, got error: %v", err) + } + if resp.Content != "ok" { + t.Errorf("content = %q, want ok", resp.Content) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("server calls = %d, want 2", got) + } +} + +func TestCall_NoRetryWhenMaxRetriesZero(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) // retries disabled -> single attempt + if _, err := adapter.Call(context.Background(), retryReq(server.URL)); err == nil { + t.Fatal("expected error, got nil") + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("server calls = %d, want 1 (no retry)", got) + } +} + +// AC #5 ("timeout por tentativa"): a per-attempt timeout must surface as ErrAITimeout +// and must NOT be retried, even with retries enabled. Regression guard for the +// per-attempt timeout scoping in doOnce. +func TestCall_TimeoutIsNotRetried(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + time.Sleep(1200 * time.Millisecond) // exceed the 1s per-attempt timeout below + writeOK(w) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(1, 3, 1) // 1s per-attempt timeout, retries enabled + _, err := adapter.Call(context.Background(), retryReq(server.URL)) + if !errors.Is(err, brtErrors.ErrAITimeout) { + t.Fatalf("expected ErrAITimeout, got %v", err) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("server calls = %d, want 1 (per-attempt timeout must NOT be retried)", got) + } +} diff --git a/pkg/ai/service/ai_adapter_test.go b/pkg/ai/service/ai_adapter_test.go index fe031b7..2b12e42 100644 --- a/pkg/ai/service/ai_adapter_test.go +++ b/pkg/ai/service/ai_adapter_test.go @@ -85,7 +85,7 @@ func TestCall_Success(t *testing.T) { })) defer server.Close() - adapter := aiService.NewAIAdapter(30) + adapter := aiService.NewAIAdapter(30, 0, 0) resp, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ OutgoingURL: server.URL + "/api/v1/a2a/agent-123", Message: "hello world", @@ -128,7 +128,7 @@ func TestCall_ContextID_FallsBackToNumericID(t *testing.T) { })) defer server.Close() - adapter := aiService.NewAIAdapter(30) + adapter := aiService.NewAIAdapter(30, 0, 0) _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ OutgoingURL: server.URL, Message: "test", @@ -156,7 +156,7 @@ func TestCall_Success_MessageFormat(t *testing.T) { })) defer server.Close() - adapter := aiService.NewAIAdapter(30) + adapter := aiService.NewAIAdapter(30, 0, 0) resp, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ OutgoingURL: server.URL, Message: "test", @@ -183,7 +183,7 @@ func TestCall_ContextCancellation_ReturnsPipelineCancelled(t *testing.T) { server.Close() }) - adapter := aiService.NewAIAdapter(30) + adapter := aiService.NewAIAdapter(30, 0, 0) ctx, cancel := context.WithCancel(context.Background()) go func() { @@ -210,7 +210,7 @@ func TestCall_Timeout_ReturnsAITimeout(t *testing.T) { server.Close() }) - adapter := aiService.NewAIAdapter(1) // 1 s timeout + adapter := aiService.NewAIAdapter(1, 0, 0) // 1 s timeout _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{OutgoingURL: server.URL, Message: "test", ApiKey: "key"}) if !errors.Is(err, brtErrors.ErrAITimeout) { t.Errorf("expected ErrAITimeout, got %v", err) @@ -223,7 +223,7 @@ func TestCall_NonOKStatus_ReturnsError(t *testing.T) { })) defer server.Close() - adapter := aiService.NewAIAdapter(30) + adapter := aiService.NewAIAdapter(30, 0, 0) _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{OutgoingURL: server.URL, Message: "test", ApiKey: "key"}) if err == nil { t.Fatal("expected error for non-200 response, got nil")