Real-Time GPU K-Means Clustering for Mutating Nodes¶

// Node changes → Reassign to nearest centroid
func (km *GPUKMeans) ReassignNode(nodeID string, newEmbedding []float32) int {
    // GPU kernel: compute distances to K centroids
    // Time: ~0.01-0.1ms for single node
    return km.findNearestCentroid(newEmbedding)
}

// Accumulate changes, update centroids in batches
func (km *GPUKMeans) UpdateBatch(nodes []NodeUpdate) {
    // Every 100 node updates:
    // 1. Reassign all changed nodes (parallel on GPU)
    // 2. Update affected centroids (parallel reduction)
    // 3. Check for cluster drift

    // Time: ~1-5ms for 100 nodes
}

// Periodic full re-clustering
func (km *GPUKMeans) ReclusterAll() {
    // Every 10K mutations or 1 hour:
    // Full k-means from scratch

    // Time: 10-100ms for 10K nodes
}

// Detect when node embeddings change significantly
type ConceptDriftMonitor struct {
    gpu *GPUKMeans
    driftLog map[storage.NodeID][]ClusterChange
}

type ClusterChange struct {
    Timestamp   time.Time
    OldCluster  int
    NewCluster  int
    Confidence  float32
}

func (m *ConceptDriftMonitor) OnNodeUpdate(
    nodeID storage.NodeID,
    oldEmbedding, newEmbedding []float32,
) {
    oldCluster := m.gpu.Predict(oldEmbedding)
    newCluster := m.gpu.Predict(newEmbedding)

    if oldCluster != newCluster {
        // Node changed topics!
        m.logClusterChange(nodeID, ClusterChange{
            Timestamp:  time.Now(),
            OldCluster: oldCluster,
            NewCluster: newCluster,
            Confidence: m.gpu.GetAssignmentConfidence(nodeID),
        })

        // Update centroid incrementally
        m.gpu.UpdateCentroid(oldCluster, -oldEmbedding) // remove
        m.gpu.UpdateCentroid(newCluster, +newEmbedding) // add

        // Trigger event
        m.notifyClusterChange(nodeID, oldCluster, newCluster)
    }
}

// Example: Detect trending topics
func (m *ConceptDriftMonitor) GetTrendingClusters(
    window time.Duration,
) []int {
    // Find clusters gaining the most nodes
    gains := make(map[int]int)
    cutoff := time.Now().Add(-window)

    for _, changes := range m.driftLog {
        for _, change := range changes {
            if change.Timestamp.After(cutoff) {
                gains[change.NewCluster]++
            }
        }
    }

    // Return top growing clusters
    return topK(gains, 10)
}

// Keep "related documents" fresh as nodes change
func (e *InferenceEngine) OnNodeEmbeddingUpdate(
    ctx context.Context,
    nodeID storage.NodeID,
    newEmbedding []float32,
) {
    // Reassign cluster (0.1ms on GPU)
    newCluster := e.clustering.ReassignNode(nodeID, newEmbedding)

    // Update related documents instantly
    e.mu.Lock()
    e.clusterIndex[newCluster] = append(e.clusterIndex[newCluster], nodeID)

    // Remove from old cluster if needed
    oldCluster := e.nodeToCluster[nodeID]
    if oldCluster != newCluster {
        e.removeFromCluster(oldCluster, nodeID)
    }

    e.nodeToCluster[nodeID] = newCluster
    e.mu.Unlock()

    // Invalidate cached recommendations
    delete(e.relatedDocsCache, nodeID)

    // Pre-compute new recommendations (async)
    go e.precomputeRecommendations(nodeID)
}

// GetRelatedDocuments - always fresh, sub-millisecond lookup
func (e *InferenceEngine) GetRelatedDocuments(
    nodeID storage.NodeID,
    topK int,
) []storage.NodeID {
    cluster := e.nodeToCluster[nodeID]
    candidates := e.clusterIndex[cluster]

    // O(1) cluster lookup, then refine if needed
    if len(candidates) <= topK {
        return candidates
    }

    // Refine with vector similarity
    return e.rankByProximity(nodeID, candidates)[:topK]
}

