feat: report devices appearing in a fleet

.changeset/growth-signals-devices.md

@@ -0,0 +1,5 @@
+---
+"server": patch
+---
+
+Report devices appearing in an organization's fleet as `gram_activity`. The MDM upsert now reports whether it inserted, which is the only way to tell a first sighting from a re-sighting, and a config's first successful sync is treated as a backfill rather than a stream of new devices.

docs/superpowers/specs/2026-09-03-posthog-slack-notifications-design.md

@@ -277,28 +277,29 @@ points at a different channel.
 
 ### New destinations
 
-All six are created on Slack workspace integration 57009, which already posts
+All five are created on Slack workspace integration 57009, which already posts
 successfully to all three channels. Each is created disabled, tested with a
 sample invocation, then enabled.
 
 | Name                      | Channel                   | Fires on                                                                                                                                                                              |
 | ------------------------- | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | Gram significant          | `#ops-significant-events` | `gram_activity` where `activity` is one of `organization_created`, `user_signed_up`, `member_joined_organization`, `project_created`, `mcp_server_created`, `security_policy_created` |
 | Gram firehose             | `#ops-aicp-events`        | every `gram_activity`                                                                                                                                                                 |
-| Gram MCP settings         | `#ops-all-events`         | `gram_activity` where `activity` is `mcp_server_updated`                                                                                                                              |
 | Gram subscription changes | `#ops-significant-events` | `gram_subscription_changed`                                                                                                                                                           |
 | Gram feature requests     | `#ops-significant-events` | `feature_requested`                                                                                                                                                                   |
 | Gram enterprise gate      | `#ops-significant-events` | `enterprise_gate_viewed`                                                                                                                                                              |
 
 Every MCP server creation reaches the significant channel, not only the first per
-project. MCP settings changes reach both `#ops-aicp-events` (via the firehose)
-and `#ops-all-events` (via its own destination).
+project. MCP settings changes ride the firehose into `#ops-aicp-events` and get
+no destination of their own: the firehose already carries every activity, so a
+second destination would only duplicate them into a channel nobody watches for
+this. `#ops-all-events` therefore receives nothing from this design.
 
 The last three carry over today's behaviour on their own events; they are
 recreated cleanly rather than folded into `gram_activity` because they are
 genuinely different events with different properties.
 
-All six share one message template: actor, activity, org and project, and a
+All five share one message template: actor, activity, org and project, and a
 button linking to the subject in the Gram dashboard.
 
 ## Testing
@@ -322,7 +323,7 @@ button linking to the subject in the Gram dashboard.
 1. Ship the Go change. Events begin flowing to PostHog with no destinations
    consuming them yet, so Slack stays quiet.
 2. Confirm `gram_activity` volume and shape in PostHog over a day.
-3. Create the six new destinations disabled, test-invoke each, enable them.
+3. Create the five new destinations disabled, test-invoke each, enable them.
 4. Disable the eleven old destinations in the same sitting, so there is no
    window of doubled notifications.
 5. After a week of clean signal, delete the eleven disabled destinations.
@@ -340,7 +341,7 @@ button linking to the subject in the Gram dashboard.
 
 ## Appendix: the shared Slack message template
 
-All six destinations use one template, so every Gram notification reads the same
+All five destinations use one template, so every Gram notification reads the same
 way regardless of channel. PostHog's Slack destination templates support hog
 expressions, including `??` and ternaries, which today's destinations already use.
 

server/internal/background/activities.go

