test(e2e): add full e2e test suite and fix three production bugs found during testing

E2E tests (test/e2e/e2e_test.go):
- FullHappyPath: validates end-to-end wiring (PostbackURL, BotConfig propagation)
- DebounceAggregation: verifies buffer concatenation through the full HTTP→AI chain
- AIInterruption: exercises real HTTP cancellation integrated with pipeline context
- ResponseTime: validates NFR-05 (202 < 1s regardless of pipeline duration)
- ExactlyOnce: exercises Redlock under real goroutine pressure
- PipelineIsolation: verifies one pair's failure does not affect another
- DispatchSegmentation: validates BotConfig segmentation wiring to dispatch
- DispatchInterruption: verifies mid-dispatch cancellation drops remaining parts
- RecoveryAfterRestart: validates NFR-01 — debounce pairs survive a restart
- PostbackFailure: verifies state is cleared and no retry on non-2xx postback

Production bug fixes surfaced by the tests:
- PipelineState now persists BotConfig and PostbackURL at StageDebounce so
  Start() recovery can reconstruct pipelineEntry correctly after restart (NFR-01)
- Removed entries.Delete from ErrDispatchInterrupted handler — it raced with the
  replacement event's Store and deleted the new entry before launchAIStage loaded it
- Shutdown is now idempotent via sync.Once (close on closed channel panicked)

Test infrastructure:
- E2E suite uses Redis DB 14; unit tests use DB 15 — prevents FlushDB interference
  when go test ./... runs packages concurrently

Author: gomessguii

pkg/pipeline/model/pipeline.go

@@ -24,9 +24,13 @@ type PairID struct {
 
 // PipelineState is what is stored in Redis (JSON-serializable).
 // The cancel func is NOT stored here — it lives in PipelineService memory (Story 2.2).
+// BotConfig and PostbackURL are persisted for StageDebounce so that the service can
+// reconstruct the pipelineEntry correctly after a restart (NFR-01 recovery).
 type PipelineState struct {
-	Stage     Stage     `json:"stage"`
-	CreatedAt time.Time `json:"created_at"`
+	Stage       Stage     `json:"stage"`
+	CreatedAt   time.Time `json:"created_at"`
+	BotConfig   BotConfig `json:"bot_config,omitempty"`
+	PostbackURL string    `json:"postback_url,omitempty"`
 }
 
 // MessageEvent is the inbound payload from evo-ai-crm AgentBotListener.

pkg/pipeline/service/pipeline_service.go

@@ -45,6 +45,7 @@ type pipelineService struct {
 	entries     sync.Map      // string → pipelineEntry
 	stopCh      chan struct{}  // closed by Shutdown to stop pollDebounceExpiry
 	stoppedCh   chan struct{}  // closed by pollDebounceExpiry when it exits
+	stopOnce    sync.Once     // ensures stopCh is closed exactly once
 }
 
 // NewPipelineService constructs the service. Returns interface (GEAR R03).
@@ -86,7 +87,12 @@ func (s *pipelineService) Start() error {
 		key := pairKey(pair.ContactID, pair.ConversationID)
 		if _, alreadyExists := s.entries.Load(key); !alreadyExists {
 			pipelineCtx, cancel := context.WithCancel(context.Background())
-			s.entries.Store(key, pipelineEntry{ctx: pipelineCtx, cancel: cancel})
+			s.entries.Store(key, pipelineEntry{
+				ctx:         pipelineCtx,
+				cancel:      cancel,
+				cfg:         state.BotConfig,
+				postbackURL: state.PostbackURL,
+			})
 		}
 
 		// If timer already expired during the restart window → advance immediately.
@@ -115,6 +121,12 @@ func (s *pipelineService) Process(ctx context.Context, event *model.MessageEvent
 
 	switch {
 	case state == nil || state.Stage == model.StageIncoming:
+		// The HTTP handler always writes StageIncoming before launching the Process
+		// goroutine (NFR-01 durability). When two events for the same pair arrive
+		// rapidly, the second handler call overwrites any active StageAI/StageDispatch
+		// with StageIncoming before this goroutine runs — bypassing the interrupt
+		// branch below. cancelPair is a no-op when no entry exists.
+		s.cancelPair(event.ContactID, event.ConversationID)
 		if event.BotConfig.DebounceTime == 0 {
 			return s.skipDebounce(ctx, event)
 		}
@@ -154,7 +166,14 @@ func (s *pipelineService) startDebounce(ctx context.Context, event *model.Messag
 		return fmt.Errorf("pipeline.debounce.start: %w", err)
 	}
 
-	newState := &model.PipelineState{Stage: model.StageDebounce, CreatedAt: time.Now()}
+	// Persist BotConfig and PostbackURL alongside the stage so that Start()
+	// recovery after a restart can reconstruct the pipelineEntry correctly (NFR-01).
+	newState := &model.PipelineState{
+		Stage:       model.StageDebounce,
+		CreatedAt:   time.Now(),
+		BotConfig:   event.BotConfig,
+		PostbackURL: event.PostbackURL,
+	}
 	if err := s.repo.SetState(ctx, event.ContactID, event.ConversationID, newState); err != nil {
 		cancel()
 		s.entries.Delete(key)
@@ -402,14 +421,13 @@ func (s *pipelineService) runDispatchStage(
 		case errors.Is(err, brtErrors.ErrDispatchInterrupted):
 			// New message arrived — Process already set StageDebounce.
 			// Do NOT call ClearState: would destroy the new active state.
+			// Do NOT call entries.Delete: cancelPair already did LoadAndDelete
+			// atomically. A Delete here would race with the new event's Store
+			// and could delete the replacement entry.
 			slog.Info("pipeline.dispatch.cancelled",
 				"contact_id",      contactID,
 				"conversation_id", conversationID,
 			)
-			// Entry already removed by cancelPair (LoadAndDelete) — this delete is
-			// idempotent; it guards against future refactors where cancelPair may not
-			// delete the entry from s.entries.
-			s.entries.Delete(pairKey(contactID, conversationID))
 		default:
 			slog.Error("pipeline.dispatch.error",
 				"contact_id",      contactID,
@@ -516,6 +534,8 @@ func (s *pipelineService) Cancel(contactID, conversationID int64) error {
 
 // Shutdown stops the polling goroutine and cancels all in-flight pipeline
 // contexts. It blocks until the poller exits or ctx is cancelled.
+// Safe to call multiple times — subsequent calls are no-ops for the stop signal
+// but still wait for the poller to finish if it hasn't already.
 func (s *pipelineService) Shutdown(ctx context.Context) {
 	// Cancel all in-flight pipeline goroutines.
 	s.entries.Range(func(k, v any) bool {
@@ -525,8 +545,9 @@ func (s *pipelineService) Shutdown(ctx context.Context) {
 		return true
 	})
 
-	// Signal poller to stop and wait for it to exit.
-	close(s.stopCh)
+	// Signal poller to stop exactly once — close on a closed channel panics.
+	s.stopOnce.Do(func() { close(s.stopCh) })
+
 	select {
 	case <-s.stoppedCh:
 		slog.Info("pipeline.shutdown.complete")