// Update clustering as documents are added/modified
func (e *InferenceEngine) OnStore(
    ctx context.Context,
    nodeID storage.NodeID,
    embedding []float32,
) ([]EdgeSuggestion, error) {
    // 1. Get cluster assignment (GPU, <0.1ms)
    cluster := e.clustering.AssignToCluster(embedding)

    // 2. Get related docs from cluster (O(1) lookup)
    relatedDocs := e.clustering.GetClusterMembers(cluster)

    // 3. Refine with vector similarity if needed
    candidates := e.filterByMinScore(relatedDocs, 0.7)
    suggestions := e.rankBySimilarity(embedding, candidates)

    // 4. Update centroid if threshold reached (async)
    go e.clustering.UpdateCentroidIfNeeded(cluster)

    // 5. Check if we should re-cluster (async)
    go e.clustering.CheckReclusterTriggers()

    return suggestions, nil
}

// DON'T DO THIS
for i := 0; i < 1000; i++ {
    node.Embedding[i] += delta
    km.ReassignNode(node.ID, node.Embedding) // GPU call per loop iteration!
}
// Problem: 1000 GPU kernel launches, ~100ms total

// DO THIS INSTEAD
batchUpdates := make([]NodeUpdate, 0, 1000)
for i := 0; i < 1000; i++ {
    node.Embedding[i] += delta
    batchUpdates = append(batchUpdates, NodeUpdate{
        NodeID:   node.ID,
        Embedding: node.Embedding,
    })
}
km.ReassignBatch(batchUpdates) // Single GPU call, ~1-2ms
// 50-100x faster!

// Rule of thumb: GPU overhead is ~0.1-0.5ms
// Use CPU if updates are very small

func (e *InferenceEngine) ReassignNodes(nodes []NodeUpdate) {
    totalNodes := len(e.allNodes)
    changedNodes := len(nodes)

    // If < 1% of nodes changed, use CPU
    if changedNodes < totalNodes*0.01 {
        // CPU assignment is faster for small batches
        for _, node := range nodes {
            cluster := e.cpuKMeans.Predict(node.Embedding)
            e.updateCluster(node.ID, cluster)
        }
    } else {
        // GPU is faster for larger batches
        clusters := e.gpuKMeans.PredictBatch(nodes)
        for i, node := range nodes {
            e.updateCluster(node.ID, clusters[i])
        }
    }
}

// DON'T DO THIS
func (e *InferenceEngine) OnNodeUpdate(nodeID, embedding) {
    e.gpuKMeans.Fit(e.getAllEmbeddings()) // 10-100ms EVERY update!
}
// Problem: Wastes GPU cycles, slows down all updates

// DO THIS INSTEAD
func (e *InferenceEngine) OnNodeUpdate(nodeID, embedding) {
    // Fast reassignment (0.1ms)
    e.gpuKMeans.ReassignNode(nodeID, embedding)

    // Track drift
    e.updateCount++

    // Re-cluster only when needed
    if e.shouldRecluster() {
        go e.reclusterAsync() // Async, doesn't block
    }
}

┌─────────────────────────────────────────────────────┐
│ TIER 1: Instant Reassignment (<0.1ms)              │
│ • Run on EVERY node update                          │
│ • Reassign to nearest centroid                      │
│ • Update lookup tables                              │
└─────────────────────────────────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────────┐
│ TIER 2: Batch Centroid Update (1-5ms)              │
│ • Run every 100 node updates                        │
│ • Adjust centroids for changed nodes                │
│ • Detect significant drift                          │
└─────────────────────────────────────────────────────┘
                      ↓
┌─────────────────────────────────────────────────────┐
│ TIER 3: Full Re-clustering (10-100ms)              │
│ • Run every 10K updates or 1 hour                   │
│ • Full k-means from scratch                         │
│ • Ground truth recalibration                        │
└─────────────────────────────────────────────────────┘

// pkg/inference/realtime_clustering.go
package inference

import (
    "context"
    "math"
    "sync"
    "time"

    "github.com/orneryd/nornicdb/pkg/gpu/kmeans"
    "github.com/orneryd/nornicdb/pkg/storage"
)

