From fa3fa40b13f8d2e9231ad42d69e3fafb894830e6 Mon Sep 17 00:00:00 2001 From: Matheus Pastorini Date: Sat, 22 Aug 2026 09:06:03 -0300 Subject: [PATCH 1/6] fix(pipeline): tell the customer when the AI backend fails (CRM-236) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the LLM provider degrades the turn exceeds the ceiling and the pipeline only logged and cleared state — nothing reached the chat. For the customer that is indistinguishable from a bot ignoring them, and it is worse than it looks: the tool's side effect may ALREADY be applied. In the live run the pipeline card moved at ~20s and the timeout fired at 30s, so the funnel advanced while the conversation stayed silent. Two changes. 1. AI_CALL_TIMEOUT_SECONDS default 30 -> 90. Measured against gemini-2.5-flash, the SAME trivial prompt answered in 0.74s / 0.79s / 1.36s / 10.34s / 20.40s — a 27x spread on the provider's tail. A tool-calling turn makes at least two of those round trips (decide the tool, then write the reply), so two bad tails alone exceed 30s with nothing wrong in the code. 90s covers that and still bounds a genuinely hung provider. Note this is NOT the fix on its own: raising a ceiling only moves it. The notice below is what protects the customer when the ceiling IS reached. 2. On timeout or error, dispatch a plain sentence to the conversation instead of silence. The provider's raw error never reaches the customer — it carries model names, quota ids and URLs (litellm.RateLimitError: ... limit: 20, model: gemini-2.5-flash) — it goes to the operator's log as `cause`. AI_FAILURE_NOTICE overrides the text; setting it empty keeps today's silence for operators who prefer it. Tests: 6 in pkg/pipeline/service/ai_failure_notice_test.go — the customer is told, the provider error never leaks, the text is overridable, empty disables it, a missing postback url is not a crash, and the default holds when the env is unset. go build + go vet clean; full suite green (pkg/... and internal/..., Redis-backed). Not addressed here, deliberately: the processor keeps working after the bot-runtime gives up, and the provider's 429/503 still surfaces as a generic 500 on the A2A route. Both are real and belong to the processor side. --- internal/config/config.go | 13 +- .../service/ai_failure_notice_test.go | 125 ++++++++++++++++++ pkg/pipeline/service/pipeline_service.go | 62 +++++++++ 3 files changed, 199 insertions(+), 1 deletion(-) create mode 100644 pkg/pipeline/service/ai_failure_notice_test.go diff --git a/internal/config/config.go b/internal/config/config.go index 9781220..8a1905a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,7 +33,18 @@ func Load() (*Config, error) { if err != nil { return nil, err } - aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 30) + // CRM-236: 30s was dimensioned for a single, well-behaved model call. Measured + // against gemini-2.5-flash, the SAME trivial prompt answered in + // 0.74s / 0.79s / 1.36s / 10.34s / 20.40s — a 27x spread on the provider's + // tail. A tool-calling turn makes at least TWO of those round trips (decide the + // tool, then write the reply), so two bad tails alone exceed 30s with nothing + // wrong in the code — and the customer got silence while the tool's side effect + // (a moved pipeline card) had already been applied. + // + // 90s covers two tail-latency calls plus the CRM round trip, and still bounds a + // genuinely hung provider. It is not a licence to hang: the fallback message + // below is what protects the customer when the ceiling IS reached. + aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 90) if err != nil { return nil, err } 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..26bc0fb --- /dev/null +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -0,0 +1,125 @@ +package service + +import ( + "context" + "errors" + "os" + "testing" + + brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" + "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" +) + +// CRM-236: when the LLM provider degrades (429 quota, 503 high demand, or just a +// slow tail) the turn exceeds the ceiling. The pipeline used to only log and +// clear state, so NOTHING reached the chat — indistinguishable, for the +// customer, from a bot that is ignoring them. Worse, the tool's side effect may +// already be applied: the pipeline card moves at ~20s and the timeout fires +// later, so the funnel advances while the conversation stays silent. + +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 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) { + 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 contains(haystack, needle string) bool { + return len(haystack) >= len(needle) && (func() bool { + for i := 0; i+len(needle) <= len(haystack); i++ { + if haystack[i:i+len(needle)] == needle { + return true + } + } + return false + })() +} diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index 63c610a..5b0a38f 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 } @@ -646,6 +652,62 @@ func cleanupCtx() (context.Context, context.CancelFunc) { // 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. +// 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" + +const defaultAIFailureNotice = "Estou com uma instabilidade no momento e não consegui responder agora. Já retorno." + +// sendAIFailureNotice tells the customer something went wrong instead of leaving +// the turn silent. +// +// CRM-236: when the provider degrades (429 quota, 503 high demand, or a slow +// tail) the turn exceeds the ceiling and the pipeline used to just log and clear +// state. Nothing reached the chat, so the customer could not tell the difference +// between a broken bot and one ignoring them — and the tool's side effect may +// ALREADY be applied (the pipeline card moves at ~20s, the timeout fires later). +// +// The reason is logged for the operator; the customer gets a plain sentence, +// never the provider's raw error (it carries model names, quotas and URLs). +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(), + ) + + // Own context: the pipeline ctx is already cancelled or timed out by now. + ctx, cancel := cleanupCtx() + defer cancel() + s.runDispatchStage(ctx, contactID, conversationID, notice, cfg, postbackURL) +} + func (s *pipelineService) clearStateWithLog(contactID, conversationID int64) { ctx, cancel := cleanupCtx() defer cancel() From d84f8b012ed2a7d2563dc7b36822f2a560725ebd Mon Sep 17 00:00:00 2001 From: Matheus Pastorini Date: Sat, 22 Aug 2026 14:19:18 -0300 Subject: [PATCH 2/6] fix(pipeline): the failure notice must not destroy the next turn (CRM-236 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the CRITICAL findings on the bot-runtime side. 1. The notice could delete the follow-up turn's entry. sendAIFailureNotice dispatched through runDispatchStage, which owns the turn's bookkeeping: its success path runs SetState(StageDone) -> ClearState -> entries.Delete(pairKey). That Delete is what runDispatchStage's own comments forbid ("A Delete here would race with the new event's Store and could delete the replacement entry"). The race lands on the very scenario this feature targets — the customer who waited and follows up. They send "oi?" while the notice is being dispatched, startDebounce stores the new turn, then the notice finishes and deletes THAT entry and clears its state. The new turn is orphaned: the next message cannot cancel it, so two pipelines run concurrently on the same pair and the customer gets a duplicated reply. The notice now dispatches directly. There was nothing to book-keep anyway: both call sites already run clearStateWithLog before asking for it. 2. The 5s budget truncated the notice and then lied about why. cleanupCtx's 5s are documented for cleanup calls (ClearState, SetState). A Dispatch is a different animal: it segments by TextSegmentationLimit and sleeps DelayPerCharacter per rune between parts. With segmentation on, the 84-rune default became 2-3 parts and the delays alone exceeded 5s -> ErrDispatchInterrupted -> the log said "New message arrived" when nothing had arrived. Now bounded by noticeCtx (30s). 3. The 90s ceiling was still pinned at 30 inside this repo. .env.example:8 and k8s/configmap.yaml:8 both set it explicitly, and by this PR's own reasoning ("an explicit env beats the code default") the fix had no effect where it runs — the ConfigMap is what actually serves staging/prod. NOT fixed here: evolution-ecosystem/k8s/base/bot-runtime.yaml:44-45 pins 30 too, but that is the SaaS repo, outside these three PRs. 4. The default notice was hardcoded pt-BR (finding 9). bot-runtime ships in community/self-hosted installs worldwide, so customers of installations that never chose Portuguese were answered in it. Now English, with AI_FAILURE_NOTICE documented in .env.example for localisation and the empty value still restoring silence. Tests: 10 (was 6). The new ones pin that the notice leaves the follow-up entry and its state untouched, that its budget fits a segmented dispatch, and that the default carries no pt-BR. Negative proof: restoring the runDispatchStage call fails "DoesNotTouchTheEntryOfTheNextTurn" with the orphaned-turn message. gofmt: pipeline_service.go is already unformatted on develop (CRLF); left as-is rather than reformatting the whole file into this diff. --- .env.example | 11 ++- k8s/configmap.yaml | 5 +- .../service/ai_failure_notice_test.go | 76 +++++++++++++++++++ pkg/pipeline/service/pipeline_service.go | 57 +++++++++++++- 4 files changed, 143 insertions(+), 6 deletions(-) diff --git a/.env.example b/.env.example index a2b6115..eaeba2c 100644 --- a/.env.example +++ b/.env.example @@ -5,7 +5,16 @@ 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: 30s was dimensioned for one well-behaved model call. Measured against +# gemini-2.5-flash, the SAME trivial prompt answered in 0.74s / 0.79s / 1.36s / +# 10.34s / 20.40s - a 27x spread on the provider's tail - and a tool-calling turn +# makes at least two of those round trips. An explicit value here OVERRIDES the +# code default, so leaving 30 kept the fix from having any effect. +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/k8s/configmap.yaml b/k8s/configmap.yaml index d1bee04..821544d 100644 --- a/k8s/configmap.yaml +++ b/k8s/configmap.yaml @@ -5,7 +5,10 @@ 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, so pinning 30 here kept + # the raised ceiling from taking effect - and this ConfigMap is what actually + # 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 index 26bc0fb..be94f35 100644 --- a/pkg/pipeline/service/ai_failure_notice_test.go +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -5,6 +5,7 @@ import ( "errors" "os" "testing" + "time" brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors" "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" @@ -123,3 +124,78 @@ func contains(haystack, needle string) bool { return false })() } + +// --- CRM-236 review, finding 2 ------------------------------------------- +// +// The notice used to be dispatched through runDispatchStage, which owns the +// turn's bookkeeping: its success path runs SetState(StageDone) -> ClearState -> +// entries.Delete(pairKey). That Delete is exactly what runDispatchStage's own +// comments forbid here ("A Delete here would race with the new event's Store and +// could delete the replacement entry"). +// +// The race lands on the scenario this feature targets: the customer waited, gave +// up, and follows up. Their new turn gets stored while the notice is still being +// dispatched; the notice then finishes and deletes THAT entry, orphaning the new +// turn — the next message cannot cancel it, so two pipelines run concurrently on +// the same pair and the customer receives a duplicated reply. + +func TestAIFailureNotice_DoesNotTouchTheEntryOfTheNextTurn(t *testing.T) { + engine, _ := captureDispatch(t) + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + + key := pairKey(1, 2) + replacement := &pipelineEntry{} + svc.entries.Store(key, replacement) + + svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) + + got, ok := svc.entries.Load(key) + if !ok { + t.Fatal("the notice deleted the entry of the follow-up turn: it would be orphaned, " + + "and two pipelines would run on the same pair") + } + if got != replacement { + t.Fatal("the entry was replaced by the notice") + } +} + +func TestAIFailureNotice_DoesNotWriteTurnState(t *testing.T) { + engine, _ := captureDispatch(t) + svc, rdb := 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) + + state, err := svc.repo.GetState(context.Background(), 1, 2) + if err == nil && state != nil { + t.Fatalf("the notice wrote turn state: stage=%v", state.Stage) + } + _ = rdb +} + +// 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 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 5b0a38f..d615d33 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -657,7 +657,11 @@ func cleanupCtx() (context.Context, context.CancelFunc) { // operators who prefer silence to a canned reply. const aiFailureNoticeEnv = "AI_FAILURE_NOTICE" -const defaultAIFailureNotice = "Estou com uma instabilidade no momento e não consegui responder agora. Já retorno." +// English by default: bot-runtime ships in community/self-hosted installs +// worldwide, and a hardcoded pt-BR sentence reached customers of installations +// that never chose Portuguese. Operators localise it with AI_FAILURE_NOTICE +// (documented in .env.example), and an empty value restores the old silence. +const defaultAIFailureNotice = "We are having a temporary issue and could not answer right now. We will get back to you shortly." // sendAIFailureNotice tells the customer something went wrong instead of leaving // the turn silent. @@ -702,10 +706,55 @@ func (s *pipelineService) sendAIFailureNotice( "cause", cause.Error(), ) - // Own context: the pipeline ctx is already cancelled or timed out by now. - ctx, cancel := cleanupCtx() + // Dispatch DIRECTLY — never through runDispatchStage. + // + // CRM-236 review: runDispatchStage owns the turn's bookkeeping. Its success + // path runs SetState(StageDone) → ClearState → entries.Delete(pairKey), and + // its own comments (see the ErrDispatchInterrupted branch above) spell out + // why that Delete must not run here: "A Delete here would race with the new + // event's Store and could delete the replacement entry." + // + // The race is not hypothetical, and it lands on the very scenario this + // feature targets — the customer who waited and follows up. They send "oi?" + // while the notice is being dispatched, startDebounce does entries.Store for + // the new turn, then the notice finishes and deletes THAT entry and clears + // its state. The new turn is orphaned: the next message cannot cancel it, so + // two pipelines run concurrently on the same pair and the customer gets a + // duplicated reply. + // + // There is nothing to book-keep here anyway: both callers already ran + // clearStateWithLog before reaching this function. + ctx, cancel := noticeCtx() defer cancel() - s.runDispatchStage(ctx, contactID, conversationID, notice, cfg, postbackURL) + 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 failure notice's dispatch. +// +// NOT cleanupCtx(): those 5s are documented for cleanup calls (ClearState, +// SetState), and a Dispatch is a different animal — it segments the text by +// TextSegmentationLimit and sleeps DelayPerCharacter per rune between parts +// (dispatch_engine.go). With segmentation on, the 84-rune default notice +// becomes 2-3 parts and the delays alone exceed 5s, so the dispatch was being +// interrupted mid-way. Worse, the old path then logged +// "pipeline.dispatch.cancelled — New message arrived" when no message had +// arrived at all: the operator was told the wrong reason. +func noticeCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 30*time.Second) } func (s *pipelineService) clearStateWithLog(contactID, conversationID int64) { From 54264de5c72614b75feda83d9eac04530f8b322b Mon Sep 17 00:00:00 2001 From: Matheus Pastorini Date: Sat, 22 Aug 2026 16:28:43 -0300 Subject: [PATCH 3/6] chore(pipeline): tidy the review's low findings (CRM-236 review 11-13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 11 — the doc comment of clearStateWithLog had been orphaned: the new constants landed between it and its function, so godoc showed aiFailureNoticeEnv documented as "clearStateWithLog calls ClearState...". Moved back. 12 — the test helper reimplemented strings.Contains in 9 lines with a closure. 13 — os.Unsetenv mutated the process env without restoring it, so every test running after that one inherited the change. Restores via t.Cleanup now. No behaviour change. build/vet clean, full suite green with a real Redis. --- .../service/ai_failure_notice_test.go | 54 +++---------------- pkg/pipeline/service/pipeline_service.go | 6 +-- 2 files changed, 11 insertions(+), 49 deletions(-) diff --git a/pkg/pipeline/service/ai_failure_notice_test.go b/pkg/pipeline/service/ai_failure_notice_test.go index be94f35..5bacc59 100644 --- a/pkg/pipeline/service/ai_failure_notice_test.go +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "os" + "strings" "testing" "time" @@ -59,7 +60,7 @@ func TestAIFailureNotice_NeverLeaksTheProviderError(t *testing.T) { t.Fatalf("expected one notice, got %d", len(*sent)) } for _, leak := range []string{"gemini", "Quota", "RateLimitError", "googleapis"} { - if contains((*sent)[0], leak) { + if strings.Contains((*sent)[0], leak) { t.Errorf("provider detail %q leaked to the customer: %q", leak, (*sent)[0]) } } @@ -103,6 +104,11 @@ func TestAIFailureNotice_NoPostbackUrlIsNotACrash(t *testing.T) { // 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) @@ -114,50 +120,6 @@ func TestAIFailureNotice_DefaultWhenEnvUnset(t *testing.T) { } } -func contains(haystack, needle string) bool { - return len(haystack) >= len(needle) && (func() bool { - for i := 0; i+len(needle) <= len(haystack); i++ { - if haystack[i:i+len(needle)] == needle { - return true - } - } - return false - })() -} - -// --- CRM-236 review, finding 2 ------------------------------------------- -// -// The notice used to be dispatched through runDispatchStage, which owns the -// turn's bookkeeping: its success path runs SetState(StageDone) -> ClearState -> -// entries.Delete(pairKey). That Delete is exactly what runDispatchStage's own -// comments forbid here ("A Delete here would race with the new event's Store and -// could delete the replacement entry"). -// -// The race lands on the scenario this feature targets: the customer waited, gave -// up, and follows up. Their new turn gets stored while the notice is still being -// dispatched; the notice then finishes and deletes THAT entry, orphaning the new -// turn — the next message cannot cancel it, so two pipelines run concurrently on -// the same pair and the customer receives a duplicated reply. - -func TestAIFailureNotice_DoesNotTouchTheEntryOfTheNextTurn(t *testing.T) { - engine, _ := captureDispatch(t) - svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) - - key := pairKey(1, 2) - replacement := &pipelineEntry{} - svc.entries.Store(key, replacement) - - svc.sendAIFailureNotice(1, 2, model.BotConfig{}, "http://crm.test/postback/2", brtErrors.ErrAITimeout) - - got, ok := svc.entries.Load(key) - if !ok { - t.Fatal("the notice deleted the entry of the follow-up turn: it would be orphaned, " + - "and two pipelines would run on the same pair") - } - if got != replacement { - t.Fatal("the entry was replaced by the notice") - } -} func TestAIFailureNotice_DoesNotWriteTurnState(t *testing.T) { engine, _ := captureDispatch(t) @@ -194,7 +156,7 @@ func TestAIFailureNotice_HasRoomForASegmentedDispatch(t *testing.T) { // 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 contains(defaultAIFailureNotice, ptBR) { + 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 d615d33..395a17c 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -649,9 +649,6 @@ func cleanupCtx() (context.Context, context.CancelFunc) { return context.WithTimeout(context.Background(), 5*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. // 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. @@ -757,6 +754,9 @@ 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. func (s *pipelineService) clearStateWithLog(contactID, conversationID int64) { ctx, cancel := cleanupCtx() defer cancel() From 211759fc3f3fb23bb5d9a5f9daeb062556d4ce84 Mon Sep 17 00:00:00 2001 From: Matheus Pastorini Date: Sat, 22 Aug 2026 17:10:56 -0300 Subject: [PATCH 4/6] test(e2e): pipeline isolation now expects the failure notice (CRM-236) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI was red on TestE2E_PipelineIsolation, and the test was right to fail: postback content = "We are having a temporary issue…", want "pair-b response" It asserted callCount == 1 with the comment "only pair B should deliver". That described the OLD behaviour, where pair A's AI error was swallowed — log, clear state, nothing reaching the chat. Making that pair speak is the entire point of this card, so the assertion had to move. Updated rather than relaxed. The test is about isolation, and it now checks that more strictly than before: both pairs deliver, and each must receive ITS OWN message. A crossed delivery fails here even though the count would be right — the previous version could not have caught that. The notice is pinned through AI_FAILURE_NOTICE instead of reaching for the package constant: it keeps test/e2e out of service's internals and exercises the operator-facing env on the way. On how this reached CI: I ran ./pkg/... and ./internal/... and called it "the full suite". test/e2e was never in it. Same mistake as the migration parity spec on CRM-210 — a hand-picked list reported as a green suite. Ran ./... this time. --- test/e2e/e2e_test.go | 39 ++++++++++++++++++++++++++++++++++----- 1 file changed, 34 insertions(+), 5 deletions(-) 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 From 181f246e755e45d59eb5a92895d1163d2a5b9902 Mon Sep 17 00:00:00 2001 From: Guilherme Gomes Date: Sat, 22 Aug 2026 19:45:33 -0300 Subject: [PATCH 5/6] test(pipeline): guard the entry and state of the next turn (CRM-236 review) The round-1 defect was runDispatchStage's success bookkeeping running for the failure notice: SetState(StageDone) -> ClearState -> entries.Delete(pairKey), against a pair that may already belong to a follow-up turn. The fix is in, but nothing pinned it: the existing test read Redis after the fact, and SetState followed by ClearState leaves nothing to read, so it passed with or without the regression. DoesNotTouchTheEntryOfTheNextTurn seeds the pair the way startDebounce leaves it (entry in the map, StageDebounce in Redis) and asserts both survive. Restoring the runDispatchStage call fails all three assertions. Also make DoesNotWriteTurnState fail on a read error instead of passing on it, and drop its unused rdb binding. --- .../service/ai_failure_notice_test.go | 62 +++++++++++++++++-- 1 file changed, 58 insertions(+), 4 deletions(-) diff --git a/pkg/pipeline/service/ai_failure_notice_test.go b/pkg/pipeline/service/ai_failure_notice_test.go index 5bacc59..b921817 100644 --- a/pkg/pipeline/service/ai_failure_notice_test.go +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -120,21 +120,75 @@ func TestAIFailureNotice_DefaultWhenEnvUnset(t *testing.T) { } } - func TestAIFailureNotice_DoesNotWriteTurnState(t *testing.T) { engine, _ := captureDispatch(t) - svc, rdb := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, engine) + 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 && state != nil { + 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) } - _ = rdb +} + +// The entry, not just the state: entries.Delete(pairKey) was the half of +// runDispatchStage's bookkeeping that did the real damage. The customer who +// waited follows up mid-dispatch, startDebounce stores a NEW entry under the +// same key, and deleting it orphans that turn — the next message can no longer +// cancel it, so two pipelines answer the same pair. +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 From 7237ce3bd03b8cac6a24b1623cf2c4c265a2d11a Mon Sep 17 00:00:00 2001 From: Guilherme Gomes Date: Sat, 22 Aug 2026 19:52:56 -0300 Subject: [PATCH 6/6] style(comments): cut the PR narrative out of the code (CRM-236 review 14) The measurement table, the incident timeline and the race walkthrough belong in the PR body, where they already are. In the source they were 11 lines above one call to getEnvIntOrDefault, 18 above one call to Dispatch, and 10 above a context.WithTimeout. What is kept is what the code cannot say: why 90 and not 30, why the notice must not go through runDispatchStage, why noticeCtx is not cleanupCtx. 102 added comment lines to 58, no behaviour touched. --- .env.example | 7 +-- internal/config/config.go | 13 +---- k8s/configmap.yaml | 5 +- .../service/ai_failure_notice_test.go | 15 ++---- pkg/pipeline/service/pipeline_service.go | 51 +++---------------- 5 files changed, 18 insertions(+), 73 deletions(-) diff --git a/.env.example b/.env.example index eaeba2c..8504a71 100644 --- a/.env.example +++ b/.env.example @@ -5,11 +5,8 @@ REDIS_URL=redis://localhost:6379 # An empty value would authenticate any caller that omits the header. BOT_RUNTIME_SECRET= -# CRM-236: 30s was dimensioned for one well-behaved model call. Measured against -# gemini-2.5-flash, the SAME trivial prompt answered in 0.74s / 0.79s / 1.36s / -# 10.34s / 20.40s - a 27x spread on the provider's tail - and a tool-calling turn -# makes at least two of those round trips. An explicit value here OVERRIDES the -# code default, so leaving 30 kept the fix from having any effect. +# 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 diff --git a/internal/config/config.go b/internal/config/config.go index 8a1905a..fa5be62 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,17 +33,8 @@ func Load() (*Config, error) { if err != nil { return nil, err } - // CRM-236: 30s was dimensioned for a single, well-behaved model call. Measured - // against gemini-2.5-flash, the SAME trivial prompt answered in - // 0.74s / 0.79s / 1.36s / 10.34s / 20.40s — a 27x spread on the provider's - // tail. A tool-calling turn makes at least TWO of those round trips (decide the - // tool, then write the reply), so two bad tails alone exceed 30s with nothing - // wrong in the code — and the customer got silence while the tool's side effect - // (a moved pipeline card) had already been applied. - // - // 90s covers two tail-latency calls plus the CRM round trip, and still bounds a - // genuinely hung provider. It is not a licence to hang: the fallback message - // below is what protects the customer when the ceiling IS reached. + // 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 821544d..d3bbbda 100644 --- a/k8s/configmap.yaml +++ b/k8s/configmap.yaml @@ -5,9 +5,8 @@ metadata: data: LISTEN_ADDR: ":8080" AI_PROCESSOR_URL: "http://ai-processor:8000" - # CRM-236: an explicit value OVERRIDES the code default, so pinning 30 here kept - # the raised ceiling from taking effect - and this ConfigMap is what actually - # runs in staging/production. + # 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. diff --git a/pkg/pipeline/service/ai_failure_notice_test.go b/pkg/pipeline/service/ai_failure_notice_test.go index b921817..85a4b1d 100644 --- a/pkg/pipeline/service/ai_failure_notice_test.go +++ b/pkg/pipeline/service/ai_failure_notice_test.go @@ -12,12 +12,8 @@ import ( "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" ) -// CRM-236: when the LLM provider degrades (429 quota, 503 high demand, or just a -// slow tail) the turn exceeds the ceiling. The pipeline used to only log and -// clear state, so NOTHING reached the chat — indistinguishable, for the -// customer, from a bot that is ignoring them. Worse, the tool's side effect may -// already be applied: the pipeline card moves at ~20s and the timeout fires -// later, so the funnel advances while the conversation stays silent. +// 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() @@ -140,11 +136,8 @@ func TestAIFailureNotice_DoesNotWriteTurnState(t *testing.T) { } } -// The entry, not just the state: entries.Delete(pairKey) was the half of -// runDispatchStage's bookkeeping that did the real damage. The customer who -// waited follows up mid-dispatch, startDebounce stores a NEW entry under the -// same key, and deleting it orphans that turn — the next message can no longer -// cancel it, so two pipelines answer the same pair. +// 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) diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index 395a17c..c2983d2 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -654,23 +654,12 @@ func cleanupCtx() (context.Context, context.CancelFunc) { // operators who prefer silence to a canned reply. const aiFailureNoticeEnv = "AI_FAILURE_NOTICE" -// English by default: bot-runtime ships in community/self-hosted installs -// worldwide, and a hardcoded pt-BR sentence reached customers of installations -// that never chose Portuguese. Operators localise it with AI_FAILURE_NOTICE -// (documented in .env.example), and an empty value restores the old silence. +// 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 tells the customer something went wrong instead of leaving -// the turn silent. -// -// CRM-236: when the provider degrades (429 quota, 503 high demand, or a slow -// tail) the turn exceeds the ceiling and the pipeline used to just log and clear -// state. Nothing reached the chat, so the customer could not tell the difference -// between a broken bot and one ignoring them — and the tool's side effect may -// ALREADY be applied (the pipeline card moves at ~20s, the timeout fires later). -// -// The reason is logged for the operator; the customer gets a plain sentence, -// never the provider's raw error (it carries model names, quotas and URLs). +// 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, @@ -703,24 +692,8 @@ func (s *pipelineService) sendAIFailureNotice( "cause", cause.Error(), ) - // Dispatch DIRECTLY — never through runDispatchStage. - // - // CRM-236 review: runDispatchStage owns the turn's bookkeeping. Its success - // path runs SetState(StageDone) → ClearState → entries.Delete(pairKey), and - // its own comments (see the ErrDispatchInterrupted branch above) spell out - // why that Delete must not run here: "A Delete here would race with the new - // event's Store and could delete the replacement entry." - // - // The race is not hypothetical, and it lands on the very scenario this - // feature targets — the customer who waited and follows up. They send "oi?" - // while the notice is being dispatched, startDebounce does entries.Store for - // the new turn, then the notice finishes and deletes THAT entry and clears - // its state. The new turn is orphaned: the next message cannot cancel it, so - // two pipelines run concurrently on the same pair and the customer gets a - // duplicated reply. - // - // There is nothing to book-keep here anyway: both callers already ran - // clearStateWithLog before reaching this function. + // 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) @@ -740,16 +713,8 @@ func (s *pipelineService) sendAIFailureNotice( ) } -// noticeCtx bounds the failure notice's dispatch. -// -// NOT cleanupCtx(): those 5s are documented for cleanup calls (ClearState, -// SetState), and a Dispatch is a different animal — it segments the text by -// TextSegmentationLimit and sleeps DelayPerCharacter per rune between parts -// (dispatch_engine.go). With segmentation on, the 84-rune default notice -// becomes 2-3 parts and the delays alone exceed 5s, so the dispatch was being -// interrupted mid-way. Worse, the old path then logged -// "pipeline.dispatch.cancelled — New message arrived" when no message had -// arrived at all: the operator was told the wrong reason. +// 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) }