feat(monitoring): generate daily tasks and optimize heartbeat writes

Adds a MonitoringDailyTaskWorker that materializes and dispatches daily collect tasks from active subscription/desktop-client state, with per-run atomic metrics and a Prometheus collector. Dashboard and question-detail APIs now surface a platform_authorization_status so the UI can distinguish no-desktop-client, no-authorized-platforms, and authorized states; quota/access-state lookups resolve by request workspace instead of tenant. Hardens heartbeat: a desktop_client_primary_leases table gives per-(tenant, workspace) sticky primary selection, read-mostly lease resolution avoids a per-heartbeat transaction, and unchanged platform access reports skip snapshot upserts outside a 10-minute refresh window. Adds heartbeat primary/snapshot metrics exposed via token-protected tenant-api /api/internal/metrics endpoints plus Prometheus. Monitoring quota seed is removed from dev-seed since daily plans now derive from desktop bindings, and the projection uses the canonical six-platform catalog so platforms with zero samples still render. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-24 22:19:57 +08:00
parent ffe4d335aa
commit ff86933c24
26 changed files with 3546 additions and 277 deletions
@@ -30,6 +30,14 @@ const (
 	maxMonitoringLeaseLimit            = 50
 	monitoringLeaseTTL                 = 15 * time.Minute
 	monitoringTaskResultQueueIOTimeout = 5 * time.Second
+	// The current primary renews the DB lease only after this interval. With the
+	// desktop client's 25s heartbeat and a 90s lease TTL, this keeps normal
+	// heartbeats read-only while still tolerating one missed heartbeat.
+	monitoringHeartbeatPrimaryRenewAfter = desktopClientPresenceTTL / 3
+	// Access snapshots are semantic state, not a liveness clock. Stable heartbeat
+	// reports skip writes and only refresh the row at this interval so dashboards
+	// do not depend on per-heartbeat database churn.
+	monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute
 )

 type MonitoringCallbackService struct {
@@ -205,6 +213,16 @@ type monitoringPlatformAccessSnapshotState struct {
 	Accessible    bool
 	ReasonCode    *string
 	ReasonMessage *string
+	UpdatedAt     time.Time
+}
+
+type monitoringHeartbeatSnapshotDecision struct {
+	WriteSnapshot             bool
+	StateChanged              bool
+	RefreshDue                bool
+	PendingReconcileCandidate bool
+	ReconcilePending          bool
+	SnapshotOperation         string
 }

 type monitoringCollectTask struct {
@@ -434,7 +452,18 @@ func (s *MonitoringCallbackService) handleTaskResultForInstallation(
 	}, nil
 }

-func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID string, installationToken string, req MonitoringHeartbeatRequest) (*MonitoringHeartbeatResponse, error) {
+func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID string, installationToken string, req MonitoringHeartbeatRequest) (_ *MonitoringHeartbeatResponse, err error) {
+	startedAt := time.Now()
+	snapshotMetrics := monitoringHeartbeatSnapshotMetricsEvent{}
+	defer func() {
+		if !snapshotMetrics.Active {
+			return
+		}
+		snapshotMetrics.Duration = time.Since(startedAt)
+		snapshotMetrics.Err = err
+		recordMonitoringHeartbeatSnapshotMetrics(snapshotMetrics)
+	}()
+
 	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
 	if err != nil {
 		return nil, err
@@ -451,28 +480,111 @@ func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationI
 	if err != nil {
 		return nil, err
 	}
+	snapshotMetrics.Active = true
+	snapshotMetrics.PlatformCount = len(platforms)
+
+	if len(platforms) == 0 {
+		_ = s.touchInstallation(ctx, installation.ID)
+		return &MonitoringHeartbeatResponse{
+			InstallationID:             installation.ID,
+			BusinessDate:               businessDate,
+			UpdatedPlatformCount:       0,
+			ReconciledSkippedTaskCount: 0,
+		}, nil
+	}
+
+	platformIDs := monitoringHeartbeatPlatformIDs(platforms)
+	previousStates, err := s.loadPlatformAccessSnapshotStates(ctx, installation.TenantID, installation.ID, businessDate, platformIDs)
+	if err != nil {
+		return nil, err
+	}
+
+	type heartbeatPlatformAction struct {
+		platform MonitoringHeartbeatPlatform
+		decision monitoringHeartbeatSnapshotDecision
+	}
+
+	actions := make([]heartbeatPlatformAction, 0, len(platforms))
+	inaccessiblePlatformIDs := make([]string, 0, len(platforms))
+	needsTransaction := false
+	for _, platform := range platforms {
+		nextState := monitoringPlatformAccessSnapshotStateFromHeartbeat(platform)
+		decision := decideMonitoringHeartbeatSnapshot(previousStates[platform.AIPlatformID], nextState, now, isPrimaryInstallation)
+		if decision.WriteSnapshot {
+			needsTransaction = true
+		} else {
+			snapshotMetrics.SnapshotSkippedCount++
+		}
+		if decision.PendingReconcileCandidate {
+			inaccessiblePlatformIDs = append(inaccessiblePlatformIDs, platform.AIPlatformID)
+		}
+		actions = append(actions, heartbeatPlatformAction{
+			platform: platform,
+			decision: decision,
+		})
+	}
+
+	if isPrimaryInstallation && len(inaccessiblePlatformIDs) > 0 {
+		pendingPlatforms, pendingErr := s.loadPendingMonitoringTaskPlatformIDs(ctx, installation.TenantID, businessDate, inaccessiblePlatformIDs)
+		if pendingErr != nil {
+			return nil, pendingErr
+		}
+		for idx := range actions {
+			if !actions[idx].decision.PendingReconcileCandidate {
+				continue
+			}
+			if _, ok := pendingPlatforms[actions[idx].platform.AIPlatformID]; ok {
+				actions[idx].decision.ReconcilePending = true
+				needsTransaction = true
+			}
+		}
+	}
+
+	if !needsTransaction {
+		_ = s.touchInstallation(ctx, installation.ID)
+		return &MonitoringHeartbeatResponse{
+			InstallationID:             installation.ID,
+			BusinessDate:               businessDate,
+			UpdatedPlatformCount:       len(platforms),
+			ReconciledSkippedTaskCount: 0,
+		}, nil
+	}

 	tx, err := s.monitoringPool.Begin(ctx)
 	if err != nil {
 		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring heartbeat")
 	}
+	snapshotMetrics.TransactionOpened = true
 	defer tx.Rollback(ctx)

 	var reconciledSkippedTaskCount int64
+	var snapshotInsertedCount int64
+	var snapshotUpdatedCount int64
+	var snapshotRefreshedCount int64
+	var projectionRebuildCount int64
 	affectedBrandIDs := make(map[int64]struct{})
-	for _, platform := range platforms {
-		accessChanged, err := s.upsertPlatformAccessSnapshot(ctx, tx, installation, businessDate, platform)
-		if err != nil {
-			return nil, err
+	for _, action := range actions {
+		if action.decision.WriteSnapshot {
+			if err := s.upsertPlatformAccessSnapshot(ctx, tx, installation, businessDate, action.platform); err != nil {
+				return nil, err
+			}
+			switch action.decision.SnapshotOperation {
+			case "insert":
+				snapshotInsertedCount++
+			case "update":
+				snapshotUpdatedCount++
+			case "refresh":
+				snapshotRefreshedCount++
+			}
 		}

 		if !isPrimaryInstallation {
 			continue
 		}

-		needsProjectionRefresh := accessChanged
-		if monitoringStringValue(platform.AccessStatus) != "accessible" {
-			count, reconcileErr := s.reconcilePendingTasksForPlatform(ctx, tx, installation.TenantID, businessDate, platform.AIPlatformID)
+		needsProjectionRefresh := action.decision.StateChanged
+		if action.decision.ReconcilePending {
+			count, reconcileErr := s.reconcilePendingTasksForPlatform(ctx, tx, installation.TenantID, businessDate, action.platform.AIPlatformID)
 			if reconcileErr != nil {
 				return nil, reconcileErr
 			}
@@ -486,7 +598,7 @@ func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationI
 			continue
 		}

-		brandIDs, loadErr := s.loadMonitoringAffectedBrandIDs(ctx, tx, installation.TenantID, businessDay, platform.AIPlatformID)
+		brandIDs, loadErr := s.loadMonitoringAffectedBrandIDs(ctx, tx, installation.TenantID, businessDay, action.platform.AIPlatformID)
 		if loadErr != nil {
 			return nil, loadErr
 		}
@@ -500,6 +612,7 @@ func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationI
 			if err := s.rebuildBrandDailyProjection(ctx, tx, installation.TenantID, brandID, businessDay); err != nil {
 				return nil, err
 			}
+			projectionRebuildCount++
 		}
 	}

@@ -507,6 +620,12 @@ func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationI
 		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring heartbeat")
 	}

+	snapshotMetrics.SnapshotInsertedCount = snapshotInsertedCount
+	snapshotMetrics.SnapshotUpdatedCount = snapshotUpdatedCount
+	snapshotMetrics.SnapshotRefreshedCount = snapshotRefreshedCount
+	snapshotMetrics.ReconciledSkippedTaskCount = reconciledSkippedTaskCount
+	snapshotMetrics.ProjectionRebuildCount = projectionRebuildCount
+
 	_ = s.touchInstallation(ctx, installation.ID)
 	if isPrimaryInstallation && len(affectedBrandIDs) > 0 {
 		s.dispatchMonitoringProjectionRebuilds(installation.TenantID, businessDay, affectedBrandIDs, "heartbeat")
@@ -963,80 +1082,224 @@ func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context
 }

 func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) (bool, error) {
-	var preferred sql.NullString
-	err := s.businessPool.QueryRow(ctx, `
-		SELECT primary_client_id::text
-		FROM tenant_monitoring_quotas
-		WHERE tenant_id = $1
-	`, tenantID).Scan(&preferred)
-	if err != nil && err != pgx.ErrNoRows {
-		return false, response.ErrInternal(50041, "query_failed", "failed to inspect primary monitoring desktop client")
+	if installation == nil || strings.TrimSpace(installation.ID) == "" {
+		return false, nil
+	}
+	isPrimary, err := s.claimHeartbeatPrimaryLease(ctx, tenantID, installation.WorkspaceID, installation.ID)
+	if err != nil {
+		return false, err
+	}
+	return isPrimary, nil
+}
+
+func (s *MonitoringCallbackService) claimHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
+	startedAt := time.Now()
+	installationID = strings.TrimSpace(installationID)
+	leaseTTL := formatPgInterval(desktopClientPresenceTTL)
+
+	state, err := s.loadHeartbeatPrimaryLeaseState(ctx, tenantID, workspaceID, installationID)
+	if err != nil {
+		recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
+		return false, err
+	}
+	if !state.Found {
+		inserted, insertErr := s.tryInsertHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, leaseTTL)
+		if insertErr != nil {
+			recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
+			return false, insertErr
+		}
+		if inserted {
+			recordHeartbeatPrimaryLeaseMetric("inserted", time.Since(startedAt))
+			recordHeartbeatPrimaryLeaseWriteMetric("insert")
+			return true, nil
+		}
+		recordHeartbeatPrimaryLeaseMetric("contention", time.Since(startedAt))
+		return s.resolveHeartbeatPrimaryAfterContention(ctx, tenantID, workspaceID, installationID)
 	}

-	if !preferred.Valid || preferred.String == installation.ID {
-		if !preferred.Valid {
-			if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installation); err != nil {
-				return false, err
-			}
-		}
+	decision := decideHeartbeatPrimaryLease(state)
+	switch decision.Action {
+	case heartbeatPrimaryLeaseActionPrimaryHit:
+		recordHeartbeatPrimaryLeaseMetric("primary_hit", time.Since(startedAt))
 		return true, nil
-	}
-
-	online, err := s.isMonitoringInstallationRecentlyOnline(ctx, tenantID, installation.WorkspaceID, preferred.String)
-	if err != nil {
-		return false, err
-	}
-	if online {
+	case heartbeatPrimaryLeaseActionNonPrimaryLive:
+		recordHeartbeatPrimaryLeaseMetric("non_primary_live", time.Since(startedAt))
 		return false, nil
-	}
-
-	if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installation); err != nil {
-		return false, err
-	}
-	return true, nil
-}
-
-func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
-	var lastSeenAt sql.NullTime
-	err := s.businessPool.QueryRow(ctx, `
-		SELECT last_seen_at
-		FROM desktop_clients
-		WHERE id = $1::uuid
-		  AND tenant_id = $2
-		  AND workspace_id = $3
-		  AND revoked_at IS NULL
-	`, strings.TrimSpace(installationID), tenantID, workspaceID).Scan(&lastSeenAt)
-	if err != nil {
-		if err == pgx.ErrNoRows {
-			return false, nil
+	case heartbeatPrimaryLeaseActionRenew:
+		updated, updateErr := s.renewHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, leaseTTL)
+		if updateErr != nil {
+			recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
+			return false, updateErr
+		}
+		if updated {
+			recordHeartbeatPrimaryLeaseMetric("renewed", time.Since(startedAt))
+			recordHeartbeatPrimaryLeaseWriteMetric("renew")
+			return true, nil
+		}
+	case heartbeatPrimaryLeaseActionClaimExpired, heartbeatPrimaryLeaseActionClaimUnavailable:
+		updated, updateErr := s.claimExpiredHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, state.PrimaryInstallationID, leaseTTL, decision.Action)
+		if updateErr != nil {
+			recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
+			return false, updateErr
+		}
+		if updated {
+			if decision.Action == heartbeatPrimaryLeaseActionClaimExpired {
+				recordHeartbeatPrimaryLeaseMetric("claimed_expired", time.Since(startedAt))
+			} else {
+				recordHeartbeatPrimaryLeaseMetric("claimed_unavailable", time.Since(startedAt))
+			}
+			recordHeartbeatPrimaryLeaseWriteMetric("claim")
+			return true, nil
 		}