type RealtimeClusteringEngine struct {
    mu sync.RWMutex

    // GPU k-means (persistent in VRAM)
    gpuKMeans *kmeans.GPUKMeans
    storage   storage.Engine

    // Fast lookup tables (in RAM)
    nodeCluster   map[storage.NodeID]int        // node → cluster
    clusterNodes  map[int][]storage.NodeID      // cluster → nodes
    centroids     [][]float32                   // cached centroids

    // Incremental update tracking
    pendingUpdates []NodeUpdate                 // buffer for batch updates
    updateCount    int                          // count since last re-cluster
    lastRecluster  time.Time

    // Drift detection
    centroidDrift  map[int]float32              // cluster → cumulative drift

    // Configuration
    batchSize      int     // Reassign in batches of N (default: 100)
    reclusterEvery int     // Full re-cluster every N updates (default: 10000)
    driftThreshold float32 // Re-cluster if centroid drift > threshold (default: 0.1)
    maxAge         time.Duration // Max time between re-clusters (default: 1 hour)
}

type NodeUpdate struct {
    NodeID       storage.NodeID
    OldEmbedding []float32
    NewEmbedding []float32
    OldCluster   int
    NewCluster   int
    Timestamp    time.Time
}

type Config struct {
    NumClusters    int
    Dimensions     int
    BatchSize      int
    ReclusterEvery int
    DriftThreshold float32
    MaxAge         time.Duration
    Device         string  // "cuda", "metal", "auto"
}

func DefaultConfig() Config {
    return Config{
        NumClusters:    500,
        Dimensions:     1024,
        BatchSize:      100,
        ReclusterEvery: 10000,
        DriftThreshold: 0.1,
        MaxAge:         1 * time.Hour,
        Device:         "auto",
    }
}

// NewRealtimeClusteringEngine creates a new real-time clustering engine
func NewRealtimeClusteringEngine(
    storage storage.Engine,
    config Config,
) (*RealtimeClusteringEngine, error) {
    km, err := kmeans.NewGPUKMeans(kmeans.Config{
        NumClusters: config.NumClusters,
        Dimensions:  config.Dimensions,
        Device:      config.Device,
    })
    if err != nil {
        return nil, err
    }

    return &RealtimeClusteringEngine{
        gpuKMeans:      km,
        storage:        storage,
        nodeCluster:    make(map[storage.NodeID]int),
        clusterNodes:   make(map[int][]storage.NodeID),
        centroidDrift:  make(map[int]float32),
        batchSize:      config.BatchSize,
        reclusterEvery: config.ReclusterEvery,
        driftThreshold: config.DriftThreshold,
        maxAge:         config.MaxAge,
        lastRecluster:  time.Now(),
    }, nil
}

// Initialize performs initial clustering
func (e *RealtimeClusteringEngine) Initialize(ctx context.Context) error {
    // Get all nodes with embeddings
    nodes, err := e.storage.GetAllNodesWithEmbeddings(ctx)
    if err != nil {
        return err
    }

    embeddings := make([][]float32, len(nodes))
    nodeIDs := make([]storage.NodeID, len(nodes))

    for i, node := range nodes {
        embeddings[i] = node.Embedding
        nodeIDs[i] = node.ID
    }

    // Initial clustering on GPU
    if err := e.gpuKMeans.Fit(embeddings); err != nil {
        return err
    }

    // Build lookup tables
    assignments := e.gpuKMeans.GetClusterAssignments()

    e.mu.Lock()
    defer e.mu.Unlock()

    for i, nodeID := range nodeIDs {
        cluster := assignments[i]
        e.nodeCluster[nodeID] = cluster
        e.clusterNodes[cluster] = append(e.clusterNodes[cluster], nodeID)
    }

    e.centroids = e.gpuKMeans.GetClusterCentroids()
    e.lastRecluster = time.Now()

    return nil
}

// ========================================================================
// TIER 1: Instant Reassignment (<0.1ms)
// ========================================================================

