diff --git a/pkg/ai/model/a2a.go b/pkg/ai/model/a2a.go index c668978..0a986b3 100644 --- a/pkg/ai/model/a2a.go +++ b/pkg/ai/model/a2a.go @@ -55,12 +55,21 @@ type A2AMessage struct { } type A2APart struct { - Type string `json:"type"` - Text string `json:"text"` + Type string `json:"type"` + Text string `json:"text,omitempty"` + Items []SelectItem `json:"items,omitempty"` +} + +// SelectItem represents a single selectable option from a Typebot choice block. +type SelectItem struct { + Title string `json:"title"` + Value string `json:"value"` } // NormalizedResponse is the internal format after parsing A2AResponse. // No JSON tags — this type never crosses a service boundary. type NormalizedResponse struct { - Content string + Content string + ContentType string // "text" (default) or "input_select" + Items []SelectItem // populated when ContentType == "input_select" } diff --git a/pkg/ai/service/ai_adapter.go b/pkg/ai/service/ai_adapter.go index 30a94b1..82cb3a5 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -121,42 +121,57 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor return nil, fmt.Errorf("pipeline.ai.decode: %w", err) } - content := extractResponseText(&a2aResp) + nr := extractNormalized(&a2aResp) slog.Info("pipeline.ai.http.completed", - "contact_id", req.ContactID, + "contact_id", req.ContactID, "conversation_id", req.ConversationID, "duration_ms", time.Since(start).Milliseconds(), + "content_type", nr.ContentType, + "has_items", len(nr.Items) > 0, ) - return &model.NormalizedResponse{Content: content}, nil + return nr, nil } -// extractResponseText extracts the text content from the A2A JSON-RPC response. -// Tries result.artifacts[0].parts[0].text first, then result.message.parts[0].text. -func extractResponseText(resp *model.A2AResponse) string { +// extractNormalized parses an A2A JSON-RPC response into a NormalizedResponse. +// It scans all artifacts and parts, capturing: +// - the first non-empty "text" part as Content +// - the first "select" part (if present) as ContentType=="input_select" + Items +// Falls back to result.message.parts when no artifact text is found. +func extractNormalized(resp *model.A2AResponse) *model.NormalizedResponse { + nr := &model.NormalizedResponse{ContentType: "text"} if resp.Result == nil { - return "" + return nr } - // Try artifacts first (primary response format) - if len(resp.Result.Artifacts) > 0 { - for _, artifact := range resp.Result.Artifacts { - for _, part := range artifact.Parts { - if part.Text != "" { - return part.Text + + // Scan all artifacts and all parts in each artifact. + for _, artifact := range resp.Result.Artifacts { + for _, part := range artifact.Parts { + switch part.Type { + case "text": + if nr.Content == "" && part.Text != "" { + nr.Content = part.Text + } + case "select": + if nr.ContentType == "text" && len(part.Items) > 0 { + nr.ContentType = "input_select" + nr.Items = part.Items } } } } - // Fallback to message format - if resp.Result.Message != nil { + + // Fallback to message format when no artifact text found. + if nr.Content == "" && resp.Result.Message != nil { for _, part := range resp.Result.Message.Parts { if part.Text != "" { - return part.Text + nr.Content = part.Text + break } } } - return "" + return nr } // nonNilMetadata ensures metadata is never nil (avoids "null" in JSON). diff --git a/pkg/ai/service/ai_adapter_test.go b/pkg/ai/service/ai_adapter_test.go index fe031b7..6c5368e 100644 --- a/pkg/ai/service/ai_adapter_test.go +++ b/pkg/ai/service/ai_adapter_test.go @@ -217,6 +217,51 @@ func TestCall_Timeout_ReturnsAITimeout(t *testing.T) { } } +func TestCall_SelectResponse_PropagatesItems(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + resp := aiModel.A2AResponse{ + Result: &aiModel.A2AResult{ + Artifacts: []aiModel.A2AArtifact{ + {Parts: []aiModel.A2APart{ + {Type: "text", Text: "Choose an option"}, + {Type: "select", Items: []aiModel.SelectItem{ + {Title: "Option A", Value: "a"}, + {Title: "Option B", Value: "b"}, + }}, + }}, + }, + }, + } + json.NewEncoder(w).Encode(resp) + })) + defer server.Close() + + adapter := aiService.NewAIAdapter(30) + resp, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: server.URL, + Message: "test", + ApiKey: "key", + }) + if err != nil { + t.Fatalf("Call returned unexpected error: %v", err) + } + if resp.ContentType != "input_select" { + t.Errorf("resp.ContentType = %q, want %q", resp.ContentType, "input_select") + } + if resp.Content != "Choose an option" { + t.Errorf("resp.Content = %q, want %q", resp.Content, "Choose an option") + } + if len(resp.Items) != 2 { + t.Fatalf("resp.Items length = %d, want 2", len(resp.Items)) + } + if resp.Items[0].Title != "Option A" || resp.Items[0].Value != "a" { + t.Errorf("resp.Items[0] = %+v, want {Title: Option A, Value: a}", resp.Items[0]) + } + if resp.Items[1].Title != "Option B" || resp.Items[1].Value != "b" { + t.Errorf("resp.Items[1] = %+v, want {Title: Option B, Value: b}", resp.Items[1]) + } +} + func TestCall_NonOKStatus_ReturnsError(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) diff --git a/pkg/dispatch/service/dispatch_engine.go b/pkg/dispatch/service/dispatch_engine.go index 125f212..d85c81c 100644 --- a/pkg/dispatch/service/dispatch_engine.go +++ b/pkg/dispatch/service/dispatch_engine.go @@ -14,6 +14,7 @@ import ( "unicode/utf8" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" + aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" ) @@ -25,7 +26,7 @@ type DispatchEngine interface { ctx context.Context, contactID int64, conversationID int64, - content string, + resp *aiModel.NormalizedResponse, cfg model.BotConfig, postbackURL string, ) error @@ -36,6 +37,7 @@ type postbackRequest struct { Content string `json:"content"` MessageType string `json:"message_type"` ContentType string `json:"content_type"` + Items []aiModel.SelectItem `json:"items,omitempty"` Attachments []postbackAttachment `json:"attachments,omitempty"` } @@ -69,17 +71,25 @@ func (d *dispatchEngineImpl) Dispatch( ctx context.Context, contactID int64, conversationID int64, - content string, + resp *aiModel.NormalizedResponse, cfg model.BotConfig, postbackURL string, ) error { - // Pull media URLs out of the full response BEFORE segmenting, so media is - // not split across text parts. Media is delivered in a single dedicated - // postback after the text parts (see below). The CRM also re-detects media - // from the text as a fallback, so this is an optimization, not the only path. - residual, atts := extractMediaURLs(content) - - parts := segmentContent(residual, cfg) + // Structured messages (input_select) must never be segmented — send as one part. + // Media extraction only applies to text responses. + var parts []string + var atts []postbackAttachment + if resp.ContentType == "input_select" { + parts = []string{resp.Content} + } else { + // Pull media URLs out of the full response BEFORE segmenting, so media is + // not split across text parts. Media is delivered in a single dedicated + // postback after the text parts (see below). The CRM also re-detects media + // from the text as a fallback, so this is an optimization, not the only path. + residual, extracted := extractMediaURLs(resp.Content) + atts = extracted + parts = segmentContent(residual, cfg) + } // Prepend signature to the first part (FR-21) if cfg.MessageSignature != "" && len(parts) > 0 { @@ -90,8 +100,10 @@ func (d *dispatchEngineImpl) Dispatch( for i, part := range parts { // Skip empty residual (e.g. response was only a media URL): the media - // postback below still runs. - if part == "" { + // postback below still runs. input_select is exempt — its items must + // always be delivered even when the AI processor sends no accompanying + // text (e.g. a choice block with no question text). + if part == "" && resp.ContentType != "input_select" { continue } @@ -107,7 +119,13 @@ func (d *dispatchEngineImpl) Dispatch( default: } - if err := d.sendPart(ctx, postbackURL, part, nil); err != nil { + // Only the last (or only) part carries structured items to avoid duplication. + var items []aiModel.SelectItem + if i == len(parts)-1 && resp.ContentType == "input_select" { + items = resp.Items + } + + if err := d.sendPart(ctx, postbackURL, part, resp.ContentType, nil, items); err != nil { return fmt.Errorf("pipeline.dispatch.send[%d]: %w", i, err) } @@ -134,7 +152,7 @@ func (d *dispatchEngineImpl) Dispatch( return brtErrors.ErrDispatchInterrupted default: } - if err := d.sendPart(ctx, postbackURL, "", atts); err != nil { + if err := d.sendPart(ctx, postbackURL, "", "", atts, nil); err != nil { return fmt.Errorf("pipeline.dispatch.send[media]: %w", err) } } @@ -149,13 +167,24 @@ func (d *dispatchEngineImpl) Dispatch( return nil } -// sendPart sends a single content part (and optional media attachments) to the -// postback URL. -func (d *dispatchEngineImpl) sendPart(ctx context.Context, postbackURL, content string, atts []postbackAttachment) error { +// sendPart sends a single content part (and optional media attachments or +// structured items) to the postback URL. +func (d *dispatchEngineImpl) sendPart( + ctx context.Context, + postbackURL string, + content string, + contentType string, + atts []postbackAttachment, + items []aiModel.SelectItem, +) error { + if contentType == "" { + contentType = "text" + } body, err := json.Marshal(postbackRequest{ Content: content, MessageType: "outgoing", - ContentType: "text", + ContentType: contentType, + Items: items, Attachments: atts, }) if err != nil { diff --git a/pkg/dispatch/service/dispatch_engine_test.go b/pkg/dispatch/service/dispatch_engine_test.go index 08f799f..c7bdacb 100644 --- a/pkg/dispatch/service/dispatch_engine_test.go +++ b/pkg/dispatch/service/dispatch_engine_test.go @@ -12,6 +12,7 @@ import ( "time" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" + aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" "github.com/EvolutionAPI/evo-bot-runtime/pkg/dispatch/service" "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" ) @@ -52,7 +53,8 @@ func TestDispatch_MultiPart_SignatureOnFirstOnly(t *testing.T) { } // "hello world this is test" → segments of ≤15 chars - if err := eng.Dispatch(context.Background(), 1, 1, "hello world this is test", cfg, server.URL); err != nil { + resp1 := &aiModel.NormalizedResponse{Content: "hello world this is test", ContentType: "text"} + if err := eng.Dispatch(context.Background(), 1, 1, resp1, cfg, server.URL); err != nil { t.Fatalf("Dispatch returned unexpected error: %v", err) } @@ -86,7 +88,8 @@ func TestDispatch_NoSegmentation_SinglePart(t *testing.T) { DelayPerCharacter: 0, } - if err := eng.Dispatch(context.Background(), 2, 2, "full response here", cfg, server.URL); err != nil { + resp2 := &aiModel.NormalizedResponse{Content: "full response here", ContentType: "text"} + if err := eng.Dispatch(context.Background(), 2, 2, resp2, cfg, server.URL); err != nil { t.Fatalf("Dispatch returned unexpected error: %v", err) } @@ -130,7 +133,8 @@ func TestDispatch_Cancellation_ReturnsInterrupted(t *testing.T) { cancel() }() - err := eng.Dispatch(ctx, 3, 3, "alpha beta gamma delta epsilon", cfg, server.URL) + resp3 := &aiModel.NormalizedResponse{Content: "alpha beta gamma delta epsilon", ContentType: "text"} + err := eng.Dispatch(ctx, 3, 3, resp3, cfg, server.URL) if !errors.Is(err, brtErrors.ErrDispatchInterrupted) { t.Errorf("expected ErrDispatchInterrupted, got %v", err) } @@ -155,7 +159,8 @@ func TestDispatch_EmptySignature_NoSuffix(t *testing.T) { MessageSignature: "", // empty — no suffix } - if err := eng.Dispatch(context.Background(), 4, 4, "no signature here", cfg, server.URL); err != nil { + resp4 := &aiModel.NormalizedResponse{Content: "no signature here", ContentType: "text"} + if err := eng.Dispatch(context.Background(), 4, 4, resp4, cfg, server.URL); err != nil { t.Fatalf("Dispatch returned unexpected error: %v", err) } @@ -181,7 +186,8 @@ func TestDispatch_NonOKResponse_ReturnsError(t *testing.T) { eng := service.NewDispatchEngine("") cfg := model.BotConfig{TextSegmentationEnabled: false} - err := eng.Dispatch(context.Background(), 8, 8, "some content", cfg, server.URL) + resp8 := &aiModel.NormalizedResponse{Content: "some content", ContentType: "text"} + err := eng.Dispatch(context.Background(), 8, 8, resp8, cfg, server.URL) if err == nil { t.Fatal("expected error for non-2xx response, got nil") } @@ -200,7 +206,8 @@ func TestSegmentContent_MergeDoesNotExceedLimit(t *testing.T) { DelayPerCharacter: 0, } - if err := eng.Dispatch(context.Background(), 6, 6, "hello world test", cfg, server.URL); err != nil { + resp6 := &aiModel.NormalizedResponse{Content: "hello world test", ContentType: "text"} + if err := eng.Dispatch(context.Background(), 6, 6, resp6, cfg, server.URL); err != nil { t.Fatalf("Dispatch returned unexpected error: %v", err) } @@ -234,7 +241,8 @@ func TestSegmentContent_RuneAwareLimits(t *testing.T) { DelayPerCharacter: 0, } - if err := eng.Dispatch(context.Background(), 7, 7, "olá mundo", cfg, server.URL); err != nil { + resp7 := &aiModel.NormalizedResponse{Content: "olá mundo", ContentType: "text"} + if err := eng.Dispatch(context.Background(), 7, 7, resp7, cfg, server.URL); err != nil { t.Fatalf("Dispatch returned unexpected error: %v", err) } @@ -251,6 +259,189 @@ func TestSegmentContent_RuneAwareLimits(t *testing.T) { } } +// postbackBodyWithItems captures items and attachments for select/media tests. +type postbackBodyWithItems struct { + Content string `json:"content"` + MessageType string `json:"message_type"` + ContentType string `json:"content_type"` + Items []aiModel.SelectItem `json:"items,omitempty"` + Attachments []postbackAttachment `json:"attachments,omitempty"` +} + +type postbackAttachment struct { + URL string `json:"url"` + FileType string `json:"file_type"` +} + +func TestDispatch_InputSelect_NoSegmentation(t *testing.T) { + var bodies []postbackBodyWithItems + var mu sync.Mutex + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var b postbackBodyWithItems + json.NewDecoder(r.Body).Decode(&b) //nolint:errcheck + mu.Lock() + bodies = append(bodies, b) + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + eng := service.NewDispatchEngine("") + cfg := model.BotConfig{ + TextSegmentationEnabled: true, + TextSegmentationLimit: 5, // would normally split "choose an option" into multiple parts + DelayPerCharacter: 0, + } + + items := []aiModel.SelectItem{ + {Title: "Yes", Value: "yes"}, + {Title: "No", Value: "no"}, + } + resp := &aiModel.NormalizedResponse{ + Content: "choose an option", + ContentType: "input_select", + Items: items, + } + if err := eng.Dispatch(context.Background(), 10, 10, resp, cfg, server.URL); err != nil { + t.Fatalf("Dispatch returned unexpected error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + if len(bodies) != 1 { + t.Fatalf("input_select must not be segmented: expected 1 postback, got %d", len(bodies)) + } + if bodies[0].ContentType != "input_select" { + t.Errorf("ContentType = %q, want %q", bodies[0].ContentType, "input_select") + } + if bodies[0].Content != "choose an option" { + t.Errorf("Content = %q, want %q", bodies[0].Content, "choose an option") + } + if len(bodies[0].Items) != 2 { + t.Fatalf("Items length = %d, want 2", len(bodies[0].Items)) + } + if bodies[0].Items[0].Title != "Yes" || bodies[0].Items[0].Value != "yes" { + t.Errorf("Items[0] = %+v, want {Yes yes}", bodies[0].Items[0]) + } + if bodies[0].Items[1].Title != "No" || bodies[0].Items[1].Value != "no" { + t.Errorf("Items[1] = %+v, want {No no}", bodies[0].Items[1]) + } +} + +// TestDispatch_InputSelect_EmptyContent_StillSendsItems is a regression test: +// when the AI processor sends a select block with no accompanying question +// text (Content == ""), the items must still reach the postback endpoint. +// A prior version of the "skip empty residual" guard (meant for the +// text/media path) also swallowed empty input_select parts, silently +// dropping the items. +func TestDispatch_InputSelect_EmptyContent_StillSendsItems(t *testing.T) { + var bodies []postbackBodyWithItems + var mu sync.Mutex + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var b postbackBodyWithItems + json.NewDecoder(r.Body).Decode(&b) //nolint:errcheck + mu.Lock() + bodies = append(bodies, b) + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + eng := service.NewDispatchEngine("") + cfg := model.BotConfig{TextSegmentationEnabled: false} + + items := []aiModel.SelectItem{ + {Title: "Yes", Value: "yes"}, + {Title: "No", Value: "no"}, + } + resp := &aiModel.NormalizedResponse{ + Content: "", + ContentType: "input_select", + Items: items, + } + if err := eng.Dispatch(context.Background(), 12, 12, resp, cfg, server.URL); err != nil { + t.Fatalf("Dispatch returned unexpected error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + if len(bodies) != 1 { + t.Fatalf("expected 1 postback carrying the items even with empty content, got %d", len(bodies)) + } + if bodies[0].ContentType != "input_select" { + t.Errorf("ContentType = %q, want %q", bodies[0].ContentType, "input_select") + } + if len(bodies[0].Items) != 2 { + t.Fatalf("Items length = %d, want 2 — items must not be dropped when Content is empty", len(bodies[0].Items)) + } +} + +func TestDispatch_TextWithMedia_ExtractsAttachments(t *testing.T) { + var bodies []postbackBodyWithItems + var mu sync.Mutex + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var b postbackBodyWithItems + json.NewDecoder(r.Body).Decode(&b) //nolint:errcheck + mu.Lock() + bodies = append(bodies, b) + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + eng := service.NewDispatchEngine("") + cfg := model.BotConfig{ + TextSegmentationEnabled: false, + DelayPerCharacter: 0, + } + + resp := &aiModel.NormalizedResponse{ + Content: "Check this photo https://example.com/image.png", + ContentType: "text", + } + if err := eng.Dispatch(context.Background(), 11, 11, resp, cfg, server.URL); err != nil { + t.Fatalf("Dispatch returned unexpected error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + // Expect 2 postbacks: 1 text (media URL stripped) + 1 media-only + if len(bodies) != 2 { + t.Fatalf("expected 2 postbacks (text + media), got %d", len(bodies)) + } + + // First postback: text with media URL removed + if bodies[0].ContentType != "text" { + t.Errorf("bodies[0].ContentType = %q, want %q", bodies[0].ContentType, "text") + } + if strings.Contains(bodies[0].Content, "image.png") { + t.Errorf("bodies[0].Content must not contain media URL: %q", bodies[0].Content) + } + if len(bodies[0].Attachments) != 0 { + t.Errorf("bodies[0] must have no attachments, got %d", len(bodies[0].Attachments)) + } + + // Second postback: media-only (empty content, 1 attachment) + if bodies[1].Content != "" { + t.Errorf("bodies[1].Content = %q, want empty (media-only)", bodies[1].Content) + } + if len(bodies[1].Attachments) != 1 { + t.Fatalf("bodies[1].Attachments length = %d, want 1", len(bodies[1].Attachments)) + } + if bodies[1].Attachments[0].URL != "https://example.com/image.png" { + t.Errorf("Attachments[0].URL = %q, want %q", bodies[1].Attachments[0].URL, "https://example.com/image.png") + } + if bodies[1].Attachments[0].FileType != "image" { + t.Errorf("Attachments[0].FileType = %q, want %q", bodies[1].Attachments[0].FileType, "image") + } +} + func TestDispatch_ValidatesPostBody(t *testing.T) { var received postbackBody server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -265,7 +456,8 @@ func TestDispatch_ValidatesPostBody(t *testing.T) { eng := service.NewDispatchEngine("") cfg := model.BotConfig{TextSegmentationEnabled: false} - if err := eng.Dispatch(context.Background(), 5, 5, "test content", cfg, server.URL); err != nil { + resp5 := &aiModel.NormalizedResponse{Content: "test content", ContentType: "text"} + if err := eng.Dispatch(context.Background(), 5, 5, resp5, cfg, server.URL); err != nil { t.Fatalf("Dispatch returned unexpected error: %v", err) } diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index d347c57..cb05c4f 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -424,7 +424,7 @@ func (s *pipelineService) launchDispatchStage( cfg model.BotConfig, postbackURL string, ) { - go s.runDispatchStage(ctx, contactID, conversationID, resp.Content, cfg, postbackURL) + go s.runDispatchStage(ctx, contactID, conversationID, resp, cfg, postbackURL) } // runDispatchStage is the dispatch stage goroutine body. ctx is pipelineEntry.ctx — cancelled by @@ -433,7 +433,7 @@ func (s *pipelineService) runDispatchStage( ctx context.Context, contactID int64, conversationID int64, - content string, + resp *aiModel.NormalizedResponse, cfg model.BotConfig, postbackURL string, ) { @@ -445,7 +445,7 @@ func (s *pipelineService) runDispatchStage( ) start := time.Now() - err := s.dispatchEng.Dispatch(ctx, contactID, conversationID, content, cfg, postbackURL) + err := s.dispatchEng.Dispatch(ctx, contactID, conversationID, resp, cfg, postbackURL) if err != nil { switch { case errors.Is(err, brtErrors.ErrDispatchInterrupted): diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index f500faf..b94e6da 100644 --- a/pkg/pipeline/service/pipeline_service_test.go +++ b/pkg/pipeline/service/pipeline_service_test.go @@ -43,11 +43,11 @@ var _ aiIface.AIAdapter = (*mockAIAdapter)(nil) // mockDispatchEngine implements dispatchIface.DispatchEngine for testing. type mockDispatchEngine struct { - dispatchFn func(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig, postbackURL string) error + dispatchFn func(ctx context.Context, contactID, conversationID int64, resp *aiModel.NormalizedResponse, cfg model.BotConfig, postbackURL string) error } -func (m *mockDispatchEngine) Dispatch(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig, postbackURL string) error { - return m.dispatchFn(ctx, contactID, conversationID, content, cfg, postbackURL) +func (m *mockDispatchEngine) Dispatch(ctx context.Context, contactID, conversationID int64, resp *aiModel.NormalizedResponse, cfg model.BotConfig, postbackURL string) error { + return m.dispatchFn(ctx, contactID, conversationID, resp, cfg, postbackURL) } var _ dispatchIface.DispatchEngine = (*mockDispatchEngine)(nil) // compile-time check @@ -134,7 +134,7 @@ func setupSvcWithAIAndDispatch(t *testing.T, ai aiIface.AIAdapter, dispatch disp // that existing tests checking for StageDispatch see it before dispatch completes. func setupSvcWithAI(t *testing.T, ai aiIface.AIAdapter) (*pipelineService, *redis.Client) { blockingDispatch := &mockDispatchEngine{ - dispatchFn: func(ctx context.Context, _, _ int64, _ string, _ model.BotConfig, _ string) error { + dispatchFn: func(ctx context.Context, _, _ int64, _ *aiModel.NormalizedResponse, _ model.BotConfig, _ string) error { <-ctx.Done() return brtErrors.ErrDispatchInterrupted }, @@ -593,7 +593,7 @@ func TestPipeline_FullFlow_DispatchCompletes(t *testing.T) { }, } mockDispatch := &mockDispatchEngine{ - dispatchFn: func(_ context.Context, _, _ int64, _ string, _ model.BotConfig, _ string) error { + dispatchFn: func(_ context.Context, _, _ int64, _ *aiModel.NormalizedResponse, _ model.BotConfig, _ string) error { return nil }, } @@ -626,7 +626,7 @@ func TestPipeline_DispatchInterrupted_KeepsNewDebounce(t *testing.T) { } dispatchDone := make(chan struct{}) mockDispatch := &mockDispatchEngine{ - dispatchFn: func(ctx context.Context, _, _ int64, _ string, _ model.BotConfig, _ string) error { + dispatchFn: func(ctx context.Context, _, _ int64, _ *aiModel.NormalizedResponse, _ model.BotConfig, _ string) error { defer close(dispatchDone) <-ctx.Done() // block until pipeline context is cancelled return brtErrors.ErrDispatchInterrupted @@ -679,7 +679,7 @@ func TestPipeline_DispatchError_ClearsState(t *testing.T) { } dispatchDone := make(chan struct{}) mockDispatch := &mockDispatchEngine{ - dispatchFn: func(_ context.Context, _, _ int64, _ string, _ model.BotConfig, _ string) error { + dispatchFn: func(_ context.Context, _, _ int64, _ *aiModel.NormalizedResponse, _ model.BotConfig, _ string) error { defer close(dispatchDone) return fmt.Errorf("postback server unavailable") }, diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index fc1de7e..da8cab6 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) + dispatch := dispatchService.NewDispatchEngine("") pipeline := pipelineService.NewPipelineService(repo, debounce, ai, dispatch) if err := pipeline.Start(); err != nil { t.Fatalf("pipeline.Start: %v", err) @@ -709,8 +709,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) + dispatch := dispatchService.NewDispatchEngine("") aiSrv.setHandler(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json")