diff --git a/.env.example b/.env.example index ea13f0d..8504a71 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,24 @@ LISTEN_ADDR=:8090 REDIS_URL=redis://localhost:6379 + +# Required. Shared with the CRM, which sends it as X-Bot-Runtime-Secret on /events. +# An empty value would authenticate any caller that omits the header. BOT_RUNTIME_SECRET= -AI_CALL_TIMEOUT_SECONDS=30 + +# CRM-236: a tool-calling turn makes two model calls and the provider's tail adds +# up to ~20s each. An explicit value here overrides the code default. +AI_CALL_TIMEOUT_SECONDS=90 + +# Message sent to the customer when the AI cannot answer (timeout or provider +# outage). Empty string disables it and restores the old silence. +# AI_FAILURE_NOTICE=We are having a temporary issue and could not answer right now. We will get back to you shortly. + +# Required for incoming media. Hosts allowed to serve it, comma-separated, no +# scheme or port (e.g. "crm.example.com,minio.internal"). +# +# Set it to the host of the CRM's BACKEND_URL — that is what signs the attachment +# URLs — plus the storage host when ActiveStorage runs in redirect mode, or the CDN +# host. Leave it empty and no media reaches the agent: an attachment on an unlisted +# host is skipped and logged as pipeline.ai.attachment.blocked_url, and only the +# text reply goes out. +MEDIA_HOST_ALLOWLIST= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8c74f6b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,41 @@ +name: CI + +# Nothing ran the Go suite on a PR before this: test/e2e sat non-compiling from +# EVO-558 to EVO-2180 without a single red check. Redis is a service container +# because the repository tests need a real one. + +on: + pull_request: + push: + branches: [develop, main] + +jobs: + test: + runs-on: ubuntu-latest + services: + redis: + image: redis:7-alpine + ports: ['6379:6379'] + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + env: + REDIS_TEST_URL: redis://localhost:6379 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true + + - name: Build + run: go build ./... + + - name: Vet + run: go vet ./... + + - name: Test + run: go test ./... 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..fa5be62 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) { @@ -22,8 +27,23 @@ func Load() (*Config, error) { if err != nil { return nil, err } - botRuntimeSecret := os.Getenv("BOT_RUNTIME_SECRET") - aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 30) + // Required: SecretMiddleware compares the header against this, so an empty value + // authenticates every caller that omits it. + botRuntimeSecret, err := mustGetEnv("BOT_RUNTIME_SECRET") + if err != nil { + return nil, err + } + // CRM-236: 90s, not 30. A tool-calling turn makes two model calls and the + // provider's tail alone measured 20.4s on a trivial prompt. + aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 90) + 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 } @@ -33,6 +53,8 @@ func Load() (*Config, error) { RedisURL: redisURL, BotRuntimeSecret: botRuntimeSecret, AICallTimeoutSeconds: aiCallTimeout, + AICallMaxRetries: aiCallMaxRetries, + AICallRetryBaseMs: aiCallRetryBaseMs, }, nil } diff --git a/k8s/configmap.yaml b/k8s/configmap.yaml index 752820d..d3bbbda 100644 --- a/k8s/configmap.yaml +++ b/k8s/configmap.yaml @@ -5,4 +5,9 @@ metadata: data: LISTEN_ADDR: ":8080" AI_PROCESSOR_URL: "http://ai-processor:8000" - AI_CALL_TIMEOUT_SECONDS: "30" + # CRM-236: an explicit value overrides the code default, and this ConfigMap is + # what runs in staging/production. + AI_CALL_TIMEOUT_SECONDS: "90" + # Hosts allowed to serve incoming media, comma-separated (no scheme/port). + # Must include the host of the CRM's BACKEND_URL, or no media reaches the agent. + MEDIA_HOST_ALLOWLIST: "" diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index df51e7f..4695513 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -44,6 +44,11 @@ spec: configMapKeyRef: name: evo-bot-runtime-config key: AI_CALL_TIMEOUT_SECONDS + - name: MEDIA_HOST_ALLOWLIST + valueFrom: + configMapKeyRef: + name: evo-bot-runtime-config + key: MEDIA_HOST_ALLOWLIST # Secrets - name: REDIS_URL valueFrom: diff --git a/pkg/ai/model/a2a.go b/pkg/ai/model/a2a.go index c668978..5e6de87 100644 --- a/pkg/ai/model/a2a.go +++ b/pkg/ai/model/a2a.go @@ -9,14 +9,23 @@ type A2ARequest struct { ApiKey string // used for X-API-Key header (per-event auth) Message string // aggregated buffer content (FR-15) Metadata map[string]any // CRM metadata passed through to processor (tools context) + Attachments []Attachment // EVO-2180: incoming media to forward as A2A file parts +} + +// Attachment is an incoming media item (image/audio/…) the adapter downloads and +// forwards to the AI Processor as a base64 A2A file part. +type Attachment struct { + URL string // downloadable URL (Rails proxy on BACKEND_URL, reachable server-side) + ContentType string // e.g. "image/jpeg" + FileType string // CRM file_type: image/audio/video/file } // jsonRPCRequest is the JSON-RPC 2.0 envelope sent to AI Processor. type JSONRPCRequest struct { - JSONRPC string `json:"jsonrpc"` - ID string `json:"id"` - Method string `json:"method"` - Params JSONRPCParams `json:"params"` + JSONRPC string `json:"jsonrpc"` + ID string `json:"id"` + Method string `json:"method"` + Params JSONRPCParams `json:"params"` } type JSONRPCParams struct { @@ -27,13 +36,22 @@ type JSONRPCParams struct { } type JSONRPCMessage struct { - Role string `json:"role"` - Parts []JSONRPCPart `json:"parts"` + Role string `json:"role"` + Parts []JSONRPCPart `json:"parts"` } type JSONRPCPart struct { - Type string `json:"type"` - Text string `json:"text,omitempty"` + Type string `json:"type"` + Text string `json:"text,omitempty"` + File *JSONRPCFile `json:"file,omitempty"` +} + +// JSONRPCFile is a base64 file part. Field names/tags match what the AI Processor +// reads (extract_files_from_message: name / mimeType / bytes). +type JSONRPCFile struct { + Name string `json:"name,omitempty"` + MimeType string `json:"mimeType"` + Bytes string `json:"bytes"` // base64-encoded content } // A2AResponse is the JSON-RPC 2.0 response from AI Processor. diff --git a/pkg/ai/service/ai_adapter.go b/pkg/ai/service/ai_adapter.go index 30a94b1..a144be5 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -3,12 +3,19 @@ package service import ( "bytes" "context" + "encoding/base64" "encoding/json" "errors" "fmt" "io" "log/slog" + "math/rand" + "mime" "net/http" + neturl "net/url" + "os" + "path" + "strings" "time" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" @@ -18,6 +25,41 @@ import ( // maxResponseBytes caps the AI Processor response body to prevent OOM on oversized payloads. const maxResponseBytes = 1 << 20 // 1 MiB +// maxAttachmentBytes caps a single downloaded incoming attachment (EVO-2180). +// Images routinely exceed maxResponseBytes (1 MiB); base64 inflates ~33%, so keep +// this conservative relative to the processor's request-body limit. +const maxAttachmentBytes = 15 << 20 // 15 MiB + +// maxAttachmentsTotalBytes caps the SUM of every attachment forwarded in one call. +// The debounce window aggregates the media of all messages in it, so a per-file cap +// alone lets a photo burst build a body of len(attachments) x maxAttachmentBytes. +// Base64 inflates that ~33% and the gateway rejects the POST (nginx +// client_max_body_size) — and a 413 is not retryable, so the customer would lose the +// text reply as well. Over budget, the remaining attachments are dropped and the +// call proceeds with what fits. +const maxAttachmentsTotalBytes = 20 << 20 // 20 MiB (~27 MiB once base64-encoded) + +// Attachment downloads run before the AI call and outside its retry ceiling, so they +// need a bound of their own: reusing the AI timeout (AI_CALL_TIMEOUT_SECONDS, +// default 30s) meant an unreachable media host stalled every turn for +// timeout x len(attachments) before the processor was even called. +// maxAttachmentDownload bounds one download; the whole set shares +// attachmentsTotalTimeFactor times that. +const ( + maxAttachmentDownload = 10 * time.Second + attachmentsTotalTimeFactor = 3 +) + +// mediaHostAllowlistEnv names the hosts authorized to serve incoming media, +// comma-separated. It is the *only* source: an allowlist read out of the event +// would be chosen by whoever sent the event, which is exactly who it must +// constrain. Unset means no media is fetched. +const mediaHostAllowlistEnv = "MEDIA_HOST_ALLOWLIST" + +// 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 +67,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 @@ -66,6 +120,12 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor // fall back to the numeric ID only when the metadata is absent (legacy callers). userID := contactUserID(req.Metadata, req.ContactID) + // Message parts: text first, then one file part per downloaded attachment. + // EVO-2180: downloads happen ONCE here (before Marshal), so the byte-identical + // body is reused across retries. + parts := []model.JSONRPCPart{{Type: "text", Text: req.Message}} + parts = append(parts, a.buildFileParts(ctx, req)...) + rpcReq := model.JSONRPCRequest{ JSONRPC: "2.0", ID: fmt.Sprintf("%d:%d", req.ContactID, req.ConversationID), @@ -74,10 +134,8 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor ContextID: contextID, UserID: userID, Message: model.JSONRPCMessage{ - Role: "user", - Parts: []model.JSONRPCPart{ - {Type: "text", Text: req.Message}, - }, + Role: "user", + Parts: parts, }, Metadata: nonNilMetadata(req.Metadata), }, @@ -88,9 +146,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 +227,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 +235,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 +262,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. @@ -167,6 +328,268 @@ func nonNilMetadata(m map[string]any) map[string]any { return m } +// buildFileParts downloads each incoming attachment and returns it as a base64 A2A +// file part. A failure (unreachable/private URL, non-200, oversize, timeout) is +// logged and skipped — the text-only message must always survive a media failure. +// EVO-2180. +func (a *aiAdapter) buildFileParts(ctx context.Context, req *model.A2ARequest) []model.JSONRPCPart { + if len(req.Attachments) == 0 { + return nil + } + perDownload := a.attachmentTimeout() + budgetCtx, cancelBudget := context.WithTimeout(ctx, attachmentsTotalTimeFactor*perDownload) + defer cancelBudget() + + // Built once per turn: the client closes over the authorized hosts. + hosts := allowedMediaHosts() + client := a.mediaClient(hosts) + + parts := make([]model.JSONRPCPart, 0, len(req.Attachments)) + remaining := maxAttachmentsTotalBytes + for i, att := range req.Attachments { + if att.URL == "" { + continue + } + if err := checkMediaURL(att.URL, hosts); err != nil { + slog.Warn("pipeline.ai.attachment.blocked_url", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "file_type", att.FileType, + "error", err, + "hint", "add the host to "+mediaHostAllowlistEnv, + ) + continue + } + if budgetCtx.Err() != nil { + slog.Warn("pipeline.ai.attachment.budget_exhausted", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "limit", "time", + "forwarded", len(parts), + "dropped", len(req.Attachments)-i, + ) + break + } + limit := min(maxAttachmentBytes, remaining) + if limit <= 0 { + slog.Warn("pipeline.ai.attachment.budget_exhausted", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "limit", "bytes", + "forwarded", len(parts), + "dropped", len(req.Attachments)-i, + ) + break + } + data, respContentType, err := downloadAttachment(budgetCtx, client, att.URL, perDownload, limit) + if err != nil { + // Status separates the common 404-from-an-expired-signed-link case from + // an unreachable host; they logged identically before. + slog.Warn("pipeline.ai.attachment.download_failed", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "file_type", att.FileType, + "status", statusOf(err), + "error", err, + ) + continue + } + mimeType, ok := resolveMimeType(att, respContentType) + if !ok { + slog.Warn("pipeline.ai.attachment.skipped_not_media", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "file_type", att.FileType, + "declared_content_type", att.ContentType, + "response_content_type", respContentType, + ) + continue + } + remaining -= len(data) + parts = append(parts, model.JSONRPCPart{ + Type: "file", + File: &model.JSONRPCFile{ + Name: attachmentName(att.URL), + MimeType: mimeType, + Bytes: base64.StdEncoding.EncodeToString(data), + }, + }) + slog.Info("pipeline.ai.attachment.forwarded", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "file_type", att.FileType, + "mime_type", mimeType, + "bytes", len(data), + ) + } + return parts +} + +// attachmentTimeout is the per-download timeout: maxAttachmentDownload, shrunk to +// the configured AI timeout when that is smaller (so a deployment tuned for fast +// failure does not wait longer on media than on the AI call itself). +func (a *aiAdapter) attachmentTimeout() time.Duration { + d := maxAttachmentDownload + if t := time.Duration(a.timeoutSecs) * time.Second; t > 0 && t < d { + d = t + } + return d +} + +// allowedMediaHosts returns the hostnames authorized to serve incoming media. +// An empty set authorizes nothing: failing closed beats fetching a URL the caller +// chose. +func allowedMediaHosts() map[string]struct{} { + hosts := make(map[string]struct{}, 2) + for _, h := range strings.Split(os.Getenv(mediaHostAllowlistEnv), ",") { + if h = strings.ToLower(strings.TrimSpace(h)); h != "" { + hosts[h] = struct{}{} + } + } + return hosts +} + +// checkMediaURL reports why a media URL must not be fetched, or nil when it may be. +func checkMediaURL(rawURL string, hosts map[string]struct{}) error { + u, err := neturl.Parse(rawURL) + if err != nil { + return fmt.Errorf("unparseable media url: %w", err) + } + if scheme := strings.ToLower(u.Scheme); scheme != "http" && scheme != "https" { + return fmt.Errorf("scheme %q is not allowed for media", u.Scheme) + } + host := strings.ToLower(u.Hostname()) + if host == "" { + return errors.New("media url has no host") + } + if _, ok := hosts[host]; !ok { + return fmt.Errorf("host %q is not authorized to serve media for this event", host) + } + return nil +} + +// mediaClient shares the adapter's transport but re-runs checkMediaURL on every +// redirect hop, so an authorized host cannot 302 the download onto an internal one. +func (a *aiAdapter) mediaClient(hosts map[string]struct{}) *http.Client { + return &http.Client{ + Transport: a.client.Transport, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 10 { + return errors.New("stopped after 10 redirects") + } + return checkMediaURL(req.URL.String(), hosts) + }, + } +} + +// httpStatusError carries the status of a non-200 media response. +type httpStatusError struct{ status int } + +func (e *httpStatusError) Error() string { return fmt.Sprintf("unexpected status %d", e.status) } + +// statusOf returns the HTTP status of a download error, or 0 if there was no response. +func statusOf(err error) int { + var se *httpStatusError + if errors.As(err, &se) { + return se.status + } + return 0 +} + +// downloadAttachment GETs the URL with the given client and timeout, reading at most +// limit bytes. It returns the body and the response Content-Type. +func downloadAttachment(ctx context.Context, client *http.Client, url string, timeout time.Duration, limit int) ([]byte, string, error) { + dlCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + httpReq, err := http.NewRequestWithContext(dlCtx, http.MethodGet, url, nil) + if err != nil { + return nil, "", fmt.Errorf("new_request: %w", err) + } + resp, err := client.Do(httpReq) + if err != nil { + return nil, "", fmt.Errorf("do: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, "", &httpStatusError{status: resp.StatusCode} + } + // +1 so an exactly-at-cap read is distinguishable from an oversize one. + data, err := io.ReadAll(io.LimitReader(resp.Body, int64(limit)+1)) + if err != nil { + return nil, "", fmt.Errorf("read: %w", err) + } + if len(data) > limit { + return nil, "", fmt.Errorf("attachment exceeds the %d bytes still available in this call", limit) + } + return data, resp.Header.Get("Content-Type"), nil +} + +// resolveMimeType picks the mime type sent to the AI Processor, which forwards it +// verbatim into the model call (runner_utils: Blob(mime_type=content_type)) — so a +// wrong value here surfaces as a processor-side failure, not a graceful skip. +// +// It prefers the Content-Type of the response actually downloaded (that describes +// the bytes in hand), falls back to what the CRM declared, then to the URL +// extension. It returns false when the payload is a web page: a Rails proxy URL +// answering 200 with an error/login page would otherwise be forwarded as a valid +// image. It also returns false when nothing better than application/octet-stream +// can be determined — an opaque blob is rejected by the model APIs, so dropping it +// keeps the text reply alive instead of failing the whole turn. +func resolveMimeType(att model.Attachment, respContentType string) (string, bool) { + respType := mediaTypeOf(respContentType) + declared := mediaTypeOf(att.ContentType) + if isWebPage(respType) || isWebPage(declared) { + return "", false + } + for _, candidate := range []string{respType, declared, mediaTypeOf(mimeFromURL(att.URL))} { + if candidate != "" && candidate != octetStream { + return candidate, true + } + } + return "", false +} + +const octetStream = "application/octet-stream" + +// mediaTypeOf normalises a Content-Type header to its bare media type +// ("image/jpeg; charset=binary" → "image/jpeg"). +func mediaTypeOf(contentType string) string { + if contentType == "" { + return "" + } + if mt, _, err := mime.ParseMediaType(contentType); err == nil { + return mt + } + return strings.ToLower(strings.TrimSpace(strings.Split(contentType, ";")[0])) +} + +// isWebPage reports whether a media type is an HTML document rather than media. +func isWebPage(mediaType string) bool { + return mediaType == "text/html" || mediaType == "application/xhtml+xml" +} + +// mimeFromURL guesses a mime type from the URL's file extension. ActiveStorage +// proxy URLs keep the original filename, so this recovers the type when neither the +// CRM nor the storage backend declares a useful one. +func mimeFromURL(rawURL string) string { + u, err := neturl.Parse(rawURL) + if err != nil { + return "" + } + return mime.TypeByExtension(path.Ext(u.Path)) +} + +// attachmentName derives a filename from the URL path (fallback "file"). +func attachmentName(rawURL string) string { + if u, err := neturl.Parse(rawURL); err == nil { + if base := path.Base(u.Path); base != "" && base != "." && base != "/" { + return base + } + } + return "file" +} + // conversationContextID resolves the contextId for the JSON-RPC call. It reads the // conversation UUID the CRM nests at metadata.evoai_crm_data.conversation.id and // returns it; if any hop is missing or empty it falls back to the numeric diff --git a/pkg/ai/service/ai_adapter_media_test.go b/pkg/ai/service/ai_adapter_media_test.go new file mode 100644 index 0000000..329387e --- /dev/null +++ b/pkg/ai/service/ai_adapter_media_test.go @@ -0,0 +1,312 @@ +package service_test + +// EVO-2180: incoming media must be downloaded and forwarded to the AI Processor as +// base64 A2A file parts; a download failure must NOT break the text-only message. + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" + aiService "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service" +) + +// procServer stands up a fake AI Processor that captures the JSON-RPC parts it +// receives and returns a minimal successful response. +func procServer(t *testing.T, capture *[]aiModel.JSONRPCPart) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req aiModel.JSONRPCRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("decode request: %v", err) + } + *capture = req.Params.Message.Parts + _ = json.NewEncoder(w).Encode(aiModel.A2AResponse{ + Result: &aiModel.A2AResult{ + Artifacts: []aiModel.A2AArtifact{ + {Parts: []aiModel.A2APart{{Type: "text", Text: "ok"}}}, + }, + }, + }) + })) +} + +func TestCall_ForwardsAttachmentAsFilePart(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + imgBytes := []byte("\x89PNG\r\n-fake-image-bytes") + fileSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write(imgBytes) + })) + defer fileSrv.Close() + + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL + "/api/v1/a2a/agent-1", + ContactID: 1, + ConversationID: 2, + ApiKey: "k", + Message: "look at this", + Attachments: []aiModel.Attachment{ + {URL: fileSrv.URL + "/photo.png", ContentType: "image/png", FileType: "image"}, + }, + }) + if err != nil { + t.Fatalf("Call: %v", err) + } + + if len(parts) != 2 { + t.Fatalf("parts = %d, want 2 (text + file)", len(parts)) + } + if parts[0].Type != "text" || parts[0].Text != "look at this" { + t.Errorf("parts[0] = %+v, want text 'look at this'", parts[0]) + } + if parts[1].Type != "file" || parts[1].File == nil { + t.Fatalf("parts[1] = %+v, want a file part", parts[1]) + } + if parts[1].File.MimeType != "image/png" { + t.Errorf("file mimeType = %q, want image/png", parts[1].File.MimeType) + } + if parts[1].File.Name != "photo.png" { + t.Errorf("file name = %q, want photo.png", parts[1].File.Name) + } + decoded, err := base64.StdEncoding.DecodeString(parts[1].File.Bytes) + if err != nil { + t.Fatalf("file bytes not valid base64: %v", err) + } + if !bytes.Equal(decoded, imgBytes) { + t.Errorf("decoded bytes != original image bytes") + } +} + +func TestCall_AttachmentDownloadFailure_SendsTextOnly(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL + "/api/v1/a2a/agent-1", + ContactID: 1, + ConversationID: 2, + ApiKey: "k", + Message: "hi", + Attachments: []aiModel.Attachment{ + {URL: "http://127.0.0.1:1/unreachable.png", ContentType: "image/png", FileType: "image"}, + }, + }) + if err != nil { + t.Fatalf("a download failure must not error the call: %v", err) + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want a single text part when the download fails", parts) + } +} + +// mediaServer serves fixed bytes with a fixed Content-Type, counting the requests +// it answered so a test can assert which attachments were even attempted. +func mediaServer(contentType string, body []byte, hits *int32) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(hits, 1) + if contentType != "" { + w.Header().Set("Content-Type", contentType) + } + _, _ = w.Write(body) + })) +} + +func filePartsOf(parts []aiModel.JSONRPCPart) []aiModel.JSONRPCPart { + out := make([]aiModel.JSONRPCPart, 0, len(parts)) + for _, p := range parts { + if p.Type == "file" { + out = append(out, p) + } + } + return out +} + +// The debounce window aggregates the media of every message in it, so the shared +// byte budget — not just the per-file cap — is what keeps the request under the +// gateway's client_max_body_size. Over budget the extra media is dropped and the +// call still goes out. +func TestCall_TotalAttachmentBudget_DropsExcessAndStillSends(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + var hits int32 + chunk := make([]byte, 6<<20) // 6 MiB each: 4 of them exceed the 20 MiB budget + fileSrv := mediaServer("image/png", chunk, &hits) + defer fileSrv.Close() + + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + atts := make([]aiModel.Attachment, 0, 4) + for i := 0; i < 4; i++ { + atts = append(atts, aiModel.Attachment{ + URL: fmt.Sprintf("%s/%d.png", fileSrv.URL, i), ContentType: "image/png", FileType: "image", + }) + } + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "album", Attachments: atts, + }); err != nil { + t.Fatalf("a media budget overflow must not error the call: %v", err) + } + + files := filePartsOf(parts) + if len(files) != 3 { + t.Fatalf("forwarded %d file parts, want 3 (4x6 MiB capped at a 20 MiB budget)", len(files)) + } + if len(parts) != 4 || parts[0].Type != "text" { + t.Errorf("parts = %d with parts[0]=%q, want the text part plus 3 files", len(parts), parts[0].Type) + } +} + +// An unreachable media host must not hold the turn hostage: the whole set shares a +// time budget, so latency stays bounded no matter how many attachments arrive. +func TestCall_AttachmentTimeBudget_IsBounded(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + hung := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + })) + defer hung.Close() + + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + atts := make([]aiModel.Attachment, 0, 8) + for i := 0; i < 8; i++ { + atts = append(atts, aiModel.Attachment{URL: fmt.Sprintf("%s/%d.png", hung.URL, i), ContentType: "image/png"}) + } + + // timeoutSecs=1 → 1s per download, 3s for the set. Serial per-attachment + // timeouts would take 8s. + adapter := aiService.NewAIAdapter(1, 0, 1) + start := time.Now() + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", Attachments: atts, + }); err != nil { + t.Fatalf("unreachable media must not error the call: %v", err) + } + elapsed := time.Since(start) + if elapsed > 6*time.Second { + t.Errorf("attachment phase took %v, want it bounded near the 3s budget", elapsed) + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Errorf("parts = %+v, want text only when every download hangs", parts) + } +} + +// A Rails proxy URL that answers 200 with an error/login page must not be +// forwarded as an image: the processor passes the mime straight into the model call. +func TestCall_HTMLResponse_IsNotForwardedAsMedia(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + var hits int32 + htmlSrv := mediaServer("text/html; charset=utf-8", []byte("login"), &hits) + defer htmlSrv.Close() + + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: htmlSrv.URL + "/photo.jpg", ContentType: "image/jpeg", FileType: "image"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want text only — an HTML body is not media", parts) + } +} + +// When the CRM omits content_type the mime must still describe the bytes: the +// response header wins, and the URL extension is the last resort. Anything that +// resolves to an opaque blob is dropped rather than sent as octet-stream, which the +// model APIs reject. +func TestCall_MimeTypeResolution(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + cases := []struct { + name string + respContentType string + declared string + urlPath string + wantMime string // "" = attachment must be skipped + }{ + {"response header wins over a missing declaration", "image/jpeg", "", "/blobs/proxy/abc123", "image/jpeg"}, + {"declared type wins over an opaque response", "application/octet-stream", "image/png", "/blobs/proxy/abc123", "image/png"}, + {"url extension is the last resort", "application/octet-stream", "", "/photo.png", "image/png"}, + {"nothing identifiable is dropped", "application/octet-stream", "", "/blobs/proxy/abc123", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var hits int32 + srv := mediaServer(tc.respContentType, []byte("bytes"), &hits) + defer srv.Close() + + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: srv.URL + tc.urlPath, ContentType: tc.declared, FileType: "image"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + files := filePartsOf(parts) + if tc.wantMime == "" { + if len(files) != 0 { + t.Fatalf("file parts = %d, want the attachment dropped", len(files)) + } + return + } + if len(files) != 1 { + t.Fatalf("file parts = %d, want 1", len(files)) + } + if files[0].File.MimeType != tc.wantMime { + t.Errorf("mimeType = %q, want %q", files[0].File.MimeType, tc.wantMime) + } + }) + } +} + +// A single file over the per-attachment cap is skipped, and the text still goes out. +func TestCall_OversizeAttachment_SendsTextOnly(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + var hits int32 + srv := mediaServer("image/png", make([]byte, (15<<20)+1), &hits) + defer srv.Close() + + var parts []aiModel.JSONRPCPart + proc := procServer(t, &parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: srv.URL + "/big.png", ContentType: "image/png", FileType: "image"}}, + }); err != nil { + t.Fatalf("an oversize attachment must not error the call: %v", err) + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want text only", parts) + } +} 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_ssrf_test.go b/pkg/ai/service/ai_adapter_ssrf_test.go new file mode 100644 index 0000000..ab693f0 --- /dev/null +++ b/pkg/ai/service/ai_adapter_ssrf_test.go @@ -0,0 +1,193 @@ +package service_test + +// The attachment URL and the outgoing_url arrive in the same /events payload, so an +// unvalidated download is a read primitive aimed by its caller. These pin the guard. + +import ( + "context" + "encoding/base64" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" + aiService "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service" +) + +// capturingProc records the parts the adapter posts. +func capturingProc(parts *[]aiModel.JSONRPCPart) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req aiModel.JSONRPCRequest + _ = json.NewDecoder(r.Body).Decode(&req) + *parts = req.Params.Message.Parts + _ = json.NewEncoder(w).Encode(aiModel.A2AResponse{ + Result: &aiModel.A2AResult{ + Artifacts: []aiModel.A2AArtifact{{Parts: []aiModel.A2APart{{Type: "text", Text: "ok"}}}}, + }, + }) + })) +} + +func fileParts(parts []aiModel.JSONRPCPart) []aiModel.JSONRPCPart { + out := make([]aiModel.JSONRPCPart, 0, len(parts)) + for _, p := range parts { + if p.Type == "file" { + out = append(out, p) + } + } + return out +} + +// The exfiltration shape: internal attachment URL, attacker-chosen outgoing_url. +func TestCall_ForeignHostAttachment_IsNotFetched(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "evo-crm.internal") + var reached bool + internal := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write([]byte("internal-credentials")) + })) + defer internal.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: internal.URL + "/latest/meta-data/", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("a blocked attachment must not error the call: %v", err) + } + + if reached { + t.Error("the unauthorized host was fetched — the URL guard did not run") + } + if got := fileParts(parts); len(got) != 0 { + t.Fatalf("file parts = %d, want the attachment dropped", len(got)) + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want the text reply to survive the block", parts) + } +} + +// Blob storage does not always live on the CRM host, so the allowlist must work. +func TestCall_AllowlistedHost_IsFetched(t *testing.T) { + blob := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write([]byte("png-bytes")) + })) + defer blob.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: blob.URL + "/photo.png", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + + got := fileParts(parts) + if len(got) != 1 { + t.Fatalf("file parts = %d, want the allowlisted host forwarded", len(got)) + } + decoded, _ := base64.StdEncoding.DecodeString(got[0].File.Bytes) + if string(decoded) != "png-bytes" { + t.Errorf("decoded = %q, want the blob bytes", decoded) + } +} + +// Checking the declared URL alone is not enough: a 302 could walk it internal. +func TestCall_RedirectOffTheAuthorizedHost_IsRefused(t *testing.T) { + // Authorizes 127.0.0.1 (the redirector) but not "localhost" (the target). + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + var reached bool + internal := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + _, _ = w.Write([]byte("internal-after-redirect")) + })) + defer internal.Close() + + // httptest always binds 127.0.0.1, so the target uses a host string the + // allowlist does not carry. Without the redirect check this would succeed. + target := strings.Replace(internal.URL, "127.0.0.1", "localhost", 1) + redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target+"/secret", http.StatusFound) + })) + defer redirector.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: redirector.URL + "/photo.png", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("a refused redirect must not error the call: %v", err) + } + + if reached { + t.Error("the redirect target was fetched — the redirect is not re-checked") + } + if got := fileParts(parts); len(got) != 0 { + t.Fatalf("file parts = %d, want the redirected download dropped", len(got)) + } +} + +// Only http/https may be addressed. +func TestCall_NonHTTPScheme_IsRejected(t *testing.T) { + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: "file:///etc/passwd", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + if got := fileParts(parts); len(got) != 0 { + t.Fatalf("file parts = %d, want the file:// URL dropped", len(got)) + } +} + +// Nothing authorized means nothing fetched — failing closed is the point. This is +// also what an unconfigured deployment gets, which is why MEDIA_HOST_ALLOWLIST has +// to be shipped alongside the service. +func TestCall_NoAuthorizedHost_ForwardsNothing(t *testing.T) { + var reached bool + blob := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + })) + defer blob.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: blob.URL + "/photo.png", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + if reached { + t.Error("an attachment was fetched with no authorized host configured") + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want text only", parts) + } +} 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") diff --git a/pkg/debounce/service/debounce_engine.go b/pkg/debounce/service/debounce_engine.go index b866df8..1ab9fa2 100644 --- a/pkg/debounce/service/debounce_engine.go +++ b/pkg/debounce/service/debounce_engine.go @@ -12,9 +12,10 @@ import ( // DebounceEngine manages per-pair debounce timers and message buffers. type DebounceEngine interface { - Start(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig) error - Reset(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig) error + Start(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error + Reset(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error GetBuffer(ctx context.Context, contactID, conversationID int64) (string, error) + GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error) TimerExists(ctx context.Context, contactID, conversationID int64) (bool, error) } @@ -27,10 +28,13 @@ func NewDebounceEngine(repo repository.PipelineRepository) DebounceEngine { return &debounceEngine{repo: repo} } -func (d *debounceEngine) Start(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig) error { +func (d *debounceEngine) Start(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error { if err := d.repo.AppendToBuffer(ctx, contactID, conversationID, content); err != nil { return fmt.Errorf("debounce.start.append: %w", err) } + if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil { + return fmt.Errorf("debounce.start.append_attachments: %w", err) + } if cfg.DebounceTime > 0 { ttl := time.Duration(cfg.DebounceTime) * time.Second if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil { @@ -40,10 +44,13 @@ func (d *debounceEngine) Start(ctx context.Context, contactID, conversationID in return nil } -func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig) error { +func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error { if err := d.repo.AppendToBuffer(ctx, contactID, conversationID, content); err != nil { return fmt.Errorf("debounce.reset.append: %w", err) } + if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil { + return fmt.Errorf("debounce.reset.append_attachments: %w", err) + } if cfg.DebounceTime > 0 { ttl := time.Duration(cfg.DebounceTime) * time.Second if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil { @@ -61,6 +68,10 @@ func (d *debounceEngine) GetBuffer(ctx context.Context, contactID, conversationI return strings.Join(entries, "\n\n"), nil } +func (d *debounceEngine) GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error) { + return d.repo.GetAttachments(ctx, contactID, conversationID) +} + func (d *debounceEngine) TimerExists(ctx context.Context, contactID, conversationID int64) (bool, error) { return d.repo.TimerExists(ctx, contactID, conversationID) } diff --git a/pkg/debounce/service/debounce_engine_test.go b/pkg/debounce/service/debounce_engine_test.go index 85f9f9e..162ada8 100644 --- a/pkg/debounce/service/debounce_engine_test.go +++ b/pkg/debounce/service/debounce_engine_test.go @@ -49,7 +49,7 @@ func TestStart_SetsTTLAndBuffer(t *testing.T) { ctx := context.Background() cfg := model.BotConfig{DebounceTime: 5} - if err := eng.Start(ctx, 1, 1, "first message", cfg); err != nil { + if err := eng.Start(ctx, 1, 1, "first message", nil, cfg); err != nil { t.Fatalf("Start returned error: %v", err) } @@ -69,7 +69,7 @@ func TestStart_ZeroDuration_SetsNoTimer(t *testing.T) { ctx := context.Background() cfg := model.BotConfig{DebounceTime: 0} - if err := eng.Start(ctx, 2, 2, "immediate", cfg); err != nil { + if err := eng.Start(ctx, 2, 2, "immediate", nil, cfg); err != nil { t.Fatalf("Start returned error: %v", err) } @@ -84,14 +84,14 @@ func TestReset_RefreshesTTLAndAppendsBuffer(t *testing.T) { ctx := context.Background() cfg := model.BotConfig{DebounceTime: 10} - if err := eng.Start(ctx, 3, 3, "msg1", cfg); err != nil { + if err := eng.Start(ctx, 3, 3, "msg1", nil, cfg); err != nil { t.Fatalf("Start returned error: %v", err) } // Simulate time passing by setting a shorter TTL manually rdb.Expire(ctx, "bot_runtime:timer:3:3", 2*time.Second) - if err := eng.Reset(ctx, 3, 3, "msg2", cfg); err != nil { + if err := eng.Reset(ctx, 3, 3, "msg2", nil, cfg); err != nil { t.Fatalf("Reset returned error: %v", err) } @@ -111,9 +111,9 @@ func TestGetBuffer_ConcatenatesWithDoubleNewline(t *testing.T) { ctx := context.Background() cfg := model.BotConfig{DebounceTime: 5} - _ = eng.Start(ctx, 4, 4, "hello", cfg) - _ = eng.Reset(ctx, 4, 4, "world", cfg) - _ = eng.Reset(ctx, 4, 4, "again", cfg) + _ = eng.Start(ctx, 4, 4, "hello", nil, cfg) + _ = eng.Reset(ctx, 4, 4, "world", nil, cfg) + _ = eng.Reset(ctx, 4, 4, "again", nil, cfg) result, err := eng.GetBuffer(ctx, 4, 4) if err != nil { @@ -130,7 +130,7 @@ func TestTimerExists_ReturnsFalseAfterExpiry(t *testing.T) { ctx := context.Background() cfg := model.BotConfig{DebounceTime: 1} // 1 second TTL - if err := eng.Start(ctx, 5, 5, "msg", cfg); err != nil { + if err := eng.Start(ctx, 5, 5, "msg", nil, cfg); err != nil { t.Fatalf("Start returned error: %v", err) } diff --git a/pkg/pipeline/handler/handler_test.go b/pkg/pipeline/handler/handler_test.go index 6f25bf6..eb4f75a 100644 --- a/pkg/pipeline/handler/handler_test.go +++ b/pkg/pipeline/handler/handler_test.go @@ -39,6 +39,12 @@ func (m *mockRepo) AppendToBuffer(_ context.Context, _, _ int64, _ string) error func (m *mockRepo) GetBuffer(_ context.Context, _, _ int64) ([]string, error) { return nil, nil } +func (m *mockRepo) AppendAttachments(_ context.Context, _, _ int64, _ []model.Attachment) error { + return nil +} +func (m *mockRepo) GetAttachments(_ context.Context, _, _ int64) ([]model.Attachment, error) { + return nil, nil +} func (m *mockRepo) SetTimer(_ context.Context, _, _ int64, _ time.Duration) error { return nil } func (m *mockRepo) DeleteTimer(_ context.Context, _, _ int64) error { return nil } func (m *mockRepo) TimerExists(_ context.Context, _, _ int64) (bool, error) { return false, nil } @@ -48,7 +54,7 @@ func (m *mockRepo) AcquireLock(_ context.Context, _, _ int64) (repository.Mutex, func (m *mockRepo) ScanStates(_ context.Context, _ int) ([]model.PairID, error) { return nil, nil } -func (m *mockRepo) Ping(_ context.Context) error { return m.pingErr } +func (m *mockRepo) Ping(_ context.Context) error { return m.pingErr } // mockSvc satisfies pipelineService.PipelineService for handler tests. type mockSvc struct{ processErr error } diff --git a/pkg/pipeline/model/pipeline.go b/pkg/pipeline/model/pipeline.go index 583ad2c..cf39048 100644 --- a/pkg/pipeline/model/pipeline.go +++ b/pkg/pipeline/model/pipeline.go @@ -27,14 +27,14 @@ type PairID struct { // BotConfig and PostbackURL are persisted for StageDebounce so that the service can // reconstruct the pipelineEntry correctly after a restart (NFR-01 recovery). type PipelineState struct { - Stage Stage `json:"stage"` - CreatedAt time.Time `json:"created_at"` - BotConfig BotConfig `json:"bot_config,omitempty"` - PostbackURL string `json:"postback_url,omitempty"` - AgentBotID string `json:"agent_bot_id,omitempty"` - ApiKey string `json:"api_key,omitempty"` - OutgoingURL string `json:"outgoing_url,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` + Stage Stage `json:"stage"` + CreatedAt time.Time `json:"created_at"` + BotConfig BotConfig `json:"bot_config,omitempty"` + PostbackURL string `json:"postback_url,omitempty"` + AgentBotID string `json:"agent_bot_id,omitempty"` + ApiKey string `json:"api_key,omitempty"` + OutgoingURL string `json:"outgoing_url,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` } // MessageEvent is the inbound payload from evo-ai-crm AgentBotListener. @@ -45,6 +45,7 @@ type MessageEvent struct { ContactID int64 `json:"contact_id"` MessageID string `json:"message_id"` MessageContent string `json:"message_content"` + Attachments []Attachment `json:"attachments,omitempty"` // EVO-2180: incoming media (image/audio/…) ApiKey string `json:"api_key"` OutgoingURL string `json:"outgoing_url"` BotConfig BotConfig `json:"bot_config"` @@ -52,15 +53,24 @@ type MessageEvent struct { Metadata map[string]any `json:"metadata,omitempty"` } +// Attachment is an incoming media item forwarded by the CRM. The AI adapter +// downloads it and sends it to the AI Processor as a base64 A2A file part. +// EVO-2180. JSON tags match the CRM DelegationService payload. +type Attachment struct { + URL string `json:"url"` + ContentType string `json:"content_type"` + FileType string `json:"file_type"` +} + // BotConfig carries per-bot runtime configuration provided by the caller. // Bot Runtime must not make any outbound call to fetch config (FR-24). type BotConfig struct { - DebounceTime int `json:"debounce_time"` // seconds; 0 = pass-through + DebounceTime int `json:"debounce_time"` // seconds; 0 = pass-through MessageSignature string `json:"message_signature"` TextSegmentationEnabled bool `json:"text_segmentation_enabled"` - TextSegmentationLimit int `json:"text_segmentation_limit"` // max chars per segment + TextSegmentationLimit int `json:"text_segmentation_limit"` // max chars per segment TextSegmentationMinSize int `json:"text_segmentation_min_size"` - DelayPerCharacter float64 `json:"delay_per_character"` // ms per char between parts + DelayPerCharacter float64 `json:"delay_per_character"` // ms per char between parts } // Validate checks semantic constraints on a MessageEvent after JSON binding. diff --git a/pkg/pipeline/repository/pipeline_repository.go b/pkg/pipeline/repository/pipeline_repository.go index c5f18d6..97670da 100644 --- a/pkg/pipeline/repository/pipeline_repository.go +++ b/pkg/pipeline/repository/pipeline_repository.go @@ -22,6 +22,11 @@ type PipelineRepository interface { AppendToBuffer(ctx context.Context, contactID, conversationID int64, content string) error GetBuffer(ctx context.Context, contactID, conversationID int64) ([]string, error) + // EVO-2180: incoming media buffer, aggregated across the debounce window exactly + // like the text buffer. Cleared together with the text buffer in ClearState. + AppendAttachments(ctx context.Context, contactID, conversationID int64, atts []model.Attachment) error + GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error) + SetTimer(ctx context.Context, contactID, conversationID int64, ttl time.Duration) error DeleteTimer(ctx context.Context, contactID, conversationID int64) error TimerExists(ctx context.Context, contactID, conversationID int64) (bool, error) diff --git a/pkg/pipeline/repository/redis_pipeline_repository.go b/pkg/pipeline/repository/redis_pipeline_repository.go index 314a77b..48ca9d5 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository.go +++ b/pkg/pipeline/repository/redis_pipeline_repository.go @@ -54,6 +54,7 @@ func (r *redisPipelineRepository) ClearState(ctx context.Context, contactID, con keys := []string{ stateKey(contactID, conversationID), bufferKey(contactID, conversationID), + attachBufferKey(contactID, conversationID), // EVO-2180: clear media with the text buffer timerKey(contactID, conversationID), } return r.rdb.Del(ctx, keys...).Err() @@ -67,6 +68,46 @@ func (r *redisPipelineRepository) GetBuffer(ctx context.Context, contactID, conv return r.rdb.LRange(ctx, bufferKey(contactID, conversationID), 0, -1).Result() } +// AppendAttachments RPushes each attachment (JSON-encoded) onto the media buffer, +// aggregating across the debounce window like the text buffer. EVO-2180. +func (r *redisPipelineRepository) AppendAttachments(ctx context.Context, contactID, conversationID int64, atts []model.Attachment) error { + if len(atts) == 0 { + return nil + } + values := make([]interface{}, 0, len(atts)) + for _, att := range atts { + b, err := json.Marshal(att) + if err != nil { + return fmt.Errorf("pipeline.repository.append_attachments marshal: %w", err) + } + values = append(values, b) + } + key := attachBufferKey(contactID, conversationID) + if err := r.rdb.RPush(ctx, key, values...).Err(); err != nil { + return err + } + // Leak backstop for turns that die before ClearState; not the debounce window. + return r.rdb.Expire(ctx, key, attachBufferTTL).Err() +} + +// GetAttachments returns the media aggregated during the debounce window. EVO-2180. +func (r *redisPipelineRepository) GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error) { + raw, err := r.rdb.LRange(ctx, attachBufferKey(contactID, conversationID), 0, -1).Result() + if err != nil { + return nil, fmt.Errorf("pipeline.repository.get_attachments: %w", err) + } + atts := make([]model.Attachment, 0, len(raw)) + for _, s := range raw { + var att model.Attachment + if err := json.Unmarshal([]byte(s), &att); err != nil { + slog.Warn("pipeline.repository.get_attachments.skip_malformed", "error", err) + continue + } + atts = append(atts, att) + } + return atts, nil +} + func (r *redisPipelineRepository) SetTimer(ctx context.Context, contactID, conversationID int64, ttl time.Duration) error { return r.rdb.Set(ctx, timerKey(contactID, conversationID), "1", ttl).Err() } @@ -139,6 +180,14 @@ func bufferKey(contactID, conversationID int64) string { return fmt.Sprintf("bot_runtime:buffer:%d:%d", contactID, conversationID) } +// attachBufferTTL outlives any turn but stays under the 15-minute TTL the CRM signs +// the media URLs with, so an orphaned entry never outlives its links. +const attachBufferTTL = 10 * time.Minute + +func attachBufferKey(contactID, conversationID int64) string { + return fmt.Sprintf("bot_runtime:attach:%d:%d", contactID, conversationID) +} + func lockKey(contactID, conversationID int64) string { return fmt.Sprintf("bot_runtime:lock:%d:%d", contactID, conversationID) } diff --git a/pkg/pipeline/repository/redis_pipeline_repository_test.go b/pkg/pipeline/repository/redis_pipeline_repository_test.go index 05a3950..13f9460 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository_test.go +++ b/pkg/pipeline/repository/redis_pipeline_repository_test.go @@ -8,8 +8,8 @@ import ( "testing" "time" - goredis "github.com/go-redsync/redsync/v4/redis/goredis/v9" "github.com/go-redsync/redsync/v4" + goredis "github.com/go-redsync/redsync/v4/redis/goredis/v9" "github.com/redis/go-redis/v9" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" @@ -183,6 +183,45 @@ func TestSetTimer_DeleteTimer_TimerExists(t *testing.T) { } } +func TestAppendGetAttachments_RoundTrip(t *testing.T) { + repo, cleanup := setupTestRepo(t) + defer cleanup() + + ctx := context.Background() + var contactID, convID int64 = 4242, 4243 + + atts := []model.Attachment{ + {URL: "http://x/a.png", ContentType: "image/png", FileType: "image"}, + {URL: "http://x/b.ogg", ContentType: "audio/ogg", FileType: "audio"}, + } + // Two appends (as a multi-message debounce window would) must aggregate in order. + if err := repo.AppendAttachments(ctx, contactID, convID, atts[:1]); err != nil { + t.Fatalf("AppendAttachments 1: %v", err) + } + if err := repo.AppendAttachments(ctx, contactID, convID, atts[1:]); err != nil { + t.Fatalf("AppendAttachments 2: %v", err) + } + + got, err := repo.GetAttachments(ctx, contactID, convID) + if err != nil { + t.Fatalf("GetAttachments: %v", err) + } + if len(got) != 2 || got[0].URL != atts[0].URL || got[1].FileType != "audio" { + t.Fatalf("GetAttachments = %+v, want ordered %+v", got, atts) + } + + // Empty append is a no-op. + if err := repo.AppendAttachments(ctx, contactID, convID, nil); err != nil { + t.Fatalf("AppendAttachments(nil): %v", err) + } + + // Empty key returns an empty slice. + empty, err := repo.GetAttachments(ctx, 9191, 9191) + if err != nil || len(empty) != 0 { + t.Fatalf("GetAttachments(empty) = %v (err %v)", empty, err) + } +} + func TestClearState_DeletesAllKeys(t *testing.T) { repo, cleanup := setupTestRepo(t) defer cleanup() @@ -190,9 +229,10 @@ func TestClearState_DeletesAllKeys(t *testing.T) { ctx := context.Background() var contactID, convID int64 = 103, 203 - // Set state, buffer, timer + // Set state, buffer, attachments, timer _ = repo.SetState(ctx, contactID, convID, &model.PipelineState{Stage: model.StageAI}) _ = repo.AppendToBuffer(ctx, contactID, convID, "msg") + _ = repo.AppendAttachments(ctx, contactID, convID, []model.Attachment{{URL: "http://x/a.png", ContentType: "image/png", FileType: "image"}}) _ = repo.SetTimer(ctx, contactID, convID, 30*time.Second) if err := repo.ClearState(ctx, contactID, convID); err != nil { @@ -209,6 +249,11 @@ func TestClearState_DeletesAllKeys(t *testing.T) { t.Errorf("expected empty buffer after ClearState, got %v (err: %v)", buf, err) } + atts, err := repo.GetAttachments(ctx, contactID, convID) + if err != nil || len(atts) != 0 { + t.Errorf("expected empty attachments after ClearState, got %v (err: %v)", atts, err) + } + exists, err := repo.TimerExists(ctx, contactID, convID) if err != nil || exists { t.Errorf("expected timer gone after ClearState, exists=%v (err: %v)", exists, err) diff --git a/pkg/pipeline/service/ai_failure_notice_test.go b/pkg/pipeline/service/ai_failure_notice_test.go new file mode 100644 index 0000000..85a4b1d --- /dev/null +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -0,0 +1,210 @@ +package service + +import ( + "context" + "errors" + "os" + "strings" + "testing" + "time" + + brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" + "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" +) + +// CRM-236: a degraded provider used to end the turn in silence, while the tool's +// side effect (a moved pipeline card) had already been applied. + +func captureDispatch(t *testing.T) (*mockDispatchEngine, *[]string) { + t.Helper() + var sent []string + engine := &mockDispatchEngine{ + dispatchFn: func(_ context.Context, _, _ int64, content string, _ model.BotConfig, _ string) error { + sent = append(sent, content) + return nil + }, + } + return engine, &sent +} + +func TestAIFailureNotice_TimeoutTellsTheCustomer(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 1 { + t.Fatalf("expected the customer to receive one notice, got %d", len(*sent)) + } + if (*sent)[0] != defaultAIFailureNotice { + t.Errorf("unexpected notice: %q", (*sent)[0]) + } +} + +func TestAIFailureNotice_NeverLeaksTheProviderError(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + // A real provider error: model names, quota ids and URLs must not reach the customer. + cause := errors.New("litellm.RateLimitError: VertexAIException - 429 RESOURCE_EXHAUSTED " + + "Quota exceeded for metric generativelanguage.googleapis.com/generate_content_free_tier_requests, " + + "limit: 20, model: gemini-2.5-flash") + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", cause) + + if len(*sent) != 1 { + t.Fatalf("expected one notice, got %d", len(*sent)) + } + for _, leak := range []string{"gemini", "Quota", "RateLimitError", "googleapis"} { + if strings.Contains((*sent)[0], leak) { + t.Errorf("provider detail %q leaked to the customer: %q", leak, (*sent)[0]) + } + } +} + +func TestAIFailureNotice_OperatorCanCustomiseIt(t *testing.T) { + t.Setenv(aiFailureNoticeEnv, "Nosso atendimento automático está indisponível.") + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 1 || (*sent)[0] != "Nosso atendimento automático está indisponível." { + t.Fatalf("custom notice not used: %v", *sent) + } +} + +// An operator who prefers silence must be able to keep it. +func TestAIFailureNotice_EmptyEnvDisablesIt(t *testing.T) { + t.Setenv(aiFailureNoticeEnv, "") + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 0 { + t.Fatalf("notice should be disabled, got %v", *sent) + } +} + +func TestAIFailureNotice_NoPostbackUrlIsNotACrash(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "", brtErrors.ErrAITimeout) + + if len(*sent) != 0 { + t.Fatalf("nothing can be dispatched without a postback url, got %v", *sent) + } +} + +// The default must survive an env var that exists but is unrelated. +func TestAIFailureNotice_DefaultWhenEnvUnset(t *testing.T) { + // Low 13: restore whatever the process had, instead of leaving the env mutated + // for every test that runs after this one. + if previous, had := os.LookupEnv(aiFailureNoticeEnv); had { + t.Cleanup(func() { os.Setenv(aiFailureNoticeEnv, previous) }) + } + os.Unsetenv(aiFailureNoticeEnv) + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 1 || (*sent)[0] != defaultAIFailureNotice { + t.Fatalf("expected the default notice, got %v", *sent) + } +} + +func TestAIFailureNotice_DoesNotWriteTurnState(t *testing.T) { + engine, _ := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + // The callers already cleared the state before asking for the notice; writing + // StageDone here would resurrect state for a turn that is over — and, in the + // follow-up race, stamp it over the NEW turn's state. + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + // A read error must fail the test, not pass it: GetState returns (nil, nil) + // for a missing key, so nil-with-error proves nothing about what was written. + state, err := svc.repo.GetState(context.Background(), 1, 2) + if err != nil { + t.Fatalf("could not read the state back: %v", err) + } + if state != nil { + t.Fatalf("the notice wrote turn state: stage=%v", state.Stage) + } +} + +// The entry, not just the state: entries.Delete(pairKey) orphaned the follow-up +// turn, so the message after it started a second concurrent pipeline. +func TestAIFailureNotice_DoesNotTouchTheEntryOfTheNextTurn(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + // The follow-up turn, exactly as startDebounce leaves it: an entry in the map + // and StageDebounce in Redis, both under the pair the notice is about to use. + key := pairKey(1, 2) + nextTurn, cancelNextTurn := context.WithCancel(context.Background()) + defer cancelNextTurn() + svc.entries.Store(key, pipelineEntry{ctx: nextTurn, cancel: cancelNextTurn}) + debounce := &model.PipelineState{Stage: model.StageDebounce, CreatedAt: time.Now()} + if err := svc.repo.SetState(context.Background(), 1, 2, debounce); err != nil { + t.Fatalf("could not seed the next turn's state: %v", err) + } + t.Cleanup(func() { svc.repo.ClearState(context.Background(), 1, 2) }) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + // Guard the guard: the bookkeeping only runs after a successful dispatch, so + // a notice that never went out would satisfy the assertions below for free. + if len(*sent) != 1 { + t.Fatalf("the notice never dispatched, so this proves nothing: %v", *sent) + } + + stored, ok := svc.entries.Load(key) + if !ok { + t.Fatal("the notice deleted the next turn's entry: it can no longer be cancelled, so the message after it starts a second concurrent pipeline") + } + if entry, _ := stored.(pipelineEntry); entry.ctx != nextTurn { + t.Error("the next turn's entry was replaced by the notice") + } + + // SetState(StageDone) followed by ClearState leaves nothing behind, so only a + // seeded state can witness it: the next turn must still be in StageDebounce. + state, err := svc.repo.GetState(context.Background(), 1, 2) + if err != nil { + t.Fatalf("could not read the next turn's state back: %v", err) + } + if state == nil { + t.Fatal("the notice cleared the next turn's state: its debounce is lost") + } + if state.Stage != model.StageDebounce { + t.Errorf("the notice stamped the next turn's state: stage=%v, want %v", state.Stage, model.StageDebounce) + } +} + +// The notice is a real dispatch (segmented, with per-rune delays), not a cleanup +// call. Bounding it with cleanupCtx's 5s truncated it and then logged +// "New message arrived" when nothing had arrived. +func TestAIFailureNotice_HasRoomForASegmentedDispatch(t *testing.T) { + ctx, cancel := noticeCtx() + defer cancel() + + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("the notice dispatch must stay bounded") + } + if remaining := time.Until(deadline); remaining <= 10*time.Second { + t.Fatalf("notice budget is %v; a segmented dispatch with per-rune delays needs more", remaining) + } +} + +// The default reaches customers of installations that never chose Portuguese. +func TestAIFailureNotice_DefaultIsLocaleNeutralEnglish(t *testing.T) { + for _, ptBR := range []string{"instabilidade", "Já retorno", "não consegui"} { + if strings.Contains(defaultAIFailureNotice, ptBR) { + t.Errorf("default notice still hardcodes pt-BR (%q): %q", ptBR, defaultAIFailureNotice) + } + } +} diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index d347c57..c2983d2 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "os" "runtime/debug" "strconv" "strings" @@ -33,11 +34,11 @@ type PipelineService interface { type pipelineEntry struct { ctx context.Context cancel context.CancelFunc - cfg model.BotConfig // carries BotConfig from MessageEvent to dispatch stage - postbackURL string // carries PostbackURL from MessageEvent to dispatch stage - outgoingURL string // carries OutgoingURL (full A2A endpoint) from MessageEvent to AI stage - apiKey string // carries ApiKey from MessageEvent to AI stage - metadata map[string]any // carries Metadata from MessageEvent to AI stage (for tools context) + cfg model.BotConfig // carries BotConfig from MessageEvent to dispatch stage + postbackURL string // carries PostbackURL from MessageEvent to dispatch stage + outgoingURL string // carries OutgoingURL (full A2A endpoint) from MessageEvent to AI stage + apiKey string // carries ApiKey from MessageEvent to AI stage + metadata map[string]any // carries Metadata from MessageEvent to AI stage (for tools context) } type pipelineService struct { @@ -46,16 +47,16 @@ type pipelineService struct { aiAdapter aiIface.AIAdapter dispatchEng dispatchIface.DispatchEngine entries sync.Map // string → pipelineEntry - stopCh chan struct{} // closed by Shutdown to stop pollDebounceExpiry - stoppedCh chan struct{} // closed by pollDebounceExpiry when it exits + stopCh chan struct{} // closed by Shutdown to stop pollDebounceExpiry + stoppedCh chan struct{} // closed by pollDebounceExpiry when it exits stopOnce sync.Once // ensures stopCh is closed exactly once } // NewPipelineService constructs the service. Returns interface (GEAR R03). func NewPipelineService( - repo repository.PipelineRepository, - debounce debounceIface.DebounceEngine, - aiAdapter aiIface.AIAdapter, + repo repository.PipelineRepository, + debounce debounceIface.DebounceEngine, + aiAdapter aiIface.AIAdapter, dispatchEng dispatchIface.DispatchEngine, ) PipelineService { return &pipelineService{ @@ -169,7 +170,7 @@ func (s *pipelineService) startDebounce(ctx context.Context, event *model.Messag metadata: event.Metadata, }) - if err := s.debounce.Start(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.BotConfig); err != nil { + if err := s.debounce.Start(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.Attachments, event.BotConfig); err != nil { cancel() s.entries.Delete(key) return fmt.Errorf("pipeline.debounce.start: %w", err) @@ -216,7 +217,7 @@ func (s *pipelineService) skipDebounce(ctx context.Context, event *model.Message // debounce.Start appends to buffer; DebounceTime=0 means no timer (Story 2.1). if err := s.debounce.Start(ctx, event.ContactID, event.ConversationID, - event.MessageContent, event.BotConfig); err != nil { + event.MessageContent, event.Attachments, event.BotConfig); err != nil { cancel() s.entries.Delete(key) return fmt.Errorf("pipeline.skip_debounce.start: %w", err) @@ -229,6 +230,18 @@ func (s *pipelineService) skipDebounce(ctx context.Context, event *model.Message return fmt.Errorf("pipeline.skip_debounce.get_buffer: %w", err) } + // A media read failure must never cost the customer the text reply (EVO-2180): + // log it and go on with no attachments. + atts, err := s.debounce.GetAttachments(ctx, event.ContactID, event.ConversationID) + if err != nil { + slog.Warn("pipeline.skip_debounce.get_attachments_failed", + "contact_id", event.ContactID, + "conversation_id", event.ConversationID, + "error", err, + ) + atts = nil + } + newState := &model.PipelineState{Stage: model.StageAI, CreatedAt: time.Now()} if err := s.repo.SetState(ctx, event.ContactID, event.ConversationID, newState); err != nil { cancel() @@ -240,12 +253,12 @@ func (s *pipelineService) skipDebounce(ctx context.Context, event *model.Message "contact_id", event.ContactID, "conversation_id", event.ConversationID, ) - s.launchAIStage(event.ContactID, event.ConversationID, buffer) + s.launchAIStage(event.ContactID, event.ConversationID, buffer, atts) return nil } func (s *pipelineService) resetDebounce(ctx context.Context, event *model.MessageEvent) error { - if err := s.debounce.Reset(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.BotConfig); err != nil { + if err := s.debounce.Reset(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.Attachments, event.BotConfig); err != nil { return fmt.Errorf("pipeline.debounce.reset: %w", err) } slog.Info("pipeline.debounce.reset", @@ -289,6 +302,18 @@ func (s *pipelineService) advanceToAI(contactID, conversationID int64) { return } + // Media is best-effort: a failure here must not drop the turn's text reply + // (EVO-2180). + atts, err := s.debounce.GetAttachments(ctx, contactID, conversationID) + if err != nil { + slog.Warn("pipeline.debounce.get_attachments_failed", + "contact_id", contactID, + "conversation_id", conversationID, + "error", err, + ) + atts = nil + } + newState := &model.PipelineState{Stage: model.StageAI, CreatedAt: time.Now()} if err := s.repo.SetState(ctx, contactID, conversationID, newState); err != nil { slog.Error("pipeline.debounce.set_ai_state_failed", @@ -304,13 +329,13 @@ func (s *pipelineService) advanceToAI(contactID, conversationID int64) { "conversation_id", conversationID, "buffer_len", len(buffer), ) - s.launchAIStage(contactID, conversationID, buffer) + s.launchAIStage(contactID, conversationID, buffer, atts) } // launchAIStage launches the AI goroutine with the stored pipeline context. // Must be called only after pipelineEntry is stored in s.entries (guaranteed by // startDebounce/skipDebounce/advanceToAI). -func (s *pipelineService) launchAIStage(contactID, conversationID int64, buffer string) { +func (s *pipelineService) launchAIStage(contactID, conversationID int64, buffer string, atts []model.Attachment) { key := pairKey(contactID, conversationID) v, ok := s.entries.Load(key) if !ok { @@ -329,12 +354,12 @@ func (s *pipelineService) launchAIStage(contactID, conversationID int64, buffer ) return } - go s.runAIStage(entry.ctx, contactID, conversationID, buffer, entry.cfg, entry.postbackURL) + go s.runAIStage(entry.ctx, contactID, conversationID, buffer, atts, entry.cfg, entry.postbackURL) } // runAIStage is the AI stage goroutine body. ctx is pipelineEntry.ctx — cancelled by // Process when a new message arrives for the same pair. -func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversationID int64, buffer string, cfg model.BotConfig, postbackURL string) { +func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversationID int64, buffer string, atts []model.Attachment, cfg model.BotConfig, postbackURL string) { defer s.recoverPipeline(contactID, conversationID) slog.Info("pipeline.ai.started", @@ -355,6 +380,17 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio } } + // EVO-2180: forward incoming media (aggregated over the debounce window) so the + // adapter can download + base64-encode it into A2A file parts. + aiAttachments := make([]aiModel.Attachment, 0, len(atts)) + for _, a := range atts { + aiAttachments = append(aiAttachments, aiModel.Attachment{ + URL: a.URL, + ContentType: a.ContentType, + FileType: a.FileType, + }) + } + resp, err := s.aiAdapter.Call(ctx, &aiModel.A2ARequest{ OutgoingURL: outgoingURL, ContactID: contactID, @@ -362,6 +398,7 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio ApiKey: apiKey, Message: buffer, Metadata: metadata, + Attachments: aiAttachments, }) if err != nil { switch { @@ -379,6 +416,10 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio "conversation_id", conversationID, ) s.clearStateWithLog(contactID, conversationID) + // CRM-236: silence is indistinguishable from "the bot is ignoring you", + // and the tool's side effect may already be applied (the card moved at + // ~20s, the timeout fired at 30s). Tell the customer something. + s.sendAIFailureNotice(contactID, conversationID, cfg, postbackURL, err) default: slog.Error("pipeline.ai.error", "contact_id", contactID, @@ -386,6 +427,7 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio "error", fmt.Errorf("pipeline.ai: %w", err), ) s.clearStateWithLog(contactID, conversationID) + s.sendAIFailureNotice(contactID, conversationID, cfg, postbackURL, err) } return } @@ -417,12 +459,12 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio // launchDispatchStage launches the dispatch goroutine. func (s *pipelineService) launchDispatchStage( - ctx context.Context, - contactID int64, + ctx context.Context, + contactID int64, conversationID int64, - resp *aiModel.NormalizedResponse, - cfg model.BotConfig, - postbackURL string, + resp *aiModel.NormalizedResponse, + cfg model.BotConfig, + postbackURL string, ) { go s.runDispatchStage(ctx, contactID, conversationID, resp.Content, cfg, postbackURL) } @@ -430,17 +472,17 @@ func (s *pipelineService) launchDispatchStage( // runDispatchStage is the dispatch stage goroutine body. ctx is pipelineEntry.ctx — cancelled by // Process when a new message arrives for the same pair. func (s *pipelineService) runDispatchStage( - ctx context.Context, - contactID int64, + ctx context.Context, + contactID int64, conversationID int64, - content string, - cfg model.BotConfig, - postbackURL string, + content string, + cfg model.BotConfig, + postbackURL string, ) { defer s.recoverPipeline(contactID, conversationID) slog.Info("pipeline.dispatch.started", - "contact_id", contactID, + "contact_id", contactID, "conversation_id", conversationID, ) start := time.Now() @@ -455,14 +497,14 @@ func (s *pipelineService) runDispatchStage( // atomically. A Delete here would race with the new event's Store // and could delete the replacement entry. slog.Info("pipeline.dispatch.cancelled", - "contact_id", contactID, + "contact_id", contactID, "conversation_id", conversationID, ) default: slog.Error("pipeline.dispatch.error", - "contact_id", contactID, + "contact_id", contactID, "conversation_id", conversationID, - "error", err, + "error", err, ) s.clearStateWithLog(contactID, conversationID) } @@ -475,9 +517,9 @@ func (s *pipelineService) runDispatchStage( doneCtx, doneCancel := cleanupCtx() if err := s.repo.SetState(doneCtx, contactID, conversationID, doneState); err != nil { slog.Warn("pipeline.dispatch.set_done_failed", - "contact_id", contactID, + "contact_id", contactID, "conversation_id", conversationID, - "error", err, + "error", err, ) } doneCancel() @@ -485,9 +527,9 @@ func (s *pipelineService) runDispatchStage( s.entries.Delete(pairKey(contactID, conversationID)) slog.Info("pipeline.dispatch.completed", - "contact_id", contactID, + "contact_id", contactID, "conversation_id", conversationID, - "duration", dur.String(), + "duration", dur.String(), ) } @@ -607,6 +649,76 @@ func cleanupCtx() (context.Context, context.CancelFunc) { return context.WithTimeout(context.Background(), 5*time.Second) } +// aiFailureNoticeEnv overrides the message the customer receives when the AI +// backend times out or errors. Empty string disables the notice entirely, for +// operators who prefer silence to a canned reply. +const aiFailureNoticeEnv = "AI_FAILURE_NOTICE" + +// English: the runtime ships worldwide and a pt-BR default reached installs that +// never chose it. Operators localise it with AI_FAILURE_NOTICE. +const defaultAIFailureNotice = "We are having a temporary issue and could not answer right now. We will get back to you shortly." + +// sendAIFailureNotice replaces the silent turn with one sentence to the customer. +// The provider's raw error goes to the operator's log, never to the chat. +func (s *pipelineService) sendAIFailureNotice( + contactID, conversationID int64, + cfg model.BotConfig, + postbackURL string, + cause error, +) { + notice := defaultAIFailureNotice + if v, ok := os.LookupEnv(aiFailureNoticeEnv); ok { + if strings.TrimSpace(v) == "" { + slog.Info("pipeline.ai.failure_notice.disabled", + "contact_id", contactID, + "conversation_id", conversationID, + ) + return + } + notice = v + } + + if postbackURL == "" { + slog.Warn("pipeline.ai.failure_notice.no_postback", + "contact_id", contactID, + "conversation_id", conversationID, + ) + return + } + + slog.Warn("pipeline.ai.failure_notice.sending", + "contact_id", contactID, + "conversation_id", conversationID, + "cause", cause.Error(), + ) + + // Dispatch directly: runDispatchStage ends in entries.Delete(pairKey), which + // would orphan a follow-up turn. Both callers already cleared the state. + ctx, cancel := noticeCtx() + defer cancel() + defer s.recoverPipeline(contactID, conversationID) + + if err := s.dispatchEng.Dispatch(ctx, contactID, conversationID, notice, cfg, postbackURL); err != nil { + slog.Warn("pipeline.ai.failure_notice.failed", + "contact_id", contactID, + "conversation_id", conversationID, + "error", err, + ) + return + } + + slog.Info("pipeline.ai.failure_notice.sent", + "contact_id", contactID, + "conversation_id", conversationID, + ) +} + +// noticeCtx bounds the notice's dispatch. Not cleanupCtx: a Dispatch segments the +// text and sleeps per rune between parts, which overruns its 5s. +func noticeCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 30*time.Second) +} + // clearStateWithLog calls ClearState and logs a warning if it fails. // Used in all goroutine error/cleanup paths where the error is non-actionable // but should not be silently swallowed. diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index f500faf..a19197f 100644 --- a/pkg/pipeline/service/pipeline_service_test.go +++ b/pkg/pipeline/service/pipeline_service_test.go @@ -15,8 +15,8 @@ import ( brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" "github.com/EvolutionAPI/evo-bot-runtime/internal/testhelpers" - aiIface "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service" aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" + aiIface "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service" debounceService "github.com/EvolutionAPI/evo-bot-runtime/pkg/debounce/service" dispatchIface "github.com/EvolutionAPI/evo-bot-runtime/pkg/dispatch/service" "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" @@ -58,11 +58,14 @@ type mockDebounce struct { startErr error } -func (m *mockDebounce) Start(_ context.Context, _, _ int64, _ string, _ model.BotConfig) error { +func (m *mockDebounce) GetAttachments(_ context.Context, _, _ int64) ([]model.Attachment, error) { + return nil, nil +} +func (m *mockDebounce) Start(_ context.Context, _, _ int64, _ string, _ []model.Attachment, _ model.BotConfig) error { m.startCalled = true return m.startErr } -func (m *mockDebounce) Reset(_ context.Context, _, _ int64, _ string, _ model.BotConfig) error { +func (m *mockDebounce) Reset(_ context.Context, _, _ int64, _ string, _ []model.Attachment, _ model.BotConfig) error { m.resetCalled = true return nil } @@ -78,7 +81,6 @@ func (m *mockFailLockRepo) AcquireLock(_ context.Context, _, _ int64) (repositor return nil, errors.New("lock unavailable") } - // --- setup --- func TestMain(m *testing.M) { @@ -710,3 +712,107 @@ func TestPipeline_DispatchError_ClearsState(t *testing.T) { } } +// EVO-2180: the event's attachments must survive the whole path — HTTP event → +// debounce buffer in Redis → A2ARequest. The adapter-level test proves the file +// part is built from an A2ARequest; this one proves the A2ARequest actually gets +// the media, which is the seam a refactor would silently drop. +func TestProcess_ForwardsEventAttachmentsToAIRequest(t *testing.T) { + got := make(chan []aiModel.Attachment, 1) + capturingAI := &mockAIAdapter{ + callFn: func(_ context.Context, req *aiModel.A2ARequest) (*aiModel.NormalizedResponse, error) { + got <- req.Attachments + return &aiModel.NormalizedResponse{Content: "ok"}, nil + }, + } + svc, _ := setupSvcWithAI(t, capturingAI) + ctx := context.Background() + // Start from a clean pair: a state left behind by an earlier run would send + // Process down the reset path instead of the AI stage. + clearPair(t, svc, 2180) + + event := &model.MessageEvent{ + ContactID: 2180, ConversationID: 2180, + MessageContent: "olha essa foto", + BotConfig: model.BotConfig{DebounceTime: 0}, // straight to the AI stage + Attachments: []model.Attachment{ + {URL: "http://crm/rails/blob/a.png", ContentType: "image/png", FileType: "image"}, + }, + } + if err := svc.Process(ctx, event); err != nil { + t.Fatalf("Process returned error: %v", err) + } + + select { + case atts := <-got: + if len(atts) != 1 { + t.Fatalf("A2ARequest.Attachments = %+v, want the event's single attachment", atts) + } + if atts[0].URL != "http://crm/rails/blob/a.png" || atts[0].ContentType != "image/png" || atts[0].FileType != "image" { + t.Errorf("attachment reached the adapter as %+v, want the event's values intact", atts[0]) + } + case <-time.After(3 * time.Second): + t.Fatal("AI adapter was never called") + } +} + +// Media aggregates across the debounce window the same way text does: every message +// in the window contributes its attachments to the single AI call. +func TestProcess_AggregatesAttachmentsAcrossDebounceWindow(t *testing.T) { + got := make(chan []aiModel.Attachment, 1) + capturingAI := &mockAIAdapter{ + callFn: func(_ context.Context, req *aiModel.A2ARequest) (*aiModel.NormalizedResponse, error) { + got <- req.Attachments + return &aiModel.NormalizedResponse{Content: "ok"}, nil + }, + } + svc, _ := setupSvcWithAI(t, capturingAI) + ctx := context.Background() + clearPair(t, svc, 2181) + + first := &model.MessageEvent{ + ContactID: 2181, ConversationID: 2181, + MessageContent: "foto 1", + BotConfig: model.BotConfig{DebounceTime: 1}, + Attachments: []model.Attachment{{URL: "http://crm/a.png", ContentType: "image/png", FileType: "image"}}, + } + if err := svc.Process(ctx, first); err != nil { + t.Fatalf("Process(first): %v", err) + } + second := &model.MessageEvent{ + ContactID: 2181, ConversationID: 2181, + MessageContent: "foto 2", + BotConfig: model.BotConfig{DebounceTime: 1}, + Attachments: []model.Attachment{{URL: "http://crm/b.png", ContentType: "image/png", FileType: "image"}}, + } + if err := svc.Process(ctx, second); err != nil { + t.Fatalf("Process(second): %v", err) + } + + // Drop the timer instead of sleeping through the window: advanceToAI bails out + // while the timer key is still alive. + if err := svc.repo.DeleteTimer(ctx, 2181, 2181); err != nil { + t.Fatalf("DeleteTimer: %v", err) + } + svc.advanceToAI(2181, 2181) + + select { + case atts := <-got: + if len(atts) != 2 || atts[0].URL != "http://crm/a.png" || atts[1].URL != "http://crm/b.png" { + t.Fatalf("A2ARequest.Attachments = %+v, want both window attachments in order", atts) + } + case <-time.After(3 * time.Second): + t.Fatal("AI adapter was never called") + } +} + +// clearPair wipes every Redis key of a pair so a test does not inherit state from +// a previous run (the suite only flushes when TEST_REDIS_FLUSH=1). +func clearPair(t *testing.T, svc *pipelineService, id int64) { + t.Helper() + if err := svc.repo.ClearState(context.Background(), id, id); err != nil { + t.Fatalf("ClearState(%d): %v", id, err) + } + t.Cleanup(func() { + _ = svc.repo.ClearState(context.Background(), id, id) + }) +} diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index fc1de7e..0fb2e1c 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -219,8 +219,8 @@ func newHarness(t *testing.T) *harness { rs := redsync.New(pool) repo := repository.NewPipelineRepository(rdb, rs) debounce := debounceService.NewDebounceEngine(repo) - ai := aiService.NewAIAdapter(aiSrv.URL, 10) - dispatch := dispatchService.NewDispatchEngine() + ai := aiService.NewAIAdapter(10, 0, 200) + dispatch := dispatchService.NewDispatchEngine(testSecret) pipeline := pipelineService.NewPipelineService(repo, debounce, ai, dispatch) if err := pipeline.Start(); err != nil { t.Fatalf("pipeline.Start: %v", err) @@ -274,6 +274,7 @@ func (h *harness) event(contactID, convID int64, content string, debounceTime in MessageID: "e2e-msg-1", MessageContent: content, PostbackURL: h.pbServer.URL, + OutgoingURL: h.aiServer.URL, BotConfig: model.BotConfig{DebounceTime: debounceTime}, } } @@ -387,8 +388,8 @@ func TestE2E_AIInterruption(t *testing.T) { contactID, convID := nextPair() firstCallConnected := make(chan struct{}) // closed when mock server has event1's connection - unblockFirst := make(chan struct{}) // closed by test to release the blocking handler - firstCallExited := make(chan struct{}) // closed when the blocking handler goroutine exits + unblockFirst := make(chan struct{}) // closed by test to release the blocking handler + firstCallExited := make(chan struct{}) // closed when the blocking handler goroutine exits // Always unblock the handler on cleanup to prevent httptest.Server.Close() from // waiting indefinitely for the active connection to finish. @@ -532,6 +533,12 @@ func TestE2E_ExactlyOnce_ConcurrentEvents(t *testing.T) { // The panic recovery path is unit-tested but no test has exercised two concurrent // live pipelines where one fails and the other must succeed. func TestE2E_PipelineIsolation(t *testing.T) { + // Pinned through the operator-facing env instead of reaching for the + // package constant: it keeps this test out of service's internals and + // exercises AI_FAILURE_NOTICE on the way. + const pairANotice = "pair-a failure notice" + t.Setenv("AI_FAILURE_NOTICE", pairANotice) + h := newHarness(t) pairAContact, pairAConv := nextPair() pairBContact, pairBConv := nextPair() @@ -555,18 +562,50 @@ func TestE2E_PipelineIsolation(t *testing.T) { h.postEvent(t, h.event(pairAContact, pairAConv, "pair-a", 0)).Body.Close() h.postEvent(t, h.event(pairBContact, pairBConv, "pair-b", 0)).Body.Close() - h.pbServer.waitForCall(t, 3*time.Second) + // CRM-236 changed what pair A does on an AI error: it used to fail SILENTLY + // (log + clear state, nothing reaching the chat), and now it sends the + // customer a failure notice. So this pair delivers too, and the assertion + // "only pair B should deliver" no longer describes intended behaviour. + // + // What this test is actually about — isolation — is now checked more + // strictly than before: each pair must receive ITS OWN message, so a + // crossed delivery fails here even though the call count would be right. + h.pbServer.waitForNCalls(t, 2, 3*time.Second) - if n := h.pbServer.callCount(); n != 1 { - t.Errorf("postback called %d times, want 1 (only pair B should deliver)", n) + if n := h.pbServer.callCount(); n != 2 { + t.Errorf("postback called %d times, want 2 (pair B's answer + pair A's failure notice)", n) + } + + var sawResponse, sawNotice bool + for _, body := range h.pbServer.allBodies() { + switch content := decodePostbackContent(body); content { + case "pair-b response": + sawResponse = true + case pairANotice: + sawNotice = true + default: + t.Errorf("unexpected postback content %q", content) + } + } + if !sawResponse { + t.Error("pair B's answer never arrived: pair A's AI error leaked into pair B") } - if got := decodePostbackContent(h.pbServer.lastBody()); got != "pair-b response" { - t.Errorf("postback content = %q, want %q", got, "pair-b response") + if !sawNotice { + t.Error("pair A got no failure notice: its customer is left in silence (CRM-236)") } - // Pair A must leave no state in Redis after its AI error. + // Pair A must leave no state in Redis after its AI error. Its cleanup runs in + // its own goroutine, so poll instead of racing pair B's postback. stateKey := fmt.Sprintf("bot_runtime:state:%d:%d", pairAContact, pairAConv) - if n := h.rdb.Exists(context.Background(), stateKey).Val(); n != 0 { + cleared := false + for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); { + if h.rdb.Exists(context.Background(), stateKey).Val() == 0 { + cleared = true + break + } + time.Sleep(10 * time.Millisecond) + } + if !cleared { t.Errorf("pair A state key still exists in Redis after AI error — state leak") } } @@ -611,7 +650,8 @@ func TestE2E_DispatchSegmentation(t *testing.T) { t.Fatalf("postback called %d times, want 3 (one per segment)", n) } - want := []string{"hello", "world foo", "bar [sig]"} + // EVO-558: the signature is prepended to the first segment only. + want := []string{" [sig]hello", "world foo", "bar"} for i, body := range h.pbServer.allBodies() { got := decodePostbackContent(body) if got != want[i] { @@ -709,8 +749,8 @@ func TestE2E_RecoveryAfterRestart(t *testing.T) { rs := redsync.New(pool) repo := repository.NewPipelineRepository(rdb, rs) debounce := debounceService.NewDebounceEngine(repo) - ai := aiService.NewAIAdapter(aiSrv.URL, 10) - dispatch := dispatchService.NewDispatchEngine() + ai := aiService.NewAIAdapter(10, 0, 200) + dispatch := dispatchService.NewDispatchEngine(testSecret) aiSrv.setHandler(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json")