// OnNodeUpdate handles real-time node mutations
func (e *RealtimeClusteringEngine) OnNodeUpdate(
    ctx context.Context,
    nodeID storage.NodeID,
    oldEmbedding, newEmbedding []float32,
) error {
    // Fast path: predict new cluster (GPU, <0.1ms)
    newCluster := e.gpuKMeans.Predict(newEmbedding)

    e.mu.Lock()
    defer e.mu.Unlock()

    oldCluster, exists := e.nodeCluster[nodeID]
    if !exists {
        // New node - just add it
        e.addToCluster(newCluster, nodeID)
        e.nodeCluster[nodeID] = newCluster
        return nil
    }

    if newCluster != oldCluster {
        // Cluster changed - update lookup tables
        e.removeFromCluster(oldCluster, nodeID)
        e.addToCluster(newCluster, nodeID)
        e.nodeCluster[nodeID] = newCluster

        // Track update for centroid adjustment
        e.pendingUpdates = append(e.pendingUpdates, NodeUpdate{
            NodeID:       nodeID,
            OldEmbedding: oldEmbedding,
            NewEmbedding: newEmbedding,
            OldCluster:   oldCluster,
            NewCluster:   newCluster,
            Timestamp:    time.Now(),
        })

        // Track drift
        drift := e.computeDrift(oldEmbedding, newEmbedding)
        e.centroidDrift[oldCluster] += drift
        e.centroidDrift[newCluster] += drift
    }

    e.updateCount++

    // TIER 2: Mini-batch centroid update (1-5ms, every 100 updates)
    if len(e.pendingUpdates) >= e.batchSize {
        updates := e.pendingUpdates
        e.pendingUpdates = nil
        go e.updateCentroidsBatch(updates)
    }

    // TIER 3: Full re-clustering (10-100ms, conditionally)
    if e.shouldRecluster() {
        go e.reclusterAll(ctx)
    }

    return nil
}

// ========================================================================
// TIER 2: Batch Centroid Update (1-5ms)
// ========================================================================

// updateCentroidsBatch updates centroids incrementally
func (e *RealtimeClusteringEngine) updateCentroidsBatch(updates []NodeUpdate) {
    // Group updates by cluster
    clusterAdds := make(map[int][][]float32)
    clusterRemoves := make(map[int][][]float32)

    for _, update := range updates {
        // Remove old embedding from old cluster
        clusterRemoves[update.OldCluster] = append(
            clusterRemoves[update.OldCluster],
            update.OldEmbedding,
        )

        // Add new embedding to new cluster
        clusterAdds[update.NewCluster] = append(
            clusterAdds[update.NewCluster],
            update.NewEmbedding,
        )
    }

    // Update centroids on GPU (batched)
    for cluster, embeddings := range clusterRemoves {
        e.gpuKMeans.RemoveFromCentroid(cluster, embeddings)
    }

    for cluster, embeddings := range clusterAdds {
        e.gpuKMeans.AddToCentroid(cluster, embeddings)
    }

    // Refresh cached centroids
    e.mu.Lock()
    e.centroids = e.gpuKMeans.GetClusterCentroids()
    e.mu.Unlock()
}

// ========================================================================
// TIER 3: Full Re-clustering (10-100ms)
// ========================================================================

// reclusterAll performs full re-clustering from scratch
func (e *RealtimeClusteringEngine) reclusterAll(ctx context.Context) {
    // Get all current embeddings
    e.mu.RLock()
    embeddings := make([][]float32, 0, len(e.nodeCluster))
    nodeIDs := make([]storage.NodeID, 0, len(e.nodeCluster))

    for nodeID := range e.nodeCluster {
        node, err := e.storage.GetNode(nodeID)
        if err != nil {
            continue
        }
        embeddings = append(embeddings, node.Embedding)
        nodeIDs = append(nodeIDs, nodeID)
    }
    e.mu.RUnlock()

    // Full k-means on GPU (10-100ms)
    if err := e.gpuKMeans.Fit(embeddings); err != nil {
        return
    }

    // Rebuild lookup tables
    assignments := e.gpuKMeans.GetClusterAssignments()

    e.mu.Lock()
    defer e.mu.Unlock()

    e.nodeCluster = make(map[storage.NodeID]int)
    e.clusterNodes = make(map[int][]storage.NodeID)

    for i, nodeID := range nodeIDs {
        cluster := assignments[i]
        e.nodeCluster[nodeID] = cluster
        e.clusterNodes[cluster] = append(e.clusterNodes[cluster], nodeID)
    }

    e.centroids = e.gpuKMeans.GetClusterCentroids()
    e.lastRecluster = time.Now()
    e.updateCount = 0
    e.centroidDrift = make(map[int]float32) // Reset drift tracking
}

// shouldRecluster decides when to trigger full re-clustering
func (e *RealtimeClusteringEngine) shouldRecluster() bool {
    // Trigger if:

    // 1. Too many updates since last re-cluster
    if e.updateCount >= e.reclusterEvery {
        return true
    }

    // 2. Too much time elapsed
    if time.Since(e.lastRecluster) > e.maxAge {
        return true
    }

    // 3. Centroid drift exceeded threshold
    maxDrift := float32(0.0)
    for _, drift := range e.centroidDrift {
        if drift > maxDrift {
            maxDrift = drift
        }
    }

    if maxDrift > e.driftThreshold {
        return true
    }

    return false
}

