feat: extend AI debounce timer on an inbound typing/presence signal

pkg/debounce/service/debounce_engine.go

@@ -14,6 +14,10 @@ import (
 type DebounceEngine interface {
 	Start(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error
 	Reset(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error
+	// RenewTimer extends the debounce timer TTL without touching the buffer or
+	// attachments — used when a presence ("customer is typing") signal arrives
+	// mid-debounce, as opposed to Reset which is for an actual new message.
+	RenewTimer(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig) error
 	GetBuffer(ctx context.Context, contactID, conversationID int64) (string, error)
 	GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error)
 	TimerExists(ctx context.Context, contactID, conversationID int64) (bool, error)
@@ -35,13 +39,7 @@ func (d *debounceEngine) Start(ctx context.Context, contactID, conversationID in
 	if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil {
 		return fmt.Errorf("debounce.start.append_attachments: %w", err)
 	}
-	if cfg.DebounceTime > 0 {
-		ttl := time.Duration(cfg.DebounceTime) * time.Second
-		if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil {
-			return fmt.Errorf("debounce.start.set_timer: %w", err)
-		}
-	}
-	return nil
+	return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "start")
 }
 
 func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error {
@@ -51,15 +49,48 @@ func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID in
 	if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil {
 		return fmt.Errorf("debounce.reset.append_attachments: %w", err)
 	}
-	if cfg.DebounceTime > 0 {
-		ttl := time.Duration(cfg.DebounceTime) * time.Second
-		if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil {
-			return fmt.Errorf("debounce.reset.set_timer: %w", err)
-		}
+	return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "reset")
+}
+
+// RenewTimer extends an already-running timer only — unlike Start/Reset, it
+// must never create one. A presence pulse landing in the ~100ms window after
+// pollDebounceExpiry sees the key gone (but before it advances the pair to
+// StageAI) must not resurrect that window and indefinitely postpone the reply
+// (Sourcery finding on evolution-foundation/evo-bot-runtime#12).
+func (d *debounceEngine) RenewTimer(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig) error {
+	ttl, ok := d.timerTTLIfConfigured(cfg)
+	if !ok {
+		return nil
+	}
+	if _, err := d.repo.RenewTimerIfExists(ctx, contactID, conversationID, ttl); err != nil {
+		return fmt.Errorf("debounce.renew_timer.renew_timer_if_exists: %w", err)
 	}
 	return nil
 }
 
+// setTimerIfConfigured is the timer-setting logic shared by Start, Reset and
+// RenewTimer — a no-op when DebounceTime <= 0 (pass-through mode).
+func (d *debounceEngine) setTimerIfConfigured(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig, op string) error {
+	ttl, ok := d.timerTTLIfConfigured(cfg)
+	if !ok {
+		return nil
+	}
+	if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil {
+		return fmt.Errorf("debounce.%s.set_timer: %w", op, err)
+	}
+	return nil
+}
+
+// timerTTLIfConfigured is the DebounceTime<=0 pass-through check shared by
+// every timer-setting path (Start, Reset, RenewTimer) — ok is false when
+// there's no timer to set/renew at all.
+func (d *debounceEngine) timerTTLIfConfigured(cfg model.BotConfig) (ttl time.Duration, ok bool) {
+	if cfg.DebounceTime <= 0 {
+		return 0, false
+	}
+	return time.Duration(cfg.DebounceTime) * time.Second, true
+}
+
 func (d *debounceEngine) GetBuffer(ctx context.Context, contactID, conversationID int64) (string, error) {
 	entries, err := d.repo.GetBuffer(ctx, contactID, conversationID)
 	if err != nil {

pkg/debounce/service/debounce_engine_test.go

@@ -152,3 +152,68 @@ func TestTimerExists_ReturnsFalseAfterExpiry(t *testing.T) {
 		t.Error("timer must be expired after TTL")
 	}
 }
+
+func TestRenewTimer_ExtendsTTLWithoutTouchingBuffer(t *testing.T) {
+	eng, rdb := setupEngine(t)
+	ctx := context.Background()
+	cfg := model.BotConfig{DebounceTime: 5}
+
+	if err := eng.Start(ctx, 6, 6, "first message", nil, cfg); err != nil {
+		t.Fatalf("Start returned error: %v", err)
+	}
+
+	// Let some of the TTL elapse so renewal is observable.
+	time.Sleep(200 * time.Millisecond)
+
+	if err := eng.RenewTimer(ctx, 6, 6, cfg); err != nil {
+		t.Fatalf("RenewTimer returned error: %v", err)
+	}
+
+	ttl := rdb.TTL(ctx, "bot_runtime:timer:6:6").Val()
+	if ttl <= 4*time.Second || ttl > 5*time.Second {
+		t.Errorf("timer TTL should be renewed to ~5s, got %v", ttl)
+	}
+
+	buffer, err := eng.GetBuffer(ctx, 6, 6)
+	if err != nil {
+		t.Fatalf("GetBuffer returned error: %v", err)
+	}
+	if buffer != "first message" {
+		t.Errorf("buffer should be unchanged by RenewTimer, got %q", buffer)
+	}
+}
+
+func TestRenewTimer_NoopWhenDebounceTimeIsZero(t *testing.T) {
+	eng, rdb := setupEngine(t)
+	ctx := context.Background()
+
+	if err := eng.RenewTimer(ctx, 7, 7, model.BotConfig{DebounceTime: 0}); err != nil {
+		t.Fatalf("RenewTimer returned error: %v", err)
+	}
+
+	exists := rdb.Exists(ctx, "bot_runtime:timer:7:7").Val()
+	if exists != 0 {
+		t.Errorf("no timer key should have been created for DebounceTime=0")
+	}
+}
+
+// Sourcery finding on PR #12: RenewTimer must never resurrect a timer key
+// that has already expired (or was never started) — otherwise a presence
+// pulse landing in the poller's detection window could indefinitely
+// postpone a reply the pipeline had already decided to send.
+func TestRenewTimer_DoesNotResurrectAnExpiredTimer(t *testing.T) {
+	eng, rdb := setupEngine(t)
+	ctx := context.Background()
+	cfg := model.BotConfig{DebounceTime: 5}
+
+	// No Start() call for this pair — the timer key was never created
+	// (equivalent to: it already expired and nothing has recreated it yet).
+	if err := eng.RenewTimer(ctx, 8, 8, cfg); err != nil {
+		t.Fatalf("RenewTimer returned error: %v", err)
+	}
+
+	exists := rdb.Exists(ctx, "bot_runtime:timer:8:8").Val()
+	if exists != 0 {
+		t.Error("RenewTimer must not create a timer key that didn't already exist")
+	}
+}

pkg/pipeline/handler/handler.go

@@ -25,6 +25,7 @@ func NewHandler(repo repository.PipelineRepository, svc pipelineService.Pipeline
 
 func (h *Handler) RegisterRoutes(r *gin.Engine) {
 	r.POST("/events", SecretMiddleware(h.secret), h.handleEvent)
+	r.POST("/events/presence", SecretMiddleware(h.secret), h.handlePresence)
 	r.GET("/health", h.handleHealth)
 }
 
@@ -39,20 +40,35 @@ func (h *Handler) handleHealth(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{"status": "ok"})
 }
 
-func (h *Handler) handleEvent(c *gin.Context) {
-	var event model.MessageEvent
-	if err := c.ShouldBindJSON(&event); err != nil {
+// validatable is any inbound event with post-JSON-bind semantic checks.
+type validatable interface {
+	Validate() error
+}
+
+// bindAndValidate binds the request body into event and runs its Validate.
+// On failure it writes the 400 response itself and returns false — the
+// caller returns immediately without doing any further work.
+func bindAndValidate[T validatable](c *gin.Context, event T) bool {
+	if err := c.ShouldBindJSON(event); err != nil {
 		c.JSON(http.StatusBadRequest, gin.H{
 			"error": "invalid payload",
 			"code":  "ERR_INVALID_EVENT",
 		})
-		return
+		return false
 	}
 	if err := event.Validate(); err != nil {
 		c.JSON(http.StatusBadRequest, gin.H{
 			"error": err.Error(),
 			"code":  "ERR_INVALID_EVENT",
 		})
+		return false
+	}
+	return true
+}
+
+func (h *Handler) handleEvent(c *gin.Context) {
+	var event model.MessageEvent
+	if !bindAndValidate(c, &event) {
 		return
 	}
 
@@ -77,6 +93,29 @@ func (h *Handler) handleEvent(c *gin.Context) {
 	c.JSON(http.StatusAccepted, gin.H{"status": "accepted"})
 }
 
+// handlePresence renews an in-flight debounce timer when the customer is
+// composing/recording. Unlike handleEvent, it does NOT persist StageIncoming
+// before processing — a presence pulse must never overwrite an active
+// StageAI/StageDispatch state, only ever renew StageDebounce. It always
+// responds 202 on a structurally valid payload: "accepted and renewed" and
+// "accepted and no-op" are not distinguished to the caller.
+func (h *Handler) handlePresence(c *gin.Context) {
+	var event model.PresenceEvent
+	if !bindAndValidate(c, &event) {
+		return
+	}
+
+	if err := h.svc.Presence(c.Request.Context(), event.ContactID, event.ConversationID); err != nil {
+		slog.Warn("pipeline.presence.failed",
+			"contact_id", event.ContactID,
+			"conversation_id", event.ConversationID,
+			"error", err,
+		)
+	}
+
+	c.JSON(http.StatusAccepted, gin.H{"status": "accepted"})
+}
+
 func (h *Handler) runProcess(event model.MessageEvent) {
 	if err := h.svc.Process(context.Background(), &event); err != nil {
 		slog.Error("pipeline.process.failed",

pkg/pipeline/handler/handler_test.go

@@ -46,6 +46,9 @@ func (m *mockRepo) GetAttachments(_ context.Context, _, _ int64) ([]model.Attach
 	return nil, nil
 }
 func (m *mockRepo) SetTimer(_ context.Context, _, _ int64, _ time.Duration) error { return nil }
+func (m *mockRepo) RenewTimerIfExists(_ context.Context, _, _ int64, _ time.Duration) (bool, error) {
+	return true, nil
+}
 func (m *mockRepo) DeleteTimer(_ context.Context, _, _ int64) error               { return nil }
 func (m *mockRepo) TimerExists(_ context.Context, _, _ int64) (bool, error)       { return false, nil }
 func (m *mockRepo) AcquireLock(_ context.Context, _, _ int64) (repository.Mutex, error) {
@@ -57,12 +60,20 @@ func (m *mockRepo) ScanStates(_ context.Context, _ int) ([]model.PairID, error)
 func (m *mockRepo) Ping(_ context.Context) error { return m.pingErr }
 
 // mockSvc satisfies pipelineService.PipelineService for handler tests.
-type mockSvc struct{ processErr error }
+type mockSvc struct {
+	processErr    error
+	presenceErr   error
+	presenceCalls [][2]int64
+}
 
 func (m *mockSvc) Process(_ context.Context, _ *model.MessageEvent) error { return m.processErr }
-func (m *mockSvc) Cancel(_, _ int64) error                                { return nil }
-func (m *mockSvc) Start() error                                           { return nil }
-func (m *mockSvc) Shutdown(_ context.Context)                             {}
+func (m *mockSvc) Presence(_ context.Context, contactID, conversationID int64) error {
+	m.presenceCalls = append(m.presenceCalls, [2]int64{contactID, conversationID})
+	return m.presenceErr
+}
+func (m *mockSvc) Cancel(_, _ int64) error    { return nil }
+func (m *mockSvc) Start() error               { return nil }
+func (m *mockSvc) Shutdown(_ context.Context) {}
 
 var _ pipelineService.PipelineService = (*mockSvc)(nil)
 
@@ -318,3 +329,63 @@ func TestHealth_503_WhenRedisUnreachable(t *testing.T) {
 		t.Errorf("body.detail: got %q, want %q", body["detail"], "redis unreachable")
 	}
 }
+
+func TestHandlePresence_202_OnValidRequest(t *testing.T) {
+	svc := &mockSvc{}
+	r := setupRouter(&mockRepo{}, svc)
+
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":1,"conversation_id":1}`)))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Bot-Runtime-Secret", testSecret)
+	r.ServeHTTP(w, req)
+
+	if w.Code != http.StatusAccepted {
+		t.Errorf("status: got %d, want %d", w.Code, http.StatusAccepted)
+	}
+	if len(svc.presenceCalls) != 1 || svc.presenceCalls[0] != [2]int64{1, 1} {
+		t.Errorf("expected Presence(1, 1) to have been called, got %v", svc.presenceCalls)
+	}
+}
+
+func TestHandlePresence_202_EvenWhenPresenceFails(t *testing.T) {
+	// Presence failures are best-effort (see Global Constraints) — must never
+	// surface as a non-2xx to the caller.
+	svc := &mockSvc{presenceErr: errors.New("redis unavailable")}
+	r := setupRouter(&mockRepo{}, svc)
+
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":1,"conversation_id":1}`)))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Bot-Runtime-Secret", testSecret)
+	r.ServeHTTP(w, req)
+
+	if w.Code != http.StatusAccepted {
+		t.Errorf("status: got %d, want %d", w.Code, http.StatusAccepted)
+	}
+}
+
+func TestHandlePresence_400_OnMissingRequiredFields(t *testing.T) {
+	r := setupRouter(&mockRepo{}, &mockSvc{})
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":0,"conversation_id":1}`)))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Bot-Runtime-Secret", testSecret)
+	r.ServeHTTP(w, req)
+
+	if w.Code != http.StatusBadRequest {
+		t.Errorf("status: got %d, want %d", w.Code, http.StatusBadRequest)
+	}
+}
+
+func TestHandlePresence_401_OnMissingSecret(t *testing.T) {
+	r := setupRouter(&mockRepo{}, &mockSvc{})
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":1,"conversation_id":1}`)))
+	req.Header.Set("Content-Type", "application/json")
+	r.ServeHTTP(w, req)
+
+	if w.Code != http.StatusUnauthorized {
+		t.Errorf("status: got %d, want %d", w.Code, http.StatusUnauthorized)
+	}
+}

pkg/pipeline/model/pipeline.go

@@ -73,14 +73,23 @@ type BotConfig struct {
 	DelayPerCharacter       float64 `json:"delay_per_character"` // ms per char between parts
 }
 
-// Validate checks semantic constraints on a MessageEvent after JSON binding.
-func (e *MessageEvent) Validate() error {
-	if e.ContactID <= 0 {
+// validatePairID checks the ContactID/ConversationID constraints shared by
+// every event type keyed on a (contact, conversation) pair.
+func validatePairID(contactID, conversationID int64) error {
+	if contactID <= 0 {
 		return errors.New("contact_id must be > 0")
 	}
-	if e.ConversationID <= 0 {
+	if conversationID <= 0 {
 		return errors.New("conversation_id must be > 0")
 	}
+	return nil
+}
+
+// Validate checks semantic constraints on a MessageEvent after JSON binding.
+func (e *MessageEvent) Validate() error {
+	if err := validatePairID(e.ContactID, e.ConversationID); err != nil {
+		return err
+	}
 	if e.PostbackURL == "" {
 		return errors.New("postback_url is required")
 	}
@@ -95,3 +104,16 @@ func (e *MessageEvent) Validate() error {
 	}
 	return nil
 }
+
+// PresenceEvent is the inbound "customer is typing" signal from evo-ai-crm's
+// presence webhook ingestion. Unlike MessageEvent, it carries no content —
+// it only ever renews an already-running debounce timer.
+type PresenceEvent struct {
+	ContactID      int64 `json:"contact_id"`
+	ConversationID int64 `json:"conversation_id"`
+}
+
+// Validate checks semantic constraints on a PresenceEvent after JSON binding.
+func (e *PresenceEvent) Validate() error {
+	return validatePairID(e.ContactID, e.ConversationID)
+}

pkg/pipeline/model/presence_event_test.go

@@ -0,0 +1,28 @@
+package model_test
+
+import (
+	"testing"
+
+	"github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model"
+)
+
+func TestPresenceEventValidate_ValidEvent(t *testing.T) {
+	e := model.PresenceEvent{ContactID: 1, ConversationID: 1}
+	if err := e.Validate(); err != nil {
+		t.Errorf("expected no error, got %v", err)
+	}
+}
+
+func TestPresenceEventValidate_RejectsNonPositiveContactID(t *testing.T) {
+	e := model.PresenceEvent{ContactID: 0, ConversationID: 1}
+	if err := e.Validate(); err == nil {
+		t.Error("expected error for contact_id <= 0, got nil")
+	}
+}
+
+func TestPresenceEventValidate_RejectsNonPositiveConversationID(t *testing.T) {
+	e := model.PresenceEvent{ContactID: 1, ConversationID: 0}
+	if err := e.Validate(); err == nil {
+		t.Error("expected error for conversation_id <= 0, got nil")
+	}
+}