fix(telemetry): report each terminal outcome once and stamp usage_context on every event

control-plane/internal/observability/telemetry.go

@@ -28,7 +28,15 @@ import (
 const (
 	defaultTelemetryQueueSize = 256
 	telemetrySubscriberID     = "anonymous-oss-telemetry"
-	telemetrySchemaVersion    = 2
+	telemetrySchemaVersion    = 3
+
+	// telemetryReportedCapacity bounds the set of terminal outcomes this
+	// process remembers having already reported. An execution's outcome is
+	// reported once and never revisited, so the set only needs to span the
+	// window in which a duplicate can still arrive (an SDK's bounded callback
+	// retries, seconds at most). 8192 covers that with room to spare while
+	// keeping the footprint fixed on a control plane that runs for months.
+	telemetryReportedCapacity = 8192
 )
 
 var telemetryVersionPattern = regexp.MustCompile(`^v?[0-9]+(?:\.[0-9]+){0,3}(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$`)
@@ -52,14 +60,17 @@ type telemetrySender func(context.Context, string, time.Duration, TelemetryEvent
 // TelemetryService subscribes to internal event buses and forwards anonymous,
 // low-cardinality usage events. It never forwards raw event payloads.
 type TelemetryService struct {
-	cfg         config.TelemetryConfig
-	storageMode string
-	installHash string
-	eventIDKey  []byte
-	runtimeName string
-	version     string
-	timeout     time.Duration
-	sender      telemetrySender
+	cfg          config.TelemetryConfig
+	storageMode  string
+	installHash  string
+	eventIDKey   []byte
+	runtimeName  string
+	usageContext string
+	version      string
+	timeout      time.Duration
+	sender       telemetrySender
+
+	reported telemetryReportedSet
 
 	queue  chan TelemetryEvent
 	ctx    context.Context
@@ -93,16 +104,19 @@ func NewTelemetryService(cfg config.TelemetryConfig, agentfieldHome, storageMode
 		return nil, err
 	}
 
+	runtimeName := detectRuntime()
+
 	return &TelemetryService{
-		cfg:         cfg,
-		storageMode: normalizeStorageMode(storageMode),
-		installHash: hashInstallID(installID),
-		eventIDKey:  eventIdentityKey(installID),
-		runtimeName: detectRuntime(),
-		version:     emptyTo(version, "unknown"),
-		timeout:     timeout,
-		sender:      sendTelemetryEvent,
-		queue:       make(chan TelemetryEvent, defaultTelemetryQueueSize),
+		cfg:          cfg,
+		storageMode:  normalizeStorageMode(storageMode),
+		installHash:  hashInstallID(installID),
+		eventIDKey:   eventIdentityKey(installID),
+		runtimeName:  runtimeName,
+		usageContext: detectUsageContext(runtimeName),
+		version:      emptyTo(version, "unknown"),
+		timeout:      timeout,
+		sender:       sendTelemetryEvent,
+		queue:        make(chan TelemetryEvent, defaultTelemetryQueueSize),
 	}, nil
 }
 
@@ -195,10 +209,9 @@ func (s *TelemetryService) Start(ctx context.Context) {
 	go s.subscribeExecutionEvents()
 
 	s.Enqueue("control_plane_started", map[string]interface{}{
-		"go_version":    runtime.Version(),
-		"go_os":         runtime.GOOS,
-		"go_arch":       runtime.GOARCH,
-		"usage_context": detectUsageContext(s.runtimeName),
+		"go_version": runtime.Version(),
+		"go_os":      runtime.GOOS,
+		"go_arch":    runtime.GOARCH,
 	})
 	logger.Logger.Info().Msg("anonymous OSS telemetry enabled")
 }
@@ -232,7 +245,7 @@ func (s *TelemetryService) enqueue(eventName string, properties map[string]inter
 		AgentFieldVersion:      s.version,
 		Runtime:                s.runtimeName,
 		StorageMode:            normalizeStorageMode(s.storageMode),
-		Properties:             sanitizeProperties(properties),
+		Properties:             s.withUsageContext(sanitizeProperties(properties)),
 	}
 	if identityMaterial != "" && len(s.eventIDKey) != 0 {
 		event.EventID = s.eventIdentity(eventName, identityMaterial)
@@ -244,6 +257,26 @@ func (s *TelemetryService) enqueue(eventName string, properties map[string]inter
 	}
 }
 
+// withUsageContext stamps every event with where this control plane is
+// running: "ci", "server", or "dev_or_local".
+//
+// It belongs on every event, not just control_plane_started. A CI job starts
+// the control plane on a fresh volume, so it mints a new install ID and its
+// executions are indistinguishable from a real first-time user's — and with
+// the context carried only by the startup event, no downstream query could
+// separate the two after the fact. Stamping it here is what makes
+// execution_completed (and every other event) filterable at the source.
+func (s *TelemetryService) withUsageContext(props map[string]interface{}) map[string]interface{} {
+	if s.usageContext == "" {
+		return props
+	}
+	if props == nil {
+		props = make(map[string]interface{}, 1)
+	}
+	props["usage_context"] = s.usageContext
+	return props
+}
+
 // eventIdentity returns a stable, opaque identifier suitable for idempotent
 // ingestion. Raw execution and workflow identifiers never leave the process.
 func (s *TelemetryService) eventIdentity(eventName, identityMaterial string) string {
@@ -345,7 +378,8 @@ func (s *TelemetryService) handleExecutionEvent(event events.ExecutionEvent) {
 		props["outcome"] = outcome
 	}
 	identityMaterial := ""
-	if hasStableCallbackIdentity(event, outcome) {
+	stable := hasStableCallbackIdentity(event, outcome)
+	if stable {
 		identityMaterial = event.ExecutionID
 	} else {
 		// Updated/timeout transitions can legitimately recur (for example,
@@ -357,9 +391,72 @@ func (s *TelemetryService) handleExecutionEvent(event events.ExecutionEvent) {
 		}
 	}
 	identityMaterial += "\x00" + outcome
+
+	// An execution reaches a given terminal outcome once, so a second event
+	// for one is a re-publish, not a second execution. The bus has produced
+	// those before — a status callback re-delivered after a lost 200 used to
+	// re-run every side effect, publishing another completed event and
+	// counting the execution again — and it stays one refactor away from
+	// producing them again. Suppressing here keeps a republish from becoming
+	// an inflated count even if the ingest side ignores telemetry_event_id.
+	//
+	// Only stable identities are eligible: the random ones above belong to
+	// transitions that are allowed to recur, and collapsing those would lose
+	// real events.
+	if stable && s.reported.observe(eventName+"\x00"+identityMaterial) {
+		logger.Logger.Debug().
+			Str("event", eventName).
+			Msg("anonymous telemetry: outcome already reported for this execution; not sending again")
+		return
+	}
 	s.enqueue(eventName, props, identityMaterial)
 }
 
+// telemetryReportedSet remembers which terminal outcomes have already been
+// reported, so a republished lifecycle event cannot count twice.
+//
+// Capacity is fixed: insertion past the limit evicts the oldest key, which is
+// the right trade for a process that may run for months. A duplicate arrives
+// within seconds of the original, so an eviction can only ever drop a key long
+// past the window in which it could still suppress anything.
+//
+// The zero value is usable and allocates on first observe, so suppression is
+// not something a caller can forget to wire up.
+//
+// Keys are built from raw execution IDs and never leave the process — only the
+// HMAC in eventIdentity is ever sent.
+type telemetryReportedSet struct {
+	mu       sync.Mutex
+	capacity int
+	seen     map[string]struct{}
+	order    []string
+	next     int
+}
+
+// observe records key and reports whether it had already been seen.
+func (s *telemetryReportedSet) observe(key string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.seen == nil {
+		capacity := s.capacity
+		if capacity <= 0 {
+			capacity = telemetryReportedCapacity
+		}
+		s.seen = make(map[string]struct{}, capacity)
+		s.order = make([]string, capacity)
+	}
+	if _, ok := s.seen[key]; ok {
+		return true
+	}
+	if evicted := s.order[s.next]; evicted != "" {
+		delete(s.seen, evicted)
+	}
+	s.order[s.next] = key
+	s.next = (s.next + 1) % len(s.order)
+	s.seen[key] = struct{}{}
+	return false
+}
+
 func hasStableCallbackIdentity(event events.ExecutionEvent, outcome string) bool {
 	if event.ExecutionID == "" {
 		return false