// ========================================================================
// Query Interface
// ========================================================================

// GetRelatedNodes returns nodes in same/similar clusters (instant lookup)
func (e *RealtimeClusteringEngine) GetRelatedNodes(
    nodeID storage.NodeID,
    maxNodes int,
) []storage.NodeID {
    e.mu.RLock()
    defer e.mu.RUnlock()

    cluster, exists := e.nodeCluster[nodeID]
    if !exists {
        return nil
    }

    nodes := e.clusterNodes[cluster]

    if len(nodes) <= maxNodes {
        return nodes
    }

    // If cluster too large, refine with vector similarity
    return e.rankByProximity(nodeID, nodes[:maxNodes*2])[:maxNodes]
}

// GetClusterMembers returns all nodes in a cluster
func (e *RealtimeClusteringEngine) GetClusterMembers(clusterID int) []storage.NodeID {
    e.mu.RLock()
    defer e.mu.RUnlock()

    return e.clusterNodes[clusterID]
}

// GetNodeCluster returns the cluster ID for a node
func (e *RealtimeClusteringEngine) GetNodeCluster(nodeID storage.NodeID) (int, bool) {
    e.mu.RLock()
    defer e.mu.RUnlock()

    cluster, exists := e.nodeCluster[nodeID]
    return cluster, exists
}

// GetSimilarClusters finds clusters with similar centroids
func (e *RealtimeClusteringEngine) GetSimilarClusters(
    clusterID int,
    topK int,
) []int {
    e.mu.RLock()
    defer e.mu.RUnlock()

    if clusterID < 0 || clusterID >= len(e.centroids) {
        return nil
    }

    targetCentroid := e.centroids[clusterID]

    // Compute similarity to all centroids
    similarities := make([]struct {
        ClusterID  int
        Similarity float32
    }, 0, len(e.centroids))

    for i, centroid := range e.centroids {
        if i == clusterID {
            continue
        }

        sim := cosineSimilarity(targetCentroid, centroid)
        similarities = append(similarities, struct {
            ClusterID  int
            Similarity float32
        }{i, sim})
    }

    // Sort by similarity
    sort.Slice(similarities, func(i, j int) bool {
        return similarities[i].Similarity > similarities[j].Similarity
    })

    // Return top-K
    result := make([]int, 0, topK)
    for i := 0; i < topK && i < len(similarities); i++ {
        result = append(result, similarities[i].ClusterID)
    }

    return result
}

// GetStats returns clustering statistics
func (e *RealtimeClusteringEngine) GetStats() map[string]interface{} {
    e.mu.RLock()
    defer e.mu.RUnlock()

    clusterSizes := make([]int, 0, len(e.clusterNodes))
    for _, nodes := range e.clusterNodes {
        clusterSizes = append(clusterSizes, len(nodes))
    }

    return map[string]interface{}{
        "num_clusters":        len(e.clusterNodes),
        "total_nodes":         len(e.nodeCluster),
        "cluster_sizes":       clusterSizes,
        "avg_cluster_size":    average(clusterSizes),
        "updates_since_recluster": e.updateCount,
        "last_recluster":      e.lastRecluster,
        "pending_updates":     len(e.pendingUpdates),
    }
}

// ========================================================================
// Helper Functions
// ========================================================================

func (e *RealtimeClusteringEngine) addToCluster(cluster int, nodeID storage.NodeID) {
    e.clusterNodes[cluster] = append(e.clusterNodes[cluster], nodeID)
}

func (e *RealtimeClusteringEngine) removeFromCluster(cluster int, nodeID storage.NodeID) {
    nodes := e.clusterNodes[cluster]
    for i, id := range nodes {
        if id == nodeID {
            e.clusterNodes[cluster] = append(nodes[:i], nodes[i+1:]...)
            break
        }
    }
}

func (e *RealtimeClusteringEngine) computeDrift(old, new []float32) float32 {
    sum := float32(0.0)
    for i := range old {
        diff := old[i] - new[i]
        sum += diff * diff
    }
    return float32(math.Sqrt(float64(sum)))
}

