From 9151cd54b118a05d0095247401bd3069c59ea1e8 Mon Sep 17 00:00:00 2001 From: Matheus Pastorini Date: Mon, 20 Jul 2026 20:13:27 -0300 Subject: [PATCH 1/2] feat(EVO-2180): forward incoming media to the AI Processor as A2A file parts The bot only ever sent a text part, so images/audio the customer sent never reached the AI (the agent replied "No content to process"). Accept attachments on the inbound event, carry them through the debounce window, download each and send it as a base64 A2A file part. Part of EVO-2178 (image end-to-end). - pipeline/model: MessageEvent.Attachments + Attachment{URL,ContentType,FileType}. - pipeline/repository: AppendAttachments/GetAttachments on a parallel Redis list (bot_runtime:attach:{contact}:{conv}), aggregated like the text buffer and cleared together in ClearState (no stale media leaks into the next turn). - debounce/service: Start/Reset accept attachments; GetAttachments added. - pipeline/service: thread event.Attachments through start/skip/reset/advance -> the A2ARequest (read fresh from Redis at stage launch, like the buffer). - ai/model: A2ARequest.Attachments; JSONRPCPart.File + JSONRPCFile{Name,MimeType,Bytes} (tags match the processor's extract_files_from_message). - ai/service/ai_adapter: download each attachment once (before Marshal, reused across retries) with a 15 MiB cap; base64-encode; append a file part. A download failure is logged and skipped so the text-only message always survives. - tests: adapter forwards a file part with decodable base64 + download-failure sends text only; repo AppendAttachments/GetAttachments roundtrip + ClearState clears the attach key. Full suite green (go build/vet/test ./pkg/... ./internal/...). Note: test/e2e was already incompatible with NewAIAdapter on develop (pre-existing); repo CI is docker-only. --- pkg/ai/model/a2a.go | 34 ++++-- pkg/ai/service/ai_adapter.go | 104 +++++++++++++++- pkg/ai/service/ai_adapter_media_test.go | 112 ++++++++++++++++++ pkg/debounce/service/debounce_engine.go | 19 ++- pkg/debounce/service/debounce_engine_test.go | 16 +-- pkg/pipeline/handler/handler_test.go | 8 +- pkg/pipeline/model/pipeline.go | 32 +++-- .../repository/pipeline_repository.go | 5 + .../repository/redis_pipeline_repository.go | 40 +++++++ .../redis_pipeline_repository_test.go | 49 +++++++- pkg/pipeline/service/pipeline_service.go | 101 ++++++++++------ pkg/pipeline/service/pipeline_service_test.go | 11 +- 12 files changed, 452 insertions(+), 79 deletions(-) create mode 100644 pkg/ai/service/ai_adapter_media_test.go 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 6ef82dc..cc992f2 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -3,6 +3,7 @@ package service import ( "bytes" "context" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -10,6 +11,8 @@ import ( "log/slog" "math/rand" "net/http" + neturl "net/url" + "path" "time" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" @@ -19,6 +22,11 @@ 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 + // 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 @@ -83,6 +91,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), @@ -91,10 +105,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), }, @@ -287,6 +299,90 @@ 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 + } + parts := make([]model.JSONRPCPart, 0, len(req.Attachments)) + for _, att := range req.Attachments { + if att.URL == "" { + continue + } + data, err := a.downloadAttachment(ctx, att.URL) + if err != nil { + slog.Warn("pipeline.ai.attachment.download_failed", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "file_type", att.FileType, + "error", err, + ) + continue + } + mimeType := att.ContentType + if mimeType == "" { + mimeType = "application/octet-stream" + } + 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, + "bytes", len(data), + ) + } + return parts +} + +// downloadAttachment GETs the URL with the adapter's client and a per-download +// timeout, reading at most maxAttachmentBytes. +func (a *aiAdapter) downloadAttachment(ctx context.Context, url string) ([]byte, error) { + dlCtx, cancel := context.WithTimeout(ctx, time.Duration(a.timeoutSecs)*time.Second) + defer cancel() + + httpReq, err := http.NewRequestWithContext(dlCtx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("new_request: %w", err) + } + resp, err := a.client.Do(httpReq) + if err != nil { + return nil, fmt.Errorf("do: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status %d", resp.StatusCode) + } + // +1 so an exactly-at-cap read is distinguishable from an oversize one. + data, err := io.ReadAll(io.LimitReader(resp.Body, maxAttachmentBytes+1)) + if err != nil { + return nil, fmt.Errorf("read: %w", err) + } + if len(data) > maxAttachmentBytes { + return nil, fmt.Errorf("attachment exceeds %d bytes", maxAttachmentBytes) + } + return data, nil +} + +// 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..80f212d --- /dev/null +++ b/pkg/ai/service/ai_adapter_media_test.go @@ -0,0 +1,112 @@ +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" + "net/http" + "net/http/httptest" + "testing" + + 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) { + 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) { + 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) + } +} 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..95838f0 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,41 @@ 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) + } + return r.rdb.RPush(ctx, attachBufferKey(contactID, conversationID), values...).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 +175,10 @@ func bufferKey(contactID, conversationID int64) string { return fmt.Sprintf("bot_runtime:buffer:%d:%d", contactID, conversationID) } +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/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index d347c57..7c10c34 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -33,11 +33,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 +46,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 +169,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 +216,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 +229,13 @@ func (s *pipelineService) skipDebounce(ctx context.Context, event *model.Message return fmt.Errorf("pipeline.skip_debounce.get_buffer: %w", err) } + atts, err := s.debounce.GetAttachments(ctx, event.ContactID, event.ConversationID) + if err != nil { + cancel() + s.entries.Delete(key) + return fmt.Errorf("pipeline.skip_debounce.get_attachments: %w", err) + } + 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 +247,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 +296,16 @@ func (s *pipelineService) advanceToAI(contactID, conversationID int64) { return } + atts, err := s.debounce.GetAttachments(ctx, contactID, conversationID) + if err != nil { + slog.Error("pipeline.debounce.get_attachments_failed", + "contact_id", contactID, + "conversation_id", conversationID, + "error", err, + ) + return + } + 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 +321,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 +346,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 +372,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 +390,7 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio ApiKey: apiKey, Message: buffer, Metadata: metadata, + Attachments: aiAttachments, }) if err != nil { switch { @@ -417,12 +446,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 +459,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 +484,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 +504,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 +514,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(), ) } diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index f500faf..3ca78d4 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) { @@ -709,4 +711,3 @@ func TestPipeline_DispatchError_ClearsState(t *testing.T) { t.Errorf("state must be cleared after dispatch error, got exists=%d", exists) } } - From bf97f9be1c352a7b7975e68c84c186747afc8740 Mon Sep 17 00:00:00 2001 From: Guilherme Gomes Date: Tue, 21 Jul 2026 11:58:23 -0300 Subject: [PATCH 2/2] fix(EVO-2180): bound media forwarding and validate what is forwarded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the incoming-media path. The per-file cap was the only bound, so the failure modes it did not cover fell back on the customer losing the whole reply instead of just the media. - Shared byte budget (20 MiB) across every attachment of the call. The debounce window aggregates the media of all its messages, so a photo burst built a body of len(attachments) x 15 MiB; base64 pushed that past the gateway's client_max_body_size and the resulting 413 is not retryable, killing the text reply too. Probe: 20 x 2 MiB went from a 53 MiB request to 26 MiB. - Dedicated download timeouts. Downloads run before the AI call and outside its retry ceiling, but reused AI_CALL_TIMEOUT_SECONDS (30s) per attachment, so an unreachable media host stalled the turn by 30s x len(attachments) with no bound. Now 10s per download and 30s for the whole set. - Resolve the mime type from the bytes in hand: the response Content-Type wins, then the CRM's declared type, then the URL extension. The processor feeds this straight into Blob(mime_type=...), so an HTML error/login page answered with 200 was being forwarded as a valid image, and a missing content_type became application/octet-stream. Both are now dropped or resolved. - A Redis failure on the attachment buffer no longer aborts the turn: media is best-effort everywhere else in this path, and dropping the text reply over it contradicted the card's own acceptance criterion. Tests: the event -> debounce -> Redis -> A2ARequest seam had no coverage (the debounce mock always returned nil attachments), so a refactor could silently drop the media; two pipeline tests now pin it, including aggregation across the debounce window. Adapter tests cover the byte budget, the time budget, HTML responses, oversize files and the mime resolution table. Also repairs test/e2e, which has not compiled since EVO-2167 changed NewAIAdapter/NewDispatchEngine — which is why `go vet ./...` and `go test ./...` could not be run at all. Two assertions had drifted: the message signature moved to a prefix on the first segment in EVO-558, and the state-leak check raced the cleanup goroutine it was asserting on. go build ./... && go vet ./... && go test ./... green, e2e included. --- pkg/ai/service/ai_adapter.go | 158 ++++++++++++-- pkg/ai/service/ai_adapter_media_test.go | 193 ++++++++++++++++++ pkg/pipeline/service/pipeline_service.go | 17 +- pkg/pipeline/service/pipeline_service_test.go | 105 ++++++++++ test/e2e/e2e_test.go | 29 ++- 5 files changed, 471 insertions(+), 31 deletions(-) diff --git a/pkg/ai/service/ai_adapter.go b/pkg/ai/service/ai_adapter.go index cc992f2..65b5b6d 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -10,9 +10,11 @@ import ( "io" "log/slog" "math/rand" + "mime" "net/http" neturl "net/url" "path" + "strings" "time" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" @@ -27,6 +29,26 @@ const maxResponseBytes = 1 << 20 // 1 MiB // 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 +) + // 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 @@ -307,12 +329,38 @@ func (a *aiAdapter) buildFileParts(ctx context.Context, req *model.A2ARequest) [ if len(req.Attachments) == 0 { return nil } + perDownload := a.attachmentTimeout() + budgetCtx, cancelBudget := context.WithTimeout(ctx, attachmentsTotalTimeFactor*perDownload) + defer cancelBudget() + parts := make([]model.JSONRPCPart, 0, len(req.Attachments)) - for _, att := range req.Attachments { + remaining := maxAttachmentsTotalBytes + for i, att := range req.Attachments { if att.URL == "" { continue } - data, err := a.downloadAttachment(ctx, att.URL) + 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 := a.downloadAttachment(budgetCtx, att.URL, perDownload, limit) if err != nil { slog.Warn("pipeline.ai.attachment.download_failed", "contact_id", req.ContactID, @@ -322,10 +370,18 @@ func (a *aiAdapter) buildFileParts(ctx context.Context, req *model.A2ARequest) [ ) continue } - mimeType := att.ContentType - if mimeType == "" { - mimeType = "application/octet-stream" + 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{ @@ -338,39 +394,107 @@ func (a *aiAdapter) buildFileParts(ctx context.Context, req *model.A2ARequest) [ "contact_id", req.ContactID, "conversation_id", req.ConversationID, "file_type", att.FileType, + "mime_type", mimeType, "bytes", len(data), ) } return parts } -// downloadAttachment GETs the URL with the adapter's client and a per-download -// timeout, reading at most maxAttachmentBytes. -func (a *aiAdapter) downloadAttachment(ctx context.Context, url string) ([]byte, error) { - dlCtx, cancel := context.WithTimeout(ctx, time.Duration(a.timeoutSecs)*time.Second) +// 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 +} + +// downloadAttachment GETs the URL with the adapter's client and the given timeout, +// reading at most limit bytes. It returns the body and the response Content-Type so +// the caller can decide what the bytes actually are. +func (a *aiAdapter) downloadAttachment(ctx context.Context, 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) + return nil, "", fmt.Errorf("new_request: %w", err) } resp, err := a.client.Do(httpReq) if err != nil { - return nil, fmt.Errorf("do: %w", err) + return nil, "", fmt.Errorf("do: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("unexpected status %d", resp.StatusCode) + return nil, "", fmt.Errorf("unexpected status %d", resp.StatusCode) } // +1 so an exactly-at-cap read is distinguishable from an oversize one. - data, err := io.ReadAll(io.LimitReader(resp.Body, maxAttachmentBytes+1)) + data, err := io.ReadAll(io.LimitReader(resp.Body, int64(limit)+1)) if err != nil { - return nil, fmt.Errorf("read: %w", err) + return nil, "", fmt.Errorf("read: %w", err) } - if len(data) > maxAttachmentBytes { - return nil, fmt.Errorf("attachment exceeds %d bytes", maxAttachmentBytes) + 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 data, nil + return mime.TypeByExtension(path.Ext(u.Path)) } // attachmentName derives a filename from the URL path (fallback "file"). diff --git a/pkg/ai/service/ai_adapter_media_test.go b/pkg/ai/service/ai_adapter_media_test.go index 80f212d..def62d5 100644 --- a/pkg/ai/service/ai_adapter_media_test.go +++ b/pkg/ai/service/ai_adapter_media_test.go @@ -8,9 +8,12 @@ import ( "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" @@ -110,3 +113,193 @@ func TestCall_AttachmentDownloadFailure_SendsTextOnly(t *testing.T) { 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) { + 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) { + 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) { + 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) { + 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) { + 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/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index 7c10c34..63c610a 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -229,11 +229,16 @@ 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 { - cancel() - s.entries.Delete(key) - return fmt.Errorf("pipeline.skip_debounce.get_attachments: %w", err) + 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()} @@ -296,14 +301,16 @@ 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.Error("pipeline.debounce.get_attachments_failed", + slog.Warn("pipeline.debounce.get_attachments_failed", "contact_id", contactID, "conversation_id", conversationID, "error", err, ) - return + atts = nil } newState := &model.PipelineState{Stage: model.StageAI, CreatedAt: time.Now()} diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index 3ca78d4..a19197f 100644 --- a/pkg/pipeline/service/pipeline_service_test.go +++ b/pkg/pipeline/service/pipeline_service_test.go @@ -711,3 +711,108 @@ func TestPipeline_DispatchError_ClearsState(t *testing.T) { t.Errorf("state must be cleared after dispatch error, got exists=%d", exists) } } + +// 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..2aa7eec 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. @@ -564,9 +565,18 @@ func TestE2E_PipelineIsolation(t *testing.T) { t.Errorf("postback content = %q, want %q", got, "pair-b response") } - // 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 +621,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 +720,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")