feat: propagate audio metadata and idempotency

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 {

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)
 	}
 

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)
 	}
 }
 

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 {

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)
 	}

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)
 	}
 

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")})

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
 }

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",
+	}
+}

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.

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