func (e *RealtimeClusteringEngine) rankByProximity(
    targetID storage.NodeID,
    candidates []storage.NodeID,
) []storage.NodeID {
    target, _ := e.storage.GetNode(targetID)

    type scored struct {
        NodeID storage.NodeID
        Score  float32
    }

    scores := make([]scored, 0, len(candidates))
    for _, candID := range candidates {
        if candID == targetID {
            continue
        }

        cand, _ := e.storage.GetNode(candID)
        sim := cosineSimilarity(target.Embedding, cand.Embedding)
        scores = append(scores, scored{candID, sim})
    }

    sort.Slice(scores, func(i, j int) bool {
        return scores[i].Score > scores[j].Score
    })

    result := make([]storage.NodeID, len(scores))
    for i, s := range scores {
        result[i] = s.NodeID
    }

    return result
}

func cosineSimilarity(a, b []float32) float32 {
    dot := float32(0.0)
    normA := float32(0.0)
    normB := float32(0.0)

    for i := range a {
        dot += a[i] * b[i]
        normA += a[i] * a[i]
        normB += b[i] * b[i]
    }

    return dot / (float32(math.Sqrt(float64(normA))) * float32(math.Sqrt(float64(normB))))
}

func average(values []int) float64 {
    if len(values) == 0 {
        return 0
    }
    sum := 0
    for _, v := range values {
        sum += v
    }
    return float64(sum) / float64(len(values))
}

// Cleanup releases GPU resources
func (e *RealtimeClusteringEngine) Cleanup() {
    e.gpuKMeans.Cleanup()
}

# nornicdb.example.yaml

realtime_clustering:
  # Enable/disable real-time clustering
  enabled: true

  # Device selection
  device: "auto"  # auto, cuda, metal, cpu

  # Clustering parameters
  num_clusters: 500
  dimensions: 1024
  max_iterations: 300

  # Tier 1: Always-on instant reassignment
  reassign_on_update: true

  # Tier 2: Batch centroid updates
  centroid_update_batch_size: 100
  centroid_update_enabled: true

  # Tier 3: Full re-clustering
  recluster_every_n_updates: 10000
  recluster_max_age: "1h"  # Max time between re-clusters
  recluster_drift_threshold: 0.1  # Re-cluster if centroids drift >10%
  recluster_schedule: "0 * * * *"  # Cron: every hour

  # Memory management
  keep_embeddings_in_gpu: true
  max_gpu_memory_mb: 4096

  # Monitoring
  enable_drift_detection: true
  log_cluster_changes: true
  track_cluster_stats: true

# Enable real-time clustering
export NORNICDB_REALTIME_CLUSTERING_ENABLED=true

# Device selection
export NORNICDB_CLUSTERING_DEVICE=cuda  # cuda, metal, cpu, auto

# Tuning parameters
export NORNICDB_CLUSTERING_BATCH_SIZE=100
export NORNICDB_CLUSTERING_RECLUSTER_EVERY=10000
export NORNICDB_CLUSTERING_DRIFT_THRESHOLD=0.1
export NORNICDB_KMEANS_MIN_EMBEDDINGS=1000  # Minimum embeddings before clustering

import "github.com/orneryd/nornicdb/pkg/inference"

cfg := inference.Config{
    NumClusters:    500,
    Dimensions:     1024,
    BatchSize:      100,
    ReclusterEvery: 10000,
    DriftThreshold: 0.1,
    MaxAge:         1 * time.Hour,
    Device:         "auto",
}

clustering, err := inference.NewRealtimeClusteringEngine(storage, cfg)
if err != nil {
    log.Fatal(err)
}

// Initialize with current data
clustering.Initialize(ctx)

// Hook into update events
engine.OnNodeUpdate = func(nodeID, oldEmb, newEmb) {
    clustering.OnNodeUpdate(ctx, nodeID, oldEmb, newEmb)
}

type ClusteringMetrics struct {
    // Performance metrics
    ReassignmentLatency   time.Duration
    CentroidUpdateLatency time.Duration
    ReclusterLatency      time.Duration

    // Volume metrics
    TotalUpdates          int64
    UpdatesSinceRecluster int64
    ReassignmentsPerSec   float64

    // Quality metrics
    AverageDrift          float32
    MaxDrift              float32
    ClusterBalance        float32  // Std dev of cluster sizes

    // Cluster changes
    ClusterChanges        int64  // Nodes that changed clusters
    ClusterChangeRate     float64
}

func (e *RealtimeClusteringEngine) GetMetrics() ClusteringMetrics {
    // Return current metrics
}