-		return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client heartbeat")
 	}

-	if !lastSeenAt.Valid {
-		return false, nil
-	}
-	return time.Since(lastSeenAt.Time) <= monitoringOnlineDuration, nil
+	recordHeartbeatPrimaryLeaseMetric("contention", time.Since(startedAt))
+	return s.resolveHeartbeatPrimaryAfterContention(ctx, tenantID, workspaceID, installationID)
 }

-func (s *MonitoringCallbackService) persistHeartbeatPrimaryInstallationID(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) error {
-	if _, err := s.businessPool.Exec(ctx, `
-		INSERT INTO tenant_monitoring_quotas (
-			tenant_id, workspace_id, primary_client_id
+func (s *MonitoringCallbackService) tryInsertHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID, leaseTTL string) (bool, error) {
+	tag, err := s.businessPool.Exec(ctx, `
+		INSERT INTO desktop_client_primary_leases (
+			tenant_id,
+			workspace_id,
+			primary_client_id,
+			lease_expires_at
 		)
-		VALUES (
-			$1,
-			$2,
-			$3::uuid
-		)
-		ON CONFLICT (tenant_id) DO UPDATE
-		SET workspace_id = EXCLUDED.workspace_id,
-			primary_client_id = EXCLUDED.primary_client_id,
-			updated_at = NOW()
-	`, tenantID, installation.WorkspaceID, installation.ID); err != nil {
-		return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring desktop client")
+		VALUES ($1, $2, $3::uuid, NOW() + $4::interval)
+		ON CONFLICT (tenant_id, workspace_id) DO NOTHING
+	`, tenantID, workspaceID, installationID, leaseTTL)
+	if err != nil {
+		return false, response.ErrInternal(50041, "primary_lease_insert_failed", "failed to initialize monitoring desktop primary lease")
 	}
-	return nil
+	return tag.RowsAffected() > 0, nil
+}
+
+type heartbeatPrimaryLeaseState struct {
+	Found                 bool
+	PrimaryInstallationID string
+	CurrentIsPrimary      bool
+	LeaseLive             bool
+	RenewDue              bool
+	PrimaryAvailable      bool
+}
+
+func (s *MonitoringCallbackService) loadHeartbeatPrimaryLeaseState(ctx context.Context, tenantID, workspaceID int64, installationID string) (heartbeatPrimaryLeaseState, error) {
+	var state heartbeatPrimaryLeaseState
+	renewAfter := formatPgInterval(monitoringHeartbeatPrimaryRenewAfter)
+	if err := s.businessPool.QueryRow(ctx, `
+		SELECT
+			l.primary_client_id::text,
+			l.primary_client_id = $3::uuid AS current_is_primary,
+			l.lease_expires_at > NOW() AS lease_live,
+			l.updated_at <= NOW() - $4::interval AS renew_due,
+			dc.id IS NOT NULL AND dc.revoked_at IS NULL AS primary_available
+		FROM desktop_client_primary_leases l
+		LEFT JOIN desktop_clients dc ON dc.id = l.primary_client_id
+		WHERE l.tenant_id = $1
+		  AND l.workspace_id = $2
+	`, tenantID, workspaceID, installationID, renewAfter).Scan(
+		&state.PrimaryInstallationID,
+		&state.CurrentIsPrimary,
+		&state.LeaseLive,
+		&state.RenewDue,
+		&state.PrimaryAvailable,
+	); err != nil {
+		if err == pgx.ErrNoRows {
+			return state, nil
+		}
+		return state, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop primary lease")
+	}
+	state.Found = true
+	return state, nil
+}
+
+type heartbeatPrimaryLeaseAction string
+
+const (
+	heartbeatPrimaryLeaseActionPrimaryHit       heartbeatPrimaryLeaseAction = "primary_hit"
+	heartbeatPrimaryLeaseActionNonPrimaryLive   heartbeatPrimaryLeaseAction = "non_primary_live"
+	heartbeatPrimaryLeaseActionRenew            heartbeatPrimaryLeaseAction = "renew"
+	heartbeatPrimaryLeaseActionClaimExpired     heartbeatPrimaryLeaseAction = "claim_expired"
+	heartbeatPrimaryLeaseActionClaimUnavailable heartbeatPrimaryLeaseAction = "claim_unavailable"
+)
+
+type heartbeatPrimaryLeaseDecision struct {
+	Action heartbeatPrimaryLeaseAction
+}
+
+func decideHeartbeatPrimaryLease(state heartbeatPrimaryLeaseState) heartbeatPrimaryLeaseDecision {
+	if !state.Found {
+		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionNonPrimaryLive}
+	}
+	if state.CurrentIsPrimary {
+		if !state.LeaseLive || !state.PrimaryAvailable || state.RenewDue {
+			return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionRenew}
+		}
+		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionPrimaryHit}
+	}
+	if !state.LeaseLive {
+		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionClaimExpired}
+	}
+	if !state.PrimaryAvailable {
+		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionClaimUnavailable}
+	}
+	return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionNonPrimaryLive}
+}
+
+func (s *MonitoringCallbackService) renewHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID, leaseTTL string) (bool, error) {
+	tag, err := s.businessPool.Exec(ctx, `
+		UPDATE desktop_client_primary_leases
+		SET lease_expires_at = NOW() + $4::interval,
+		    updated_at = NOW()
+		WHERE tenant_id = $1
+		  AND workspace_id = $2
+		  AND primary_client_id = $3::uuid
+	`, tenantID, workspaceID, installationID, leaseTTL)
+	if err != nil {
+		return false, response.ErrInternal(50041, "primary_lease_update_failed", "failed to renew monitoring desktop primary lease")
+	}
+	return tag.RowsAffected() > 0, nil
+}
+
+func (s *MonitoringCallbackService) claimExpiredHeartbeatPrimaryLease(
+	ctx context.Context,
+	tenantID, workspaceID int64,
+	installationID string,
+	previousPrimaryInstallationID string,
+	leaseTTL string,
+	action heartbeatPrimaryLeaseAction,
+) (bool, error) {
+	claimPredicate := "lease_expires_at <= NOW()"
+	if action == heartbeatPrimaryLeaseActionClaimUnavailable {
+		claimPredicate = `NOT EXISTS (
+			SELECT 1
+			FROM desktop_clients dc
+			WHERE dc.id = desktop_client_primary_leases.primary_client_id
+			  AND dc.revoked_at IS NULL
+		)`
+	}
+	query := `
+		UPDATE desktop_client_primary_leases
+		SET primary_client_id = $3::uuid,
+		    lease_expires_at = NOW() + $5::interval,
+		    updated_at = NOW()
+		WHERE tenant_id = $1
+		  AND workspace_id = $2
+		  AND primary_client_id = $4::uuid
+		  AND ` + claimPredicate
+	tag, err := s.businessPool.Exec(ctx, query, tenantID, workspaceID, installationID, previousPrimaryInstallationID, leaseTTL)
+	if err != nil {
+		return false, response.ErrInternal(50041, "primary_lease_update_failed", "failed to claim monitoring desktop primary lease")
+	}
+	return tag.RowsAffected() > 0, nil
+}
+
+func (s *MonitoringCallbackService) resolveHeartbeatPrimaryAfterContention(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
+	state, err := s.loadHeartbeatPrimaryLeaseState(ctx, tenantID, workspaceID, installationID)
+	if err != nil {
+		return false, err
+	}
+	return state.CurrentIsPrimary && state.LeaseLive && state.PrimaryAvailable, nil
 }

 func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantID, taskID int64) (*monitoringCollectTask, error) {
@@ -1464,20 +1727,7 @@ func (s *MonitoringCallbackService) markTaskResultReceived(ctx context.Context,
 	return nil
 }

-func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Context, tx pgx.Tx, installation *monitoringInstallationIdentity, businessDate string, platform MonitoringHeartbeatPlatform) (bool, error) {
-	previous, err := s.loadPlatformAccessSnapshotState(ctx, tx, installation.TenantID, installation.ID, businessDate, platform.AIPlatformID)
-	if err != nil {
-		return false, err
-	}
-
-	nextState := monitoringPlatformAccessSnapshotState{
-		AccessStatus:  monitoringStringValue(platform.AccessStatus),
-		Connected:     derefBool(platform.Connected),
-		Accessible:    derefBool(platform.Accessible),
-		ReasonCode:    optionalString(monitoringStringValue(platform.ReasonCode)),
-		ReasonMessage: optionalString(monitoringStringValue(platform.ReasonMessage)),
-	}
-
+func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Context, tx pgx.Tx, installation *monitoringInstallationIdentity, businessDate string, platform MonitoringHeartbeatPlatform) error {
 	if _, err := tx.Exec(ctx, `
 		INSERT INTO monitoring_platform_access_snapshots (
 			tenant_id, client_id, ai_platform_id, business_date, access_status,
@@ -1498,44 +1748,101 @@ func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Con
 		installation.ID,
 		platform.AIPlatformID,
 		businessDate,
-		platform.AccessStatus,
+		monitoringStringValue(platform.AccessStatus),
 		derefBool(platform.Connected),
 		derefBool(platform.Accessible),
 		resolveHeartbeatDetectedAt(platform.DetectedAt),
 		nullableString(platform.ReasonCode),
 		nullableString(platform.ReasonMessage),
 	); err != nil {
-		return false, response.ErrInternal(50041, "access_snapshot_upsert_failed", "failed to persist monitoring platform access snapshot")
+		return response.ErrInternal(50041, "access_snapshot_upsert_failed", "failed to persist monitoring platform access snapshot")
 	}
-	return hasPlatformAccessSnapshotChanged(previous, nextState), nil
+	return nil
 }

-func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.Context, tx pgx.Tx, tenantID int64, installationID string, businessDate, aiPlatformID string) (*monitoringPlatformAccessSnapshotState, error) {
-	item := &monitoringPlatformAccessSnapshotState{}
-	var reasonCode sql.NullString
-	var reasonMessage sql.NullString
-	if err := tx.QueryRow(ctx, `
-		SELECT access_status, connected, accessible, reason_code, reason_message
+func (s *MonitoringCallbackService) loadPlatformAccessSnapshotStates(ctx context.Context, tenantID int64, installationID string, businessDate string, aiPlatformIDs []string) (map[string]*monitoringPlatformAccessSnapshotState, error) {
+	states := make(map[string]*monitoringPlatformAccessSnapshotState, len(aiPlatformIDs))
+	if len(aiPlatformIDs) == 0 {
+		return states, nil
+	}
+
+	rows, err := s.monitoringPool.Query(ctx, `
+		SELECT ai_platform_id, access_status, connected, accessible, reason_code, reason_message, updated_at
 		FROM monitoring_platform_access_snapshots
 		WHERE tenant_id = $1
 		  AND client_id = $2::uuid
-		  AND ai_platform_id = $3
-		  AND business_date = $4::date
-	`, tenantID, installationID, aiPlatformID, businessDate).Scan(
-		&item.AccessStatus,
-		&item.Connected,
-		&item.Accessible,
-		&reasonCode,
-		&reasonMessage,
-	); err != nil {
-		if err == pgx.ErrNoRows {
-			return nil, nil
-		}
-		return nil, response.ErrInternal(50041, "access_snapshot_lookup_failed", "failed to inspect monitoring platform access snapshot")
+		  AND business_date = $3::date
+		  AND ai_platform_id = ANY($4::text[])
+	`, tenantID, installationID, businessDate, aiPlatformIDs)
+	if err != nil {
+		return nil, response.ErrInternal(50041, "access_snapshot_lookup_failed", "failed to inspect monitoring platform access snapshots")
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		item := &monitoringPlatformAccessSnapshotState{}
+		var platformID string
+		var reasonCode sql.NullString
+		var reasonMessage sql.NullString
+		if scanErr := rows.Scan(
+			&platformID,
+			&item.AccessStatus,
+			&item.Connected,
+			&item.Accessible,
+			&reasonCode,
+			&reasonMessage,
+			&item.UpdatedAt,
+		); scanErr != nil {
+			return nil, response.ErrInternal(50041, "access_snapshot_scan_failed", "failed to parse monitoring platform access snapshots")
+		}
+		if reasonCode.Valid {
+			item.ReasonCode = optionalString(reasonCode.String)
+		}
+		if reasonMessage.Valid {
+			item.ReasonMessage = optionalString(reasonMessage.String)
+		}
+		states[platformID] = item
+	}
+	if err := rows.Err(); err != nil {
+		return nil, response.ErrInternal(50041, "access_snapshot_scan_failed", "failed to iterate monitoring platform access snapshots")
+	}
+
+	return states, nil
+}
+
+func monitoringPlatformAccessSnapshotStateFromHeartbeat(platform MonitoringHeartbeatPlatform) monitoringPlatformAccessSnapshotState {
+	return monitoringPlatformAccessSnapshotState{
+		AccessStatus:  monitoringStringValue(platform.AccessStatus),
+		Connected:     derefBool(platform.Connected),
+		Accessible:    derefBool(platform.Accessible),
+		ReasonCode:    optionalString(monitoringStringValue(platform.ReasonCode)),
+		ReasonMessage: optionalString(monitoringStringValue(platform.ReasonMessage)),
+	}
+}
+
+func decideMonitoringHeartbeatSnapshot(previous *monitoringPlatformAccessSnapshotState, next monitoringPlatformAccessSnapshotState, now time.Time, isPrimary bool) monitoringHeartbeatSnapshotDecision {
+	stateChanged := hasPlatformAccessSnapshotChanged(previous, next)
+	refreshDue := false
+	if previous != nil && !stateChanged {
+		refreshDue = previous.UpdatedAt.IsZero() || !previous.UpdatedAt.After(now.Add(-monitoringHeartbeatAccessSnapshotRefreshAfter))
+	}
+
+	operation := "skip"
+	if previous == nil {
+		operation = "insert"
+	} else if stateChanged {
+		operation = "update"
+	} else if refreshDue {
+		operation = "refresh"
+	}
+
+	return monitoringHeartbeatSnapshotDecision{
+		WriteSnapshot:             stateChanged || refreshDue,
+		StateChanged:              stateChanged,
+		RefreshDue:                refreshDue,
+		PendingReconcileCandidate: isPrimary && next.AccessStatus != "accessible",
+		SnapshotOperation:         operation,
 	}
-	item.ReasonCode = optionalString(reasonCode.String)
-	item.ReasonMessage = optionalString(reasonMessage.String)
-	return item, nil
 }

 func hasPlatformAccessSnapshotChanged(previous *monitoringPlatformAccessSnapshotState, next monitoringPlatformAccessSnapshotState) bool {
@@ -1558,6 +1865,41 @@ func hasPlatformAccessSnapshotChanged(previous *monitoringPlatformAccessSnapshot
 	return false
 }

+func (s *MonitoringCallbackService) loadPendingMonitoringTaskPlatformIDs(ctx context.Context, tenantID int64, businessDate string, aiPlatformIDs []string) (map[string]struct{}, error) {
+	result := make(map[string]struct{}, len(aiPlatformIDs))
+	if len(aiPlatformIDs) == 0 {
+		return result, nil
+	}
+
+	rows, err := s.monitoringPool.Query(ctx, `
+		SELECT ai_platform_id
+		FROM monitoring_collect_tasks
+		WHERE tenant_id = $1
+		  AND ai_platform_id = ANY($2::text[])
+		  AND collector_type = $3
+		  AND business_date = $4::date
+		  AND status = 'pending'
+		GROUP BY ai_platform_id
+	`, tenantID, aiPlatformIDs, monitoringCollectorType, businessDate)
+	if err != nil {
+		return nil, response.ErrInternal(50041, "pending_task_platform_lookup_failed", "failed to inspect pending monitoring task platforms")
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var platformID string
+		if scanErr := rows.Scan(&platformID); scanErr != nil {
+			return nil, response.ErrInternal(50041, "pending_task_platform_scan_failed", "failed to parse pending monitoring task platforms")
+		}
+		result[platformID] = struct{}{}
+	}
+	if err := rows.Err(); err != nil {
+		return nil, response.ErrInternal(50041, "pending_task_platform_scan_failed", "failed to iterate pending monitoring task platforms")
+	}
+
+	return result, nil
+}
+
 func (s *MonitoringCallbackService) reconcilePendingTasksForPlatform(ctx context.Context, tx pgx.Tx, tenantID int64, businessDate, aiPlatformID string) (int64, error) {
 	tag, err := tx.Exec(ctx, `
 		UPDATE monitoring_collect_tasks
@@ -2530,6 +2872,20 @@ func normalizeMonitoringLeasePlatformIDs(values []string) []string {
 	return result
 }

+func monitoringHeartbeatPlatformIDs(platforms []MonitoringHeartbeatPlatform) []string {
+	if len(platforms) == 0 {
+		return nil
+	}
+	result := make([]string, 0, len(platforms))
+	for _, platform := range platforms {
+		platformID := strings.TrimSpace(platform.AIPlatformID)
+		if platformID != "" {
+			result = append(result, platformID)
+		}
+	}
+	return result
+}
+
 func normalizeHeartbeatPlatforms(items []MonitoringHeartbeatPlatform, fallbackDetectedAt time.Time) ([]MonitoringHeartbeatPlatform, error) {
 	if len(items) == 0 {
 		return nil, nil