diff --git a/.env.example b/.env.example index a2b6115..8504a71 100644 --- a/.env.example +++ b/.env.example @@ -5,7 +5,13 @@ REDIS_URL=redis://localhost:6379 # An empty value would authenticate any caller that omits the header. BOT_RUNTIME_SECRET= -AI_CALL_TIMEOUT_SECONDS=30 +# CRM-236: a tool-calling turn makes two model calls and the provider's tail adds +# up to ~20s each. An explicit value here overrides the code default. +AI_CALL_TIMEOUT_SECONDS=90 + +# Message sent to the customer when the AI cannot answer (timeout or provider +# outage). Empty string disables it and restores the old silence. +# AI_FAILURE_NOTICE=We are having a temporary issue and could not answer right now. We will get back to you shortly. # Required for incoming media. Hosts allowed to serve it, comma-separated, no # scheme or port (e.g. "crm.example.com,minio.internal"). diff --git a/internal/config/config.go b/internal/config/config.go index 9781220..fa5be62 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,7 +33,9 @@ func Load() (*Config, error) { if err != nil { return nil, err } - aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 30) + // CRM-236: 90s, not 30. A tool-calling turn makes two model calls and the + // provider's tail alone measured 20.4s on a trivial prompt. + aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 90) if err != nil { return nil, err } diff --git a/k8s/configmap.yaml b/k8s/configmap.yaml index d1bee04..d3bbbda 100644 --- a/k8s/configmap.yaml +++ b/k8s/configmap.yaml @@ -5,7 +5,9 @@ metadata: data: LISTEN_ADDR: ":8080" AI_PROCESSOR_URL: "http://ai-processor:8000" - AI_CALL_TIMEOUT_SECONDS: "30" + # CRM-236: an explicit value overrides the code default, and this ConfigMap is + # what runs in staging/production. + AI_CALL_TIMEOUT_SECONDS: "90" # Hosts allowed to serve incoming media, comma-separated (no scheme/port). # Must include the host of the CRM's BACKEND_URL, or no media reaches the agent. MEDIA_HOST_ALLOWLIST: "" diff --git a/pkg/pipeline/service/ai_failure_notice_test.go b/pkg/pipeline/service/ai_failure_notice_test.go new file mode 100644 index 0000000..85a4b1d --- /dev/null +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -0,0 +1,210 @@ +package service + +import ( + "context" + "errors" + "os" + "strings" + "testing" + "time" + + brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" + "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" +) + +// CRM-236: a degraded provider used to end the turn in silence, while the tool's +// side effect (a moved pipeline card) had already been applied. + +func captureDispatch(t *testing.T) (*mockDispatchEngine, *[]string) { + t.Helper() + var sent []string + engine := &mockDispatchEngine{ + dispatchFn: func(_ context.Context, _, _ int64, content string, _ model.BotConfig, _ string) error { + sent = append(sent, content) + return nil + }, + } + return engine, &sent +} + +func TestAIFailureNotice_TimeoutTellsTheCustomer(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 1 { + t.Fatalf("expected the customer to receive one notice, got %d", len(*sent)) + } + if (*sent)[0] != defaultAIFailureNotice { + t.Errorf("unexpected notice: %q", (*sent)[0]) + } +} + +func TestAIFailureNotice_NeverLeaksTheProviderError(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + // A real provider error: model names, quota ids and URLs must not reach the customer. + cause := errors.New("litellm.RateLimitError: VertexAIException - 429 RESOURCE_EXHAUSTED " + + "Quota exceeded for metric generativelanguage.googleapis.com/generate_content_free_tier_requests, " + + "limit: 20, model: gemini-2.5-flash") + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", cause) + + if len(*sent) != 1 { + t.Fatalf("expected one notice, got %d", len(*sent)) + } + for _, leak := range []string{"gemini", "Quota", "RateLimitError", "googleapis"} { + if strings.Contains((*sent)[0], leak) { + t.Errorf("provider detail %q leaked to the customer: %q", leak, (*sent)[0]) + } + } +} + +func TestAIFailureNotice_OperatorCanCustomiseIt(t *testing.T) { + t.Setenv(aiFailureNoticeEnv, "Nosso atendimento automático está indisponível.") + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 1 || (*sent)[0] != "Nosso atendimento automático está indisponível." { + t.Fatalf("custom notice not used: %v", *sent) + } +} + +// An operator who prefers silence must be able to keep it. +func TestAIFailureNotice_EmptyEnvDisablesIt(t *testing.T) { + t.Setenv(aiFailureNoticeEnv, "") + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 0 { + t.Fatalf("notice should be disabled, got %v", *sent) + } +} + +func TestAIFailureNotice_NoPostbackUrlIsNotACrash(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "", brtErrors.ErrAITimeout) + + if len(*sent) != 0 { + t.Fatalf("nothing can be dispatched without a postback url, got %v", *sent) + } +} + +// The default must survive an env var that exists but is unrelated. +func TestAIFailureNotice_DefaultWhenEnvUnset(t *testing.T) { + // Low 13: restore whatever the process had, instead of leaving the env mutated + // for every test that runs after this one. + if previous, had := os.LookupEnv(aiFailureNoticeEnv); had { + t.Cleanup(func() { os.Setenv(aiFailureNoticeEnv, previous) }) + } + os.Unsetenv(aiFailureNoticeEnv) + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + if len(*sent) != 1 || (*sent)[0] != defaultAIFailureNotice { + t.Fatalf("expected the default notice, got %v", *sent) + } +} + +func TestAIFailureNotice_DoesNotWriteTurnState(t *testing.T) { + engine, _ := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + // The callers already cleared the state before asking for the notice; writing + // StageDone here would resurrect state for a turn that is over — and, in the + // follow-up race, stamp it over the NEW turn's state. + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + // A read error must fail the test, not pass it: GetState returns (nil, nil) + // for a missing key, so nil-with-error proves nothing about what was written. + state, err := svc.repo.GetState(context.Background(), 1, 2) + if err != nil { + t.Fatalf("could not read the state back: %v", err) + } + if state != nil { + t.Fatalf("the notice wrote turn state: stage=%v", state.Stage) + } +} + +// The entry, not just the state: entries.Delete(pairKey) orphaned the follow-up +// turn, so the message after it started a second concurrent pipeline. +func TestAIFailureNotice_DoesNotTouchTheEntryOfTheNextTurn(t *testing.T) { + engine, sent := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + // The follow-up turn, exactly as startDebounce leaves it: an entry in the map + // and StageDebounce in Redis, both under the pair the notice is about to use. + key := pairKey(1, 2) + nextTurn, cancelNextTurn := context.WithCancel(context.Background()) + defer cancelNextTurn() + svc.entries.Store(key, pipelineEntry{ctx: nextTurn, cancel: cancelNextTurn}) + debounce := &model.PipelineState{Stage: model.StageDebounce, CreatedAt: time.Now()} + if err := svc.repo.SetState(context.Background(), 1, 2, debounce); err != nil { + t.Fatalf("could not seed the next turn's state: %v", err) + } + t.Cleanup(func() { svc.repo.ClearState(context.Background(), 1, 2) }) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + // Guard the guard: the bookkeeping only runs after a successful dispatch, so + // a notice that never went out would satisfy the assertions below for free. + if len(*sent) != 1 { + t.Fatalf("the notice never dispatched, so this proves nothing: %v", *sent) + } + + stored, ok := svc.entries.Load(key) + if !ok { + t.Fatal("the notice deleted the next turn's entry: it can no longer be cancelled, so the message after it starts a second concurrent pipeline") + } + if entry, _ := stored.(pipelineEntry); entry.ctx != nextTurn { + t.Error("the next turn's entry was replaced by the notice") + } + + // SetState(StageDone) followed by ClearState leaves nothing behind, so only a + // seeded state can witness it: the next turn must still be in StageDebounce. + state, err := svc.repo.GetState(context.Background(), 1, 2) + if err != nil { + t.Fatalf("could not read the next turn's state back: %v", err) + } + if state == nil { + t.Fatal("the notice cleared the next turn's state: its debounce is lost") + } + if state.Stage != model.StageDebounce { + t.Errorf("the notice stamped the next turn's state: stage=%v, want %v", state.Stage, model.StageDebounce) + } +} + +// The notice is a real dispatch (segmented, with per-rune delays), not a cleanup +// call. Bounding it with cleanupCtx's 5s truncated it and then logged +// "New message arrived" when nothing had arrived. +func TestAIFailureNotice_HasRoomForASegmentedDispatch(t *testing.T) { + ctx, cancel := noticeCtx() + defer cancel() + + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("the notice dispatch must stay bounded") + } + if remaining := time.Until(deadline); remaining <= 10*time.Second { + t.Fatalf("notice budget is %v; a segmented dispatch with per-rune delays needs more", remaining) + } +} + +// The default reaches customers of installations that never chose Portuguese. +func TestAIFailureNotice_DefaultIsLocaleNeutralEnglish(t *testing.T) { + for _, ptBR := range []string{"instabilidade", "Já retorno", "não consegui"} { + if strings.Contains(defaultAIFailureNotice, ptBR) { + t.Errorf("default notice still hardcodes pt-BR (%q): %q", ptBR, defaultAIFailureNotice) + } + } +} diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index 63c610a..c2983d2 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "os" "runtime/debug" "strconv" "strings" @@ -415,6 +416,10 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio "conversation_id", conversationID, ) s.clearStateWithLog(contactID, conversationID) + // CRM-236: silence is indistinguishable from "the bot is ignoring you", + // and the tool's side effect may already be applied (the card moved at + // ~20s, the timeout fired at 30s). Tell the customer something. + s.sendAIFailureNotice(contactID, conversationID, cfg, postbackURL, err) default: slog.Error("pipeline.ai.error", "contact_id", contactID, @@ -422,6 +427,7 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio "error", fmt.Errorf("pipeline.ai: %w", err), ) s.clearStateWithLog(contactID, conversationID) + s.sendAIFailureNotice(contactID, conversationID, cfg, postbackURL, err) } return } @@ -643,6 +649,76 @@ func cleanupCtx() (context.Context, context.CancelFunc) { return context.WithTimeout(context.Background(), 5*time.Second) } +// aiFailureNoticeEnv overrides the message the customer receives when the AI +// backend times out or errors. Empty string disables the notice entirely, for +// operators who prefer silence to a canned reply. +const aiFailureNoticeEnv = "AI_FAILURE_NOTICE" + +// English: the runtime ships worldwide and a pt-BR default reached installs that +// never chose it. Operators localise it with AI_FAILURE_NOTICE. +const defaultAIFailureNotice = "We are having a temporary issue and could not answer right now. We will get back to you shortly." + +// sendAIFailureNotice replaces the silent turn with one sentence to the customer. +// The provider's raw error goes to the operator's log, never to the chat. +func (s *pipelineService) sendAIFailureNotice( + contactID, conversationID int64, + cfg model.BotConfig, + postbackURL string, + cause error, +) { + notice := defaultAIFailureNotice + if v, ok := os.LookupEnv(aiFailureNoticeEnv); ok { + if strings.TrimSpace(v) == "" { + slog.Info("pipeline.ai.failure_notice.disabled", + "contact_id", contactID, + "conversation_id", conversationID, + ) + return + } + notice = v + } + + if postbackURL == "" { + slog.Warn("pipeline.ai.failure_notice.no_postback", + "contact_id", contactID, + "conversation_id", conversationID, + ) + return + } + + slog.Warn("pipeline.ai.failure_notice.sending", + "contact_id", contactID, + "conversation_id", conversationID, + "cause", cause.Error(), + ) + + // Dispatch directly: runDispatchStage ends in entries.Delete(pairKey), which + // would orphan a follow-up turn. Both callers already cleared the state. + ctx, cancel := noticeCtx() + defer cancel() + defer s.recoverPipeline(contactID, conversationID) + + if err := s.dispatchEng.Dispatch(ctx, contactID, conversationID, notice, cfg, postbackURL); err != nil { + slog.Warn("pipeline.ai.failure_notice.failed", + "contact_id", contactID, + "conversation_id", conversationID, + "error", err, + ) + return + } + + slog.Info("pipeline.ai.failure_notice.sent", + "contact_id", contactID, + "conversation_id", conversationID, + ) +} + +// noticeCtx bounds the notice's dispatch. Not cleanupCtx: a Dispatch segments the +// text and sleeps per rune between parts, which overruns its 5s. +func noticeCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 30*time.Second) +} + // clearStateWithLog calls ClearState and logs a warning if it fails. // Used in all goroutine error/cleanup paths where the error is non-actionable // but should not be silently swallowed. diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index 2aa7eec..0fb2e1c 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -533,6 +533,12 @@ func TestE2E_ExactlyOnce_ConcurrentEvents(t *testing.T) { // The panic recovery path is unit-tested but no test has exercised two concurrent // live pipelines where one fails and the other must succeed. func TestE2E_PipelineIsolation(t *testing.T) { + // Pinned through the operator-facing env instead of reaching for the + // package constant: it keeps this test out of service's internals and + // exercises AI_FAILURE_NOTICE on the way. + const pairANotice = "pair-a failure notice" + t.Setenv("AI_FAILURE_NOTICE", pairANotice) + h := newHarness(t) pairAContact, pairAConv := nextPair() pairBContact, pairBConv := nextPair() @@ -556,13 +562,36 @@ func TestE2E_PipelineIsolation(t *testing.T) { h.postEvent(t, h.event(pairAContact, pairAConv, "pair-a", 0)).Body.Close() h.postEvent(t, h.event(pairBContact, pairBConv, "pair-b", 0)).Body.Close() - h.pbServer.waitForCall(t, 3*time.Second) + // CRM-236 changed what pair A does on an AI error: it used to fail SILENTLY + // (log + clear state, nothing reaching the chat), and now it sends the + // customer a failure notice. So this pair delivers too, and the assertion + // "only pair B should deliver" no longer describes intended behaviour. + // + // What this test is actually about — isolation — is now checked more + // strictly than before: each pair must receive ITS OWN message, so a + // crossed delivery fails here even though the call count would be right. + h.pbServer.waitForNCalls(t, 2, 3*time.Second) - if n := h.pbServer.callCount(); n != 1 { - t.Errorf("postback called %d times, want 1 (only pair B should deliver)", n) + if n := h.pbServer.callCount(); n != 2 { + t.Errorf("postback called %d times, want 2 (pair B's answer + pair A's failure notice)", n) + } + + var sawResponse, sawNotice bool + for _, body := range h.pbServer.allBodies() { + switch content := decodePostbackContent(body); content { + case "pair-b response": + sawResponse = true + case pairANotice: + sawNotice = true + default: + t.Errorf("unexpected postback content %q", content) + } + } + if !sawResponse { + t.Error("pair B's answer never arrived: pair A's AI error leaked into pair B") } - if got := decodePostbackContent(h.pbServer.lastBody()); got != "pair-b response" { - t.Errorf("postback content = %q, want %q", got, "pair-b response") + if !sawNotice { + t.Error("pair A got no failure notice: its customer is left in silence (CRM-236)") } // Pair A must leave no state in Redis after its AI error. Its cleanup runs in