// Log significant events
func (e *RealtimeClusteringEngine) OnNodeUpdate(...) {
    // Log cluster changes
    if newCluster != oldCluster {
        log.Info("node changed cluster",
            "node_id", nodeID,
            "old_cluster", oldCluster,
            "new_cluster", newCluster,
            "drift", drift,
        )
    }

    // Log batch updates
    if len(e.pendingUpdates) == e.batchSize {
        log.Debug("triggering batch centroid update",
            "batch_size", len(e.pendingUpdates),
            "affected_clusters", len(clusterUpdates),
        )
    }

    // Log re-clustering
    if e.shouldRecluster() {
        log.Info("triggering full re-clustering",
            "updates_since_last", e.updateCount,
            "time_since_last", time.Since(e.lastRecluster),
            "max_drift", maxDrift,
        )
    }
}

// Define alert conditions
type AlertCondition struct {
    Name      string
    Threshold float64
    Check     func(metrics ClusteringMetrics) float64
}

var alerts = []AlertCondition{
    {
        Name:      "high_cluster_change_rate",
        Threshold: 0.1,  // >10% of nodes changing clusters
        Check: func(m ClusteringMetrics) float64 {
            return m.ClusterChangeRate
        },
    },
    {
        Name:      "excessive_drift",
        Threshold: 0.2,  // Max drift >20%
        Check: func(m ClusteringMetrics) float64 {
            return float64(m.MaxDrift)
        },
    },
    {
        Name:      "cluster_imbalance",
        Threshold: 2.0,  // Std dev >2x mean
        Check: func(m ClusteringMetrics) float64 {
            return float64(m.ClusterBalance)
        },
    },
}

type HierarchicalCluster struct {
    topLevel    *RealtimeClusteringEngine  // 100 broad topics
    midLevel    map[int]*RealtimeClusteringEngine  // 500 per top cluster
    bottomLevel map[int]*RealtimeClusteringEngine  // 5000 per mid cluster
}

func (h *HierarchicalCluster) OnNodeUpdate(nodeID, embedding) {
    // Update all levels
    topCluster := h.topLevel.ReassignNode(nodeID, embedding)
    midCluster := h.midLevel[topCluster].ReassignNode(nodeID, embedding)
    bottomCluster := h.bottomLevel[midCluster].ReassignNode(nodeID, embedding)
}

// Update centroids on every single update (no batching)
func (e *RealtimeClusteringEngine) OnNodeUpdateStreaming(
    nodeID, embedding,
) {
    cluster := e.Predict(embedding)

    // Update centroid immediately with exponential moving average
    alpha := 0.01  // Learning rate
    for i := range e.centroids[cluster] {
        e.centroids[cluster][i] = (1-alpha)*e.centroids[cluster][i] + 
                                    alpha*embedding[i]
    }
}

type MultiGPUCluster struct {
    gpus      []*GPUKMeans
    nodeToGPU map[storage.NodeID]int
}

func (m *MultiGPUCluster) OnNodeUpdate(nodeID, embedding) {
    // Route to appropriate GPU
    gpuID := hash(nodeID) % len(m.gpus)
    m.gpus[gpuID].ReassignNode(nodeID, embedding)
}

func (e *RealtimeClusteringEngine) adaptBatchSize() {
    updateRate := e.getRecentUpdateRate()

    if updateRate > 1000 {
        // High update rate: use larger batches
        e.batchSize = 200
    } else if updateRate < 100 {
        // Low update rate: use smaller batches
        e.batchSize = 50
    } else {
        // Medium update rate: standard batch size
        e.batchSize = 100
    }
}

func (e *RealtimeClusteringEngine) MonitorQuality() {
    // Compute silhouette score
    silhouette := e.computeSilhouetteScore()

    if silhouette < 0.3 {
        // Poor clustering quality - re-cluster with more clusters
        e.numClusters = int(float64(e.numClusters) * 1.5)
        e.reclusterAll()
    }
}

realtime_clustering:
  enabled: true
  device: "auto"
  num_clusters: 500

  # Aggressive: Update often
  centroid_update_batch_size: 50
  recluster_every_n_updates: 5000
  recluster_max_age: "30m"

  # Conservative: Update less often
  # centroid_update_batch_size: 200
  # recluster_every_n_updates: 20000
  # recluster_max_age: "2h"