From 0e8c5eb27149ce7caac33bf0acc129004e4aa561 Mon Sep 17 00:00:00 2001 From: Murillo Alves Date: Wed, 30 Sep 2026 11:11:05 -0300 Subject: [PATCH] feat: propagate audio metadata and idempotency --- pkg/ai/model/a2a.go | 6 +- pkg/ai/service/ai_adapter.go | 14 +- pkg/ai/service/ai_adapter_retry_test.go | 48 +++++ pkg/ai/service/ai_adapter_test.go | 4 + pkg/debounce/service/debounce_engine.go | 14 +- pkg/debounce/service/debounce_engine_test.go | 16 +- pkg/pipeline/handler/handler_test.go | 24 +++ pkg/pipeline/model/pipeline.go | 25 ++- pkg/pipeline/model/pipeline_test.go | 28 +++ .../repository/pipeline_repository.go | 2 + .../repository/redis_pipeline_repository.go | 26 +++ .../redis_pipeline_repository_test.go | 27 +++ pkg/pipeline/service/pipeline_service.go | 185 +++++++++++++----- pkg/pipeline/service/pipeline_service_test.go | 98 +++++++++- 14 files changed, 440 insertions(+), 77 deletions(-) create mode 100644 pkg/pipeline/model/pipeline_test.go diff --git a/pkg/ai/model/a2a.go b/pkg/ai/model/a2a.go index 5e6de87..5354577 100644 --- a/pkg/ai/model/a2a.go +++ b/pkg/ai/model/a2a.go @@ -6,6 +6,7 @@ type A2ARequest struct { OutgoingURL string // full A2A endpoint URL from agent_bot.outgoing_url ContactID int64 // used for userId in JSON-RPC params ConversationID int64 // used for contextId in JSON-RPC params + MessageID string // stable CRM message ID used for retry idempotency 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) @@ -36,8 +37,9 @@ type JSONRPCParams struct { } type JSONRPCMessage struct { - Role string `json:"role"` - Parts []JSONRPCPart `json:"parts"` + Role string `json:"role"` + Parts []JSONRPCPart `json:"parts"` + MessageID string `json:"messageId,omitempty"` } type JSONRPCPart struct { diff --git a/pkg/ai/service/ai_adapter.go b/pkg/ai/service/ai_adapter.go index a144be5..1c25911 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -134,8 +134,9 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor ContextID: contextID, UserID: userID, Message: model.JSONRPCMessage{ - Role: "user", - Parts: parts, + Role: "user", + Parts: parts, + MessageID: req.MessageID, }, Metadata: nonNilMetadata(req.Metadata), }, @@ -243,9 +244,12 @@ func (a *aiAdapter) doOnce( defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - retryable := isRetryableStatus(resp.StatusCode) - // Drain a bounded amount so the connection can be reused. - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) + // A duplicate message may meet another still-running processor request. + // Retry only that explicit 409; payload conflicts and expired claims remain + // permanent so the runtime cannot accidentally repeat side effects. + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + retryable := isRetryableStatus(resp.StatusCode) || + (resp.StatusCode == http.StatusConflict && bytes.Contains(body, []byte("Request already in progress"))) return nil, retryable, fmt.Errorf("pipeline.ai.status: unexpected %d from AI Processor", resp.StatusCode) } diff --git a/pkg/ai/service/ai_adapter_retry_test.go b/pkg/ai/service/ai_adapter_retry_test.go index 3b0ca28..0556dd9 100644 --- a/pkg/ai/service/ai_adapter_retry_test.go +++ b/pkg/ai/service/ai_adapter_retry_test.go @@ -36,6 +36,54 @@ func retryReq(url string) *aiModel.A2ARequest { ContactID: 42, ConversationID: 7, ApiKey: "k", + MessageID: "crm-message-123", + } +} + +func TestCall_RetriesOnlyActiveIdempotencyConflict(t *testing.T) { + var calls int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request aiModel.JSONRPCRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Errorf("decode request: %v", err) + } + if request.Params.Message.MessageID != "crm-message-123" { + t.Errorf("messageId = %q, want stable CRM message ID", request.Params.Message.MessageID) + } + if atomic.AddInt32(&calls, 1) == 1 { + w.WriteHeader(http.StatusConflict) + _, _ = w.Write([]byte(`{"error":{"message":"The same message is currently being processed","details":{"error":{"message":"Request already in progress"}}}}`)) + return + } + writeOK(w) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 2, 1) + response, err := adapter.Call(context.Background(), retryReq(server.URL)) + if err != nil { + t.Fatalf("expected success after active-claim retry, got %v", err) + } + if response.Content != "ok" || atomic.LoadInt32(&calls) != 2 { + t.Fatalf("response=%+v calls=%d, want ok and exactly two attempts", response, calls) + } +} + +func TestCall_DoesNotRetryIdempotencyPayloadConflict(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.StatusConflict) + _, _ = w.Write([]byte(`{"error":{"message":"Idempotency key payload mismatch"}}`)) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30, 3, 1) + if _, err := adapter.Call(context.Background(), retryReq(server.URL)); err == nil { + t.Fatal("expected permanent idempotency payload conflict") + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("calls = %d, want 1 for permanent conflict", got) } } diff --git a/pkg/ai/service/ai_adapter_test.go b/pkg/ai/service/ai_adapter_test.go index 2b12e42..3308754 100644 --- a/pkg/ai/service/ai_adapter_test.go +++ b/pkg/ai/service/ai_adapter_test.go @@ -67,6 +67,9 @@ func TestCall_Success(t *testing.T) { if req.Params.UserID != "contact-uuid-xyz" { t.Errorf("req.Params.UserID = %q, want %q", req.Params.UserID, "contact-uuid-xyz") } + if req.Params.Message.MessageID != "crm-message-123" { + t.Errorf("req.Params.Message.MessageID = %q, want %q", req.Params.Message.MessageID, "crm-message-123") + } if len(req.Params.Message.Parts) != 1 || req.Params.Message.Parts[0].Text != "hello world" { t.Errorf("message parts = %+v, want single part with text 'hello world'", req.Params.Message.Parts) } @@ -92,6 +95,7 @@ func TestCall_Success(t *testing.T) { ContactID: 42, ConversationID: 7, ApiKey: "test-key", + MessageID: "crm-message-123", Metadata: crmMetadata("conv-uuid-abc", "contact-uuid-xyz"), }) if err != nil { diff --git a/pkg/debounce/service/debounce_engine.go b/pkg/debounce/service/debounce_engine.go index 1ab9fa2..03f53ea 100644 --- a/pkg/debounce/service/debounce_engine.go +++ b/pkg/debounce/service/debounce_engine.go @@ -12,8 +12,8 @@ import ( // DebounceEngine manages per-pair debounce timers and message buffers. type DebounceEngine interface { - 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 + Start(ctx context.Context, contactID, conversationID int64, content, messageID string, atts []model.Attachment, cfg model.BotConfig) error + Reset(ctx context.Context, contactID, conversationID int64, content, messageID 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) @@ -28,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, atts []model.Attachment, cfg model.BotConfig) error { +func (d *debounceEngine) Start(ctx context.Context, contactID, conversationID int64, content, messageID 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.AppendMessageID(ctx, contactID, conversationID, messageID); err != nil { + return fmt.Errorf("debounce.start.append_message_id: %w", err) + } if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil { return fmt.Errorf("debounce.start.append_attachments: %w", err) } @@ -44,10 +47,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, atts []model.Attachment, cfg model.BotConfig) error { +func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID int64, content, messageID 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.AppendMessageID(ctx, contactID, conversationID, messageID); err != nil { + return fmt.Errorf("debounce.reset.append_message_id: %w", err) + } if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil { return fmt.Errorf("debounce.reset.append_attachments: %w", err) } diff --git a/pkg/debounce/service/debounce_engine_test.go b/pkg/debounce/service/debounce_engine_test.go index 162ada8..6b12281 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", nil, cfg); err != nil { + if err := eng.Start(ctx, 1, 1, "first message", "msg-1", 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", nil, cfg); err != nil { + if err := eng.Start(ctx, 2, 2, "immediate", "msg-2", 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", nil, cfg); err != nil { + if err := eng.Start(ctx, 3, 3, "msg1", "msg-3a", 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", nil, cfg); err != nil { + if err := eng.Reset(ctx, 3, 3, "msg2", "msg-3b", 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", nil, cfg) - _ = eng.Reset(ctx, 4, 4, "world", nil, cfg) - _ = eng.Reset(ctx, 4, 4, "again", nil, cfg) + _ = eng.Start(ctx, 4, 4, "hello", "msg-4a", nil, cfg) + _ = eng.Reset(ctx, 4, 4, "world", "msg-4b", nil, cfg) + _ = eng.Reset(ctx, 4, 4, "again", "msg-4c", 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", nil, cfg); err != nil { + if err := eng.Start(ctx, 5, 5, "msg", "msg-5", 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 eb4f75a..3d468c5 100644 --- a/pkg/pipeline/handler/handler_test.go +++ b/pkg/pipeline/handler/handler_test.go @@ -39,6 +39,10 @@ func (m *mockRepo) AppendToBuffer(_ context.Context, _, _ int64, _ string) error func (m *mockRepo) GetBuffer(_ context.Context, _, _ int64) ([]string, error) { return nil, nil } +func (m *mockRepo) AppendMessageID(_ context.Context, _, _ int64, _ string) error { return nil } +func (m *mockRepo) GetMessageIDs(_ context.Context, _, _ int64) ([]string, error) { + return nil, nil +} func (m *mockRepo) AppendAttachments(_ context.Context, _, _ int64, _ []model.Attachment) error { return nil } @@ -147,6 +151,26 @@ func TestHandleEvent_500_OnSetStateError(t *testing.T) { } } +func TestHandleEvent_400_WhenMessageIDMissing(t *testing.T) { + payload := validPayload() + var event map[string]any + if err := json.Unmarshal(payload, &event); err != nil { + t.Fatal(err) + } + delete(event, "message_id") + payload, _ = json.Marshal(event) + r := setupRouter(&mockRepo{}, &mockSvc{}) + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(payload)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bot-Runtime-Secret", testSecret) + r.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d", w.Code, http.StatusBadRequest) + } +} + func TestHandleEvent_202_EvenWhenProcessFails(t *testing.T) { // Process runs in a goroutine — its errors must not affect the HTTP response. r := setupRouter(&mockRepo{}, &mockSvc{processErr: errors.New("process error")}) diff --git a/pkg/pipeline/model/pipeline.go b/pkg/pipeline/model/pipeline.go index cf39048..ca6bbc7 100644 --- a/pkg/pipeline/model/pipeline.go +++ b/pkg/pipeline/model/pipeline.go @@ -27,14 +27,16 @@ 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"` + MessageID string `json:"message_id,omitempty"` + ResponseNotice string `json:"response_notice,omitempty"` } // MessageEvent is the inbound payload from evo-ai-crm AgentBotListener. @@ -45,6 +47,7 @@ type MessageEvent struct { ContactID int64 `json:"contact_id"` MessageID string `json:"message_id"` MessageContent string `json:"message_content"` + ResponseNotice string `json:"response_notice,omitempty"` Attachments []Attachment `json:"attachments,omitempty"` // EVO-2180: incoming media (image/audio/…) ApiKey string `json:"api_key"` OutgoingURL string `json:"outgoing_url"` @@ -93,5 +96,11 @@ func (e *MessageEvent) Validate() error { if e.BotConfig.DelayPerCharacter < 0 { return errors.New("delay_per_character must be >= 0") } + if e.MessageID == "" { + return errors.New("message_id is required for idempotent processing") + } + if len(e.ResponseNotice) > 512 { + return errors.New("response_notice must not exceed 512 bytes") + } return nil } diff --git a/pkg/pipeline/model/pipeline_test.go b/pkg/pipeline/model/pipeline_test.go new file mode 100644 index 0000000..b224956 --- /dev/null +++ b/pkg/pipeline/model/pipeline_test.go @@ -0,0 +1,28 @@ +package model + +import ( + "strings" + "testing" +) + +func TestMessageEventValidateResponseNoticeLimit(t *testing.T) { + event := validMessageEvent() + event.ResponseNotice = strings.Repeat("a", 512) + if err := event.Validate(); err != nil { + t.Fatalf("512-byte response notice should be accepted: %v", err) + } + + event.ResponseNotice += "a" + if err := event.Validate(); err == nil { + t.Fatal("513-byte response notice should be rejected") + } +} + +func validMessageEvent() *MessageEvent { + return &MessageEvent{ + ContactID: 1, + ConversationID: 1, + MessageID: "message-1", + PostbackURL: "https://example.invalid/postback", + } +} diff --git a/pkg/pipeline/repository/pipeline_repository.go b/pkg/pipeline/repository/pipeline_repository.go index 97670da..1371aaf 100644 --- a/pkg/pipeline/repository/pipeline_repository.go +++ b/pkg/pipeline/repository/pipeline_repository.go @@ -21,6 +21,8 @@ type PipelineRepository interface { AppendToBuffer(ctx context.Context, contactID, conversationID int64, content string) error GetBuffer(ctx context.Context, contactID, conversationID int64) ([]string, error) + AppendMessageID(ctx context.Context, contactID, conversationID int64, messageID string) error + GetMessageIDs(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. diff --git a/pkg/pipeline/repository/redis_pipeline_repository.go b/pkg/pipeline/repository/redis_pipeline_repository.go index 48ca9d5..56d4ade 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository.go +++ b/pkg/pipeline/repository/redis_pipeline_repository.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "log/slog" + "sort" "strconv" "strings" "time" @@ -54,6 +55,7 @@ func (r *redisPipelineRepository) ClearState(ctx context.Context, contactID, con keys := []string{ stateKey(contactID, conversationID), bufferKey(contactID, conversationID), + messageIDsKey(contactID, conversationID), attachBufferKey(contactID, conversationID), // EVO-2180: clear media with the text buffer timerKey(contactID, conversationID), } @@ -68,6 +70,26 @@ func (r *redisPipelineRepository) GetBuffer(ctx context.Context, contactID, conv return r.rdb.LRange(ctx, bufferKey(contactID, conversationID), 0, -1).Result() } +func (r *redisPipelineRepository) AppendMessageID(ctx context.Context, contactID, conversationID int64, messageID string) error { + if messageID == "" { + return nil + } + key := messageIDsKey(contactID, conversationID) + if err := r.rdb.SAdd(ctx, key, messageID).Err(); err != nil { + return err + } + return r.rdb.Expire(ctx, key, attachBufferTTL).Err() +} + +func (r *redisPipelineRepository) GetMessageIDs(ctx context.Context, contactID, conversationID int64) ([]string, error) { + ids, err := r.rdb.SMembers(ctx, messageIDsKey(contactID, conversationID)).Result() + if err != nil { + return nil, err + } + sort.Strings(ids) + return ids, nil +} + // 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 { @@ -180,6 +202,10 @@ func bufferKey(contactID, conversationID int64) string { return fmt.Sprintf("bot_runtime:buffer:%d:%d", contactID, conversationID) } +func messageIDsKey(contactID, conversationID int64) string { + return fmt.Sprintf("bot_runtime:message_ids:%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 diff --git a/pkg/pipeline/repository/redis_pipeline_repository_test.go b/pkg/pipeline/repository/redis_pipeline_repository_test.go index 13f9460..5c19968 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository_test.go +++ b/pkg/pipeline/repository/redis_pipeline_repository_test.go @@ -123,6 +123,33 @@ func TestAppendToBuffer_PreservesOrder(t *testing.T) { } } +func TestMessageIDs_AreUniqueSortedAndClearedWithPipelineState(t *testing.T) { + repo, cleanup := setupTestRepo(t) + defer cleanup() + + ctx := context.Background() + const contactID, conversationID int64 = 9871, 9872 + for _, id := range []string{"message-b", "message-a", "message-b"} { + if err := repo.AppendMessageID(ctx, contactID, conversationID, id); err != nil { + t.Fatalf("AppendMessageID(%q): %v", id, err) + } + } + got, err := repo.GetMessageIDs(ctx, contactID, conversationID) + if err != nil { + t.Fatalf("GetMessageIDs: %v", err) + } + if len(got) != 2 || got[0] != "message-a" || got[1] != "message-b" { + t.Fatalf("GetMessageIDs = %v, want unique sorted IDs", got) + } + if err := repo.ClearState(ctx, contactID, conversationID); err != nil { + t.Fatalf("ClearState: %v", err) + } + got, err = repo.GetMessageIDs(ctx, contactID, conversationID) + if err != nil || len(got) != 0 { + t.Fatalf("GetMessageIDs after ClearState = %v (err %v), want empty", got, err) + } +} + func TestGetBuffer_EmptyKey_ReturnsEmpty(t *testing.T) { repo, cleanup := setupTestRepo(t) defer cleanup() diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index c2983d2..9a449f8 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -2,6 +2,8 @@ package service import ( "context" + "crypto/sha256" + "encoding/hex" "errors" "fmt" "log/slog" @@ -32,13 +34,42 @@ 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) + 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) + responseNotice string // deterministic CRM notice appended after the AI response +} + +func appendResponseNotice(content, notice string) string { + notice = strings.TrimSpace(notice) + content = strings.TrimSpace(content) + if notice == "" || strings.Contains(content, notice) { + return content + } + if content == "" { + return notice + } + return content + "\n\n" + notice +} + +func stableMessageID(messageIDs []string) string { + hash := sha256.New() + for _, id := range messageIDs { + if id == "" { + continue + } + _, _ = hash.Write([]byte(id)) + _, _ = hash.Write([]byte{0}) + } + return "debounce:" + hex.EncodeToString(hash.Sum(nil)) +} + +func legacyDebounceMessageID(pair model.PairID, createdAt time.Time) string { + return stableMessageID([]string{fmt.Sprintf("legacy:%d:%d:%d", pair.ContactID, pair.ConversationID, createdAt.UnixNano())}) } type pipelineService struct { @@ -86,19 +117,31 @@ func (s *pipelineService) Start() error { if err != nil || state == nil || state.Stage != model.StageDebounce { continue } + messageIDs, idsErr := s.repo.GetMessageIDs(ctx, pair.ContactID, pair.ConversationID) + if idsErr != nil { + return fmt.Errorf("pipeline.start.get_message_ids for pair %s: %w", pairKey(pair.ContactID, pair.ConversationID), idsErr) + } + if len(messageIDs) > 0 { + state.MessageID = stableMessageID(messageIDs) + } else if state.MessageID == "" { + // Backward compatibility for debounce state written before CRM message + // IDs were persisted by the runtime. + state.MessageID = legacyDebounceMessageID(pair, state.CreatedAt) + } // Recreate entry only if not already in memory (avoids overwriting cancel funcs // for pairs that received a Process call concurrently with Start recovery). key := pairKey(pair.ContactID, pair.ConversationID) if _, alreadyExists := s.entries.Load(key); !alreadyExists { pipelineCtx, cancel := context.WithCancel(context.Background()) s.entries.Store(key, pipelineEntry{ - ctx: pipelineCtx, - cancel: cancel, - cfg: state.BotConfig, - postbackURL: state.PostbackURL, - outgoingURL: state.OutgoingURL, - apiKey: state.ApiKey, - metadata: state.Metadata, + ctx: pipelineCtx, + cancel: cancel, + cfg: state.BotConfig, + postbackURL: state.PostbackURL, + outgoingURL: state.OutgoingURL, + apiKey: state.ApiKey, + metadata: state.Metadata, + responseNotice: state.ResponseNotice, }) } @@ -139,7 +182,7 @@ func (s *pipelineService) Process(ctx context.Context, event *model.MessageEvent } return s.startDebounce(ctx, event) case state.Stage == model.StageDebounce: - return s.resetDebounce(ctx, event) + return s.resetDebounce(ctx, event, state) case state.Stage == model.StageAI || state.Stage == model.StageDispatch: s.cancelPair(event.ContactID, event.ConversationID) if err := s.repo.ClearState(ctx, event.ContactID, event.ConversationID); err != nil { @@ -161,16 +204,17 @@ func (s *pipelineService) startDebounce(ctx context.Context, event *model.Messag pipelineCtx, cancel := context.WithCancel(context.Background()) key := pairKey(event.ContactID, event.ConversationID) s.entries.Store(key, pipelineEntry{ - ctx: pipelineCtx, - cancel: cancel, - cfg: event.BotConfig, - postbackURL: event.PostbackURL, - outgoingURL: event.OutgoingURL, - apiKey: event.ApiKey, - metadata: event.Metadata, + ctx: pipelineCtx, + cancel: cancel, + cfg: event.BotConfig, + postbackURL: event.PostbackURL, + outgoingURL: event.OutgoingURL, + apiKey: event.ApiKey, + metadata: event.Metadata, + responseNotice: strings.TrimSpace(event.ResponseNotice), }) - if err := s.debounce.Start(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.Attachments, event.BotConfig); err != nil { + if err := s.debounce.Start(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.MessageID, event.Attachments, event.BotConfig); err != nil { cancel() s.entries.Delete(key) return fmt.Errorf("pipeline.debounce.start: %w", err) @@ -179,13 +223,15 @@ func (s *pipelineService) startDebounce(ctx context.Context, event *model.Messag // Persist BotConfig, PostbackURL, AgentBotID and ApiKey alongside the stage so that // Start() recovery after a restart can reconstruct the pipelineEntry correctly (NFR-01). newState := &model.PipelineState{ - Stage: model.StageDebounce, - CreatedAt: time.Now(), - BotConfig: event.BotConfig, - PostbackURL: event.PostbackURL, - OutgoingURL: event.OutgoingURL, - ApiKey: event.ApiKey, - Metadata: event.Metadata, + Stage: model.StageDebounce, + CreatedAt: time.Now(), + BotConfig: event.BotConfig, + PostbackURL: event.PostbackURL, + OutgoingURL: event.OutgoingURL, + ApiKey: event.ApiKey, + Metadata: event.Metadata, + MessageID: event.MessageID, + ResponseNotice: strings.TrimSpace(event.ResponseNotice), } if err := s.repo.SetState(ctx, event.ContactID, event.ConversationID, newState); err != nil { cancel() @@ -206,18 +252,19 @@ func (s *pipelineService) skipDebounce(ctx context.Context, event *model.Message pipelineCtx, cancel := context.WithCancel(context.Background()) key := pairKey(event.ContactID, event.ConversationID) s.entries.Store(key, pipelineEntry{ - ctx: pipelineCtx, - cancel: cancel, - cfg: event.BotConfig, - postbackURL: event.PostbackURL, - outgoingURL: event.OutgoingURL, - apiKey: event.ApiKey, - metadata: event.Metadata, + ctx: pipelineCtx, + cancel: cancel, + cfg: event.BotConfig, + postbackURL: event.PostbackURL, + outgoingURL: event.OutgoingURL, + apiKey: event.ApiKey, + metadata: event.Metadata, + responseNotice: strings.TrimSpace(event.ResponseNotice), }) // 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.Attachments, event.BotConfig); err != nil { + event.MessageContent, event.MessageID, event.Attachments, event.BotConfig); err != nil { cancel() s.entries.Delete(key) return fmt.Errorf("pipeline.skip_debounce.start: %w", err) @@ -242,7 +289,12 @@ func (s *pipelineService) skipDebounce(ctx context.Context, event *model.Message atts = nil } - newState := &model.PipelineState{Stage: model.StageAI, CreatedAt: time.Now()} + newState := &model.PipelineState{ + Stage: model.StageAI, + CreatedAt: time.Now(), + MessageID: event.MessageID, + ResponseNotice: strings.TrimSpace(event.ResponseNotice), + } if err := s.repo.SetState(ctx, event.ContactID, event.ConversationID, newState); err != nil { cancel() s.entries.Delete(key) @@ -253,14 +305,25 @@ 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, atts) + s.launchAIStage(event.ContactID, event.ConversationID, buffer, atts, event.MessageID) 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.Attachments, event.BotConfig); err != nil { +func (s *pipelineService) resetDebounce(ctx context.Context, event *model.MessageEvent, state *model.PipelineState) error { + if err := s.debounce.Reset(ctx, event.ContactID, event.ConversationID, event.MessageContent, event.MessageID, event.Attachments, event.BotConfig); err != nil { return fmt.Errorf("pipeline.debounce.reset: %w", err) } + state.ResponseNotice = appendResponseNotice(state.ResponseNotice, event.ResponseNotice) + if err := s.repo.SetState(ctx, event.ContactID, event.ConversationID, state); err != nil { + return fmt.Errorf("pipeline.debounce.update_response_notice: %w", err) + } + key := pairKey(event.ContactID, event.ConversationID) + if value, ok := s.entries.Load(key); ok { + if entry, valid := value.(pipelineEntry); valid { + entry.responseNotice = state.ResponseNotice + s.entries.Store(key, entry) + } + } slog.Info("pipeline.debounce.reset", "contact_id", event.ContactID, "conversation_id", event.ConversationID, @@ -313,8 +376,23 @@ func (s *pipelineService) advanceToAI(contactID, conversationID int64) { ) atts = nil } + messageIDs, err := s.repo.GetMessageIDs(ctx, contactID, conversationID) + if err != nil { + slog.Error("pipeline.debounce.get_message_ids_failed", "contact_id", contactID, "conversation_id", conversationID, "error", err) + return + } + if len(messageIDs) > 0 { + state.MessageID = stableMessageID(messageIDs) + } else if state.MessageID == "" { + state.MessageID = legacyDebounceMessageID(model.PairID{ContactID: contactID, ConversationID: conversationID}, state.CreatedAt) + } - newState := &model.PipelineState{Stage: model.StageAI, CreatedAt: time.Now()} + newState := &model.PipelineState{ + Stage: model.StageAI, + CreatedAt: time.Now(), + MessageID: state.MessageID, + ResponseNotice: state.ResponseNotice, + } if err := s.repo.SetState(ctx, contactID, conversationID, newState); err != nil { slog.Error("pipeline.debounce.set_ai_state_failed", "contact_id", contactID, @@ -329,13 +407,13 @@ func (s *pipelineService) advanceToAI(contactID, conversationID int64) { "conversation_id", conversationID, "buffer_len", len(buffer), ) - s.launchAIStage(contactID, conversationID, buffer, atts) + s.launchAIStage(contactID, conversationID, buffer, atts, state.MessageID) } // 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, atts []model.Attachment) { +func (s *pipelineService) launchAIStage(contactID, conversationID int64, buffer string, atts []model.Attachment, messageID string) { key := pairKey(contactID, conversationID) v, ok := s.entries.Load(key) if !ok { @@ -354,12 +432,12 @@ func (s *pipelineService) launchAIStage(contactID, conversationID int64, buffer ) return } - go s.runAIStage(entry.ctx, contactID, conversationID, buffer, atts, entry.cfg, entry.postbackURL) + go s.runAIStage(entry.ctx, contactID, conversationID, buffer, atts, entry.cfg, entry.postbackURL, messageID) } // 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, atts []model.Attachment, cfg model.BotConfig, postbackURL string) { +func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversationID int64, buffer string, atts []model.Attachment, cfg model.BotConfig, postbackURL, messageID string) { defer s.recoverPipeline(contactID, conversationID) slog.Info("pipeline.ai.started", @@ -372,14 +450,15 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio key := pairKey(contactID, conversationID) var outgoingURL, apiKey string var metadata map[string]any + var responseNotice string if v, ok := s.entries.Load(key); ok { if entry, ok := v.(pipelineEntry); ok { outgoingURL = entry.outgoingURL apiKey = entry.apiKey metadata = entry.metadata + responseNotice = entry.responseNotice } } - // 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)) @@ -395,6 +474,7 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio OutgoingURL: outgoingURL, ContactID: contactID, ConversationID: conversationID, + MessageID: messageID, ApiKey: apiKey, Message: buffer, Metadata: metadata, @@ -431,6 +511,15 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio } return } + if resp == nil { + slog.Error("pipeline.ai.empty_response", + "contact_id", contactID, + "conversation_id", conversationID, + ) + s.clearStateWithLog(contactID, conversationID) + return + } + resp.Content = appendResponseNotice(resp.Content, responseNotice) // Success path — use cleanupCtx() for Redis calls: pipeline ctx may be cancelled. dur := time.Since(start) diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index a19197f..106e70c 100644 --- a/pkg/pipeline/service/pipeline_service_test.go +++ b/pkg/pipeline/service/pipeline_service_test.go @@ -61,11 +61,11 @@ type mockDebounce struct { 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 { +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.Attachment, _ model.BotConfig) error { +func (m *mockDebounce) Reset(_ context.Context, _, _ int64, _, _ string, _ []model.Attachment, _ model.BotConfig) error { m.resetCalled = true return nil } @@ -500,6 +500,7 @@ func TestAIStage_Success_AdvancesToDispatch(t *testing.T) { event := &model.MessageEvent{ ContactID: 90, ConversationID: 90, MessageContent: "test message", + MessageID: "crm-message-90", BotConfig: model.BotConfig{DebounceTime: 0}, } if err := svc.Process(ctx, event); err != nil { @@ -516,6 +517,99 @@ func TestAIStage_Success_AdvancesToDispatch(t *testing.T) { } } +func TestAIStage_AppendsResponseNoticeToSingleAgentDispatch(t *testing.T) { + dispatched := make(chan string, 1) + var aiMessage string + mock := &mockAIAdapter{ + callFn: func(_ context.Context, req *aiModel.A2ARequest) (*aiModel.NormalizedResponse, error) { + aiMessage = req.Message + return &aiModel.NormalizedResponse{Content: "Encontrei opções para sua reserva."}, nil + }, + } + dispatch := &mockDispatchEngine{ + dispatchFn: func(_ context.Context, _, _ int64, content string, _ model.BotConfig, _ string) error { + dispatched <- content + return nil + }, + } + svc, _ := setupSvcWithAIAndDispatch(t, mock, dispatch) + notice := "Observação: não consegui processar o áudio porque ele tem mais de 4 minutos." + event := &model.MessageEvent{ + ContactID: 99091, ConversationID: 99091, MessageID: "crm-message-99091", + MessageContent: "Quero reservar", ResponseNotice: notice, + BotConfig: model.BotConfig{DebounceTime: 0}, + } + + if err := svc.Process(context.Background(), event); err != nil { + t.Fatalf("Process returned error: %v", err) + } + + select { + case got := <-dispatched: + want := "Encontrei opções para sua reserva.\n\n" + notice + if got != want { + t.Errorf("dispatch content = %q, want %q", got, want) + } + case <-time.After(2 * time.Second): + t.Fatal("AI response was not dispatched") + } + if aiMessage != "Quero reservar" { + t.Errorf("AI received %q; response notice must stay out of user input", aiMessage) + } +} + +func TestResetDebouncePersistsAndDeduplicatesResponseNotice(t *testing.T) { + svc, _ := setupSvcFull(t) + ctx := context.Background() + firstNotice := "O áudio é longo demais." + event := &model.MessageEvent{ + ContactID: 99092, ConversationID: 99092, MessageID: "first", + MessageContent: "Primeira parte", ResponseNotice: firstNotice, + BotConfig: model.BotConfig{DebounceTime: 60}, + } + if err := svc.Process(ctx, event); err != nil { + t.Fatalf("first Process returned error: %v", err) + } + + secondNotice := &model.MessageEvent{ + ContactID: 99092, ConversationID: 99092, MessageID: "second", + MessageContent: "Segunda parte", ResponseNotice: firstNotice, + BotConfig: model.BotConfig{DebounceTime: 60}, + } + if err := svc.Process(ctx, secondNotice); err != nil { + t.Fatalf("second Process returned error: %v", err) + } + + state, err := svc.repo.GetState(ctx, 99092, 99092) + if err != nil { + t.Fatalf("GetState returned error: %v", err) + } + if state.ResponseNotice != firstNotice { + t.Errorf("persisted response notice = %q, want exactly one %q", state.ResponseNotice, firstNotice) + } + value, ok := svc.entries.Load(pairKey(99092, 99092)) + if !ok || value.(pipelineEntry).responseNotice != firstNotice { + t.Errorf("in-memory debounce entry did not retain the deduplicated notice: %#v", value) + } +} + +func TestStart_RecoveryCreatesStableLegacyIdempotencyKey(t *testing.T) { + svc, _ := setupSvcFull(t) + ctx := context.Background() + _ = svc.repo.SetState(ctx, 99809, 99809, &model.PipelineState{Stage: model.StageDebounce, CreatedAt: time.Now()}) + + if err := svc.Start(); err != nil { + t.Fatalf("Start error: %v", err) + } + state, err := svc.repo.GetState(ctx, 99809, 99809) + if err != nil { + t.Fatalf("GetState: %v", err) + } + if state == nil || state.Stage != model.StageAI || !strings.HasPrefix(state.MessageID, "debounce:") { + t.Fatalf("legacy recovery should advance with deterministic idempotency key, got %+v", state) + } +} + func TestAIStage_Cancellation_DoesNotClearNewDebounce(t *testing.T) { mock := &mockAIAdapter{ callFn: func(ctx context.Context, _ *aiModel.A2ARequest) (*aiModel.NormalizedResponse, error) {