From 061ad621358e3b01da744d72d8ca3620aefd1b30 Mon Sep 17 00:00:00 2001
From: Xeron <xeron2333@gmail.com>
Date: Sat, 8 Aug 2026 00:57:29 +0800
Subject: [PATCH] fix(engine): enrich only newly created opportunity sessions

HandlePulseEvent returns (nil, nil) for a duplicate pulse EventID, but the
aggregator treated nil error as 'enrich again', spawning a second watch
goroutine on the same session. Enrich only when a session was actually
created; regression test counts enrichment spawns for a duplicated event.
---
 internal/engine/pipeline.go    | 16 ++++++++++---
 internal/engine/policy_test.go | 41 ++++++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 3 deletions(-)

diff --git a/internal/engine/pipeline.go b/internal/engine/pipeline.go
index d67e246..3290496 100644
--- a/internal/engine/pipeline.go
+++ b/internal/engine/pipeline.go
@@ -42,6 +42,12 @@ import (
 // exchangeNew creates exchange adapters; overridden in tests.
 var exchangeNew = exchange.New
 
+// enrichPulse spawns async OHLCV enrichment for a pulse event; a var so tests
+// can count spawns (duplicate events must not enrich twice).
+var enrichPulse = func(p *Pipeline, evt types.AnomalyEvent) {
+	p.enrichOpportunityAsync(evt)
+}
+
 // ────────────────────────────────────────────────────────────────
 // Pipeline
 // ────────────────────────────────────────────────────────────────
@@ -1052,11 +1058,15 @@ func (p *Pipeline) eventAggregator(
 			}
 
 			// One OpportunitySession per pulse event; async OHLCV enrichment may attach tickets.
+			// HandlePulseEvent returns (nil, nil) for duplicate events — only enrich
+			// when a session was actually created, or the same session would get a
+			// second watch goroutine.
 			if isMarketPulseEvent(evt.EventType) && p.opportunity != nil {
-				if _, err := p.opportunity.HandlePulseEvent(evt); err != nil {
+				sess, err := p.opportunity.HandlePulseEvent(evt)
+				if err != nil {
 					p.log.Warn("opportunity session failed", "error", err, "event", evt.EventType)
-				} else {
-					p.enrichOpportunityAsync(evt)
+				} else if sess != nil {
+					enrichPulse(p, evt)
 				}
 			}
 
diff --git a/internal/engine/policy_test.go b/internal/engine/policy_test.go
index bea38bf..ca71ea1 100644
--- a/internal/engine/policy_test.go
+++ b/internal/engine/policy_test.go
@@ -291,3 +291,44 @@ func (m *mockPipelineExchange) FetchOHLCV(context.Context, string, string, int,
 	return nil, nil
 }
 func (m *mockPipelineExchange) Close() error { return nil }
+
+func TestEventAggregator_DuplicatePulseEnrichesOnce(t *testing.T) {
+	cfg := &types.Config{
+		Opportunity: types.OpportunityConfig{Enabled: true},
+		Storage:     types.StorageConfig{DatabasePath: filepath.Join(t.TempDir(), "kairos.db")},
+	}
+	p := NewPipeline(cfg, nil)
+	if p.opportunity == nil {
+		t.Fatal("opportunity service must be wired when enabled")
+	}
+	calls := 0
+	old := enrichPulse
+	enrichPulse = func(_ *Pipeline, _ types.AnomalyEvent) { calls++ }
+	defer func() { enrichPulse = old }()
+
+	delivery := make(chan types.AnomalyEvent, 2)
+	src := make(chan types.AnomalyEvent, 2)
+	evt := types.AnomalyEvent{
+		EventType: "market_impulse",
+		EventID:   "e-dup",
+		Timestamp: float64(time.Now().Unix()),
+		Data:      map[string]any{"direction": "up", "state_to": "IMPULSE_UP", "leaders": []string{"SOL/USDT:USDT"}},
+	}
+	src <- evt
+	src <- evt // same EventID again — session already exists
+	close(src)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go p.eventAggregator(ctx, []eventSource{{ch: src, origin: "okx"}}, delivery)
+	for i := 0; i < 2; i++ {
+		select {
+		case <-delivery:
+		case <-time.After(2 * time.Second):
+			t.Fatal("timeout waiting for delivery")
+		}
+	}
+	if calls != 1 {
+		t.Fatalf("duplicate pulse must enrich exactly once, got %d", calls)
+	}
+}