@@ -47,6 +47,7 @@ import (
 	"github.com/speakeasy-api/gram/server/internal/externalmcp"
 	"github.com/speakeasy-api/gram/server/internal/feature"
 	"github.com/speakeasy-api/gram/server/internal/functions"
+	"github.com/speakeasy-api/gram/server/internal/growthsignals"
 	"github.com/speakeasy-api/gram/server/internal/guardian"
 	"github.com/speakeasy-api/gram/server/internal/k8s"
 	"github.com/speakeasy-api/gram/server/internal/killswitches"
@@ -375,6 +376,11 @@ func NewActivities(
 
 	conversionPolicyReconciler, _ := openrouterProvisioner.(activities.ConversionPolicyReconciler)
 
+	// Built here rather than threaded in: this constructor already holds every
+	// dependency the emitter needs, and only the device sync reports growth
+	// activity from the worker.
+	growthEmitter := growthsignals.NewEmitter(logger, posthogClient, growthsignals.NewDatabaseEnricher(db), siteURL)
+
 	return &Activities{
 		db:                              db,
 		temporalEnv:                     temporalEnv,
@@ -384,8 +390,8 @@ func NewActivities(
 		collectPlatformUsageMetrics:     activities.NewCollectPlatformUsageMetrics(logger, db),
 		getAIIntegrationsCandidates:     activities.NewGetAIIntegrationsCandidates(logger, db, encryption),
 		pollAIData:                      activities.NewPollAIData(logger, db, encryption, telemetryLogger, guardianPolicy, chatWriter),
-		getDeviceIntegrationCandidates:  activities.NewGetDeviceIntegrationSyncCandidates(logger, meterProvider, db, encryption, guardianPolicy, features),
-		runDeviceIntegrationSync:        activities.NewRunDeviceIntegrationSync(logger, meterProvider, db, encryption, guardianPolicy, features),
+		getDeviceIntegrationCandidates:  activities.NewGetDeviceIntegrationSyncCandidates(logger, meterProvider, db, encryption, guardianPolicy, features, growthEmitter),
+		runDeviceIntegrationSync:        activities.NewRunDeviceIntegrationSync(logger, meterProvider, db, encryption, guardianPolicy, features, growthEmitter),
 		customDomainIngress:             activities.NewCustomDomainIngress(logger, db, k8sClient),
 		customDomainHealth:              activities.NewCustomDomainHealth(logger, db, k8sClient, expectedTargetCNAME, expectedARecords, emailService, siteURL, guardianPolicy),
 		fireOpenRouterCreditsMetrics:    activities.NewFireOpenRouterCreditsMetrics(logger, meterProvider),

server/internal/background/activities/device_integration_sync.go

@@ -14,16 +14,17 @@ import (
 	"github.com/speakeasy-api/gram/server/internal/deviceintegrations"
 	"github.com/speakeasy-api/gram/server/internal/encryption"
 	"github.com/speakeasy-api/gram/server/internal/feature"
+	"github.com/speakeasy-api/gram/server/internal/growthsignals"
 	"github.com/speakeasy-api/gram/server/internal/guardian"
 )
 
 type GetDeviceIntegrationSyncCandidates struct {
 	syncer *deviceintegrations.Syncer
 }
 
-func NewGetDeviceIntegrationSyncCandidates(logger *slog.Logger, meterProvider metric.MeterProvider, db *pgxpool.Pool, encryptionClient *encryption.Client, guardianPolicy *guardian.Policy, features feature.Provider) *GetDeviceIntegrationSyncCandidates {
+func NewGetDeviceIntegrationSyncCandidates(logger *slog.Logger, meterProvider metric.MeterProvider, db *pgxpool.Pool, encryptionClient *encryption.Client, guardianPolicy *guardian.Policy, features feature.Provider, growthEmitter *growthsignals.Emitter) *GetDeviceIntegrationSyncCandidates {
 	return &GetDeviceIntegrationSyncCandidates{
-		syncer: deviceintegrations.NewSyncer(logger, meterProvider, db, encryptionClient, guardianPolicy, features),
+		syncer: deviceintegrations.NewSyncer(logger, meterProvider, db, encryptionClient, guardianPolicy, features, growthEmitter),
 	}
 }
 
@@ -52,9 +53,9 @@ type RunDeviceIntegrationSync struct {
 	syncer *deviceintegrations.Syncer
 }
 
-func NewRunDeviceIntegrationSync(logger *slog.Logger, meterProvider metric.MeterProvider, db *pgxpool.Pool, encryptionClient *encryption.Client, guardianPolicy *guardian.Policy, features feature.Provider) *RunDeviceIntegrationSync {
+func NewRunDeviceIntegrationSync(logger *slog.Logger, meterProvider metric.MeterProvider, db *pgxpool.Pool, encryptionClient *encryption.Client, guardianPolicy *guardian.Policy, features feature.Provider, growthEmitter *growthsignals.Emitter) *RunDeviceIntegrationSync {
 	return &RunDeviceIntegrationSync{
-		syncer: deviceintegrations.NewSyncer(logger, meterProvider, db, encryptionClient, guardianPolicy, features),
+		syncer: deviceintegrations.NewSyncer(logger, meterProvider, db, encryptionClient, guardianPolicy, features, growthEmitter),
 	}
 }
 

server/internal/deviceintegrations/queries.sql

@@ -520,6 +520,7 @@ SELECT
   , s.consecutive_failures
   , s.auto_paused_at
   , s.last_push_digest
+  , s.last_poll_success_at
 FROM device_integration_syncs s
 JOIN device_integration_schedules sch
   ON sch.id = s.device_integration_schedule_id
@@ -599,7 +600,7 @@ RETURNING (s.auto_paused_at IS NOT NULL)::boolean AS auto_paused;
 -- the insert a zero-row no-op, so stale-credential inventory never merges
 -- into the newly saved config.
 
--- name: UpsertMdmDevice :execrows
+-- name: UpsertMdmDevice :one
 INSERT INTO mdm_devices (
     device_integration_config_id
   , organization_id
@@ -642,7 +643,11 @@ ON CONFLICT (device_integration_config_id, external_id) DO UPDATE SET
     raw = EXCLUDED.raw,
     last_seen_at = clock_timestamp(),
     missing_since = NULL,
-    updated_at = clock_timestamp();
+    updated_at = clock_timestamp()
+-- xmax is zero on a freshly inserted row and non-zero on one this statement
+-- updated, which is the only way to tell a first sighting from a re-sighting:
+-- the row count is 1 for both. A guard failure returns no row at all.
+RETURNING (xmax = 0) AS inserted;
 
 -- MarkDevicesMissing stamps devices absent from the snapshot that started at
 -- @sync_started_at. INVARIANT: only ever called in the same transaction that

server/internal/deviceintegrations/sync.go

@@ -24,6 +24,7 @@ import (
 	"github.com/speakeasy-api/gram/server/internal/deviceintegrations/repo"
 	"github.com/speakeasy-api/gram/server/internal/encryption"
 	"github.com/speakeasy-api/gram/server/internal/feature"
+	"github.com/speakeasy-api/gram/server/internal/growthsignals"
 	"github.com/speakeasy-api/gram/server/internal/guardian"
 	"github.com/speakeasy-api/gram/server/internal/o11y"
 	"github.com/speakeasy-api/gram/server/internal/oops"
@@ -108,6 +109,7 @@ type Syncer struct {
 	guardian *guardian.Policy
 	features feature.Provider
 	metrics  *syncMetrics
+	growth   *growthsignals.Emitter
 }
 
 func NewSyncer(
@@ -117,6 +119,7 @@ func NewSyncer(
 	enc *encryption.Client,
 	guardianPolicy *guardian.Policy,
 	features feature.Provider,
+	growthEmitter *growthsignals.Emitter,
 ) *Syncer {
 	componentLogger := logger.With(attr.SlogComponent("deviceintegrations.syncer"))
 	return &Syncer{
@@ -127,6 +130,7 @@ func NewSyncer(
 		guardian: guardianPolicy,
 		features: features,
 		metrics:  newSyncMetrics(componentLogger, meterProvider),
+		growth:   growthEmitter,
 	}
 }
 
@@ -278,6 +282,13 @@ var errStaleSync = errors.New("sync outcome superseded by a config save")
 func (s *Syncer) runInventorySync(ctx context.Context, target repo.GetSyncTargetRow, source providers.InventorySource, creds providers.Credentials, settings providers.Settings, started time.Time) error {
 	cursor := ""
 	memberCache := map[string]pgtype.Text{}
+
+	// A config that has never completed a sync is being backfilled, not
+	// observed: its first snapshot inserts the whole existing fleet at once,
+	// and reporting each row would announce hundreds of devices that were
+	// already there. Only later syncs describe devices genuinely appearing.
+	backfill := !target.LastPollSuccessAt.Valid
+	var firstSeen []deviceSighting
 	for page := 0; ; page++ {
 		if page >= syncMaxPages {
 			return fmt.Errorf("inventory listing exceeded %d pages without completing", syncMaxPages)
@@ -307,7 +318,7 @@ func (s *Syncer) runInventorySync(ctx context.Context, target repo.GetSyncTarget
 			if !device.LastCheckInAt.IsZero() {
 				checkIn = conv.ToPGTimestamptz(device.LastCheckInAt)
 			}
-			rows, err := s.repo.UpsertMdmDevice(ctx, repo.UpsertMdmDeviceParams{
+			inserted, err := s.repo.UpsertMdmDevice(ctx, repo.UpsertMdmDeviceParams{
 				DeviceIntegrationConfigID: target.ConfigID,
 				OrganizationID:            target.OrganizationID,
 				ExternalID:                device.ExternalID,
@@ -322,17 +333,24 @@ func (s *Syncer) runInventorySync(ctx context.Context, target repo.GetSyncTarget
 				ConfigUpdatedAt:           target.ConfigUpdatedAt,
 			})
 			if err != nil {
+				if errors.Is(err, pgx.ErrNoRows) {
+					// The config was saved mid-pull: this run's inventory came
+					// from the pre-save credentials/settings and must not merge
+					// into the new config. Abort; the reset schedule re-runs
+					// promptly with the new configuration.
+					return errStaleSync
+				}
 				if isVendorDataError(err) {
 					return fmt.Errorf("upsert mdm device %s: %w", device.ExternalID, err)
 				}
 				return asInfra(oops.E(oops.CodeUnexpected, err, "upsert mdm device"))
 			}
-			if rows == 0 {
-				// The config was saved mid-pull: this run's inventory came
-				// from the pre-save credentials/settings and must not merge
-				// into the new config. Abort; the reset schedule re-runs
-				// promptly with the new configuration.
-				return errStaleSync
+			if inserted && !backfill && len(firstSeen) < maxDeviceActivitiesPerSync {
+				firstSeen = append(firstSeen, deviceSighting{
+					externalID: device.ExternalID,
+					hostname:   device.Hostname,
+					userEmail:  device.UserEmail,
+				})
 			}
 		}
 		cursor = devicePage.NextCursor
@@ -375,9 +393,77 @@ func (s *Syncer) runInventorySync(ctx context.Context, target repo.GetSyncTarget
 		}
 		return asInfra(oops.E(oops.CodeUnexpected, err, "finalize inventory snapshot"))
 	}
+
+	// Reported only once the snapshot is durable. Emitting inside the loop
+	// would announce devices for a run that later aborts as stale, and the
+	// abort paths above all return before this point.
+	s.reportFirstSeen(ctx, target.OrganizationID, firstSeen)
+
 	return nil
 }
 
+// maxDeviceActivitiesPerSync bounds how many first sightings one snapshot
+// reports. A fleet that doubles overnight is worth one glance at the
+// dashboard, not hundreds of notifications, and the devices themselves are
+// all in the inventory either way.
+const maxDeviceActivitiesPerSync = 10
+
+// propertyDeviceOwnerEmail carries the MDM-assigned user of a device. It is a
+// property rather than the actor because the owner did not do anything: a
+// scheduled sync observed their device.
+const propertyDeviceOwnerEmail = "device_owner_email"
+
+// deviceSighting is a device this sync inserted for the first time. It holds
+// only what the activity reports, so the vendor payload does not travel.
+type deviceSighting struct {
+	externalID string
+	hostname   string
+	userEmail  string
+}
+
+// reportFirstSeen announces devices that appeared in this snapshot. Devices
+// are organization-scoped: no MDM table carries a project, so these activities
+// never claim one.
+//
+// Reporting is at-most-once by design. A sync that inserts rows and then fails
+// on a later page has already committed those rows, so the retry sees them as
+// existing and reports nothing. The alternative — persisting pending sightings
+// so a retry could finish announcing them — buys durability for an ops
+// notification at the cost of a table and its own failure modes. The devices
+// are in the inventory either way; only the Slack line is lost.
+func (s *Syncer) reportFirstSeen(ctx context.Context, organizationID string, sightings []deviceSighting) {
+	for _, sighting := range sightings {
+		name := sighting.hostname
+		if name == "" {
+			name = sighting.externalID
+		}
+
+		// The device's assigned user is its owner, not the actor: nobody
+		// performed this, a scheduled inventory sync observed it. Reporting
+		// them as the actor would attribute the event to them and attach it to
+		// their PostHog person, so the owner travels as a plain property.
+		extra := map[string]string{}
+		if sighting.userEmail != "" {
+			extra[propertyDeviceOwnerEmail] = sighting.userEmail
+		}
+
+		s.growth.Emit(ctx, growthsignals.ActivityEvent{
+			Activity:       growthsignals.ActivityDeviceFirstSeen,
+			OrganizationID: organizationID,
+			ProjectID:      uuid.Nil,
+			ActorID:        "",
+			ActorType:      "",
+			ActorEmail:     "",
+			ActorName:      "",
+			SubjectName:    name,
+			ActingSurface:  "",
+			AuditAction:    "",
+			DashboardURL:   "",
+			Extra:          extra,
+		})
+	}
+}
+
 // runEvidencePush builds the org's coverage snapshot and delivers it to the
 // sink, unless the snapshot digest matches the last successful push — an
 // unchanged fleet is a free no-op.