glpi-neural-brain/internal/engine/vector_maintenance_v10.go

package engine

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"github.com/local/glpi-neural-brain/internal/graph"
	"github.com/local/glpi-neural-brain/internal/model"
)

func (e *Engine) effectiveVectorLayoutMode() string {
	if e.Cfg.VectorGraphLayout {
		return "full"
	}
	if e.Cfg.VectorGraphRelaxLayout {
		return "relax"
	}
	return "off"
}

func (e *Engine) vectorMaintenanceDue(now time.Time) (reevaluate bool, layout bool) {
	if !e.Cfg.VectorGraphEnabled {
		return false, false
	}
	e.stateMu.RLock()
	lastGraph := e.lastVectorGraph
	lastLayout := e.lastVectorLayout
	maintenanceStartedAt := e.vectorMaintenanceStartedAt
	e.stateMu.RUnlock()
	reevaluate = !e.Graph.HasEdgesByOrigin(graph.VectorMathOrigin) || lastGraph.IsZero() || now.Sub(lastGraph) >= e.Cfg.VectorGraphReevaluateInterval
	// Full layout is intentionally coupled to semantic reevaluation. Relax mode
	// may run on its own slower/faster cadence without enabling the hard layout.
	if !e.Cfg.VectorGraphLayout && e.Cfg.VectorGraphRelaxLayout {
		if lastLayout.IsZero() {
			// Do not fake last_vector_layout merely to delay the first relaxation.
			// Keep telemetry truthful and use the process-start marker as a separate
			// not-before timestamp.
			layout = maintenanceStartedAt.IsZero() || now.Sub(maintenanceStartedAt) >= e.Cfg.VectorGraphLayoutRelaxInterval
		} else {
			layout = now.Sub(lastLayout) >= e.Cfg.VectorGraphLayoutRelaxInterval
		}
	}
	return reevaluate, layout
}

func (e *Engine) vectorMaintenanceLoop(ctx context.Context) {
	// Give bootstrap-owned scans a short head start. If Learning is disabled the
	// loop still runs and can maintain an already persisted vector layer.
	initialDelay := 5 * time.Second
	if e.SpeedModeEnabled() {
		initialDelay = 0
	}
	initial := time.NewTimer(initialDelay)
	defer initial.Stop()
	select {
	case <-ctx.Done():
		return
	case <-initial.C:
	}
	if err := e.runVectorMaintenance(ctx, "startup-maintenance"); err != nil {
		slog.Warn("vector maintenance startup run failed", "error", err)
	}
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := e.runVectorMaintenance(ctx, "scheduled-maintenance"); err != nil {
				slog.Warn("vector maintenance run failed", "error", err)
			}
		}
	}
}

func (e *Engine) runVectorMaintenance(ctx context.Context, trigger string) error {
	reevaluateDue, layoutDue := e.vectorMaintenanceDue(time.Now().UTC())
	if !reevaluateDue && !layoutDue {
		return nil
	}
	// Serialize with Learning Scan. Both operations may replace vector-math
	// edges/positions and must never race against each other.
	e.mu.Lock()
	defer e.mu.Unlock()
	reevaluateDue, layoutDue = e.vectorMaintenanceDue(time.Now().UTC())
	if !reevaluateDue && !layoutDue {
		return nil
	}
	if e.Graph.CountVectorsByDimension(256) != 0 {
		if e.Broker != nil {
			e.Broker.Publish(model.Activity{Type: "vector.graph.maintenance.skipped", Source: "brain", Phase: "semantic-linking", Message: "Vector-Maintenance wartet auf echte Embeddings; 256D-Fallback-Vektoren werden nicht mit dem Produktionsraum gemischt", Strength: .28, Metadata: map[string]any{"trigger": trigger, "reason": "fallback_vectors_present"}})
		}
		return nil
	}
	entries := e.Graph.VectorSemanticEntries(e.effectiveLearningFilter())
	if len(entries) < 2 {
		return nil
	}
	started := time.Now().UTC()
	runID := fmt.Sprintf("vector-maintenance-%d", started.UnixNano())
	if e.Broker != nil {
		e.Broker.Publish(model.Activity{Type: "vector.graph.maintenance.started", Source: "brain", Phase: "semantic-linking", Message: "Periodische mathematische Nähepflege wurde gestartet", Strength: .4, Metadata: map[string]any{
			"run_id": runID, "trigger": trigger, "reevaluate_due": reevaluateDue, "layout_due": layoutDue, "effective_layout_mode": e.effectiveVectorLayoutMode(), "learning_enabled": e.LearningEnabled(),
		}})
	}
	execution, err := e.rebuildVectorSemanticLayer(ctx, runID, e.effectiveLearningFilter())
	if err != nil {
		if e.Broker != nil {
			e.Broker.Publish(model.Activity{Type: "vector.graph.maintenance.failed", Source: "brain", Phase: "semantic-linking", Message: "Periodische mathematische Nähepflege ist fehlgeschlagen", Strength: .32, Metadata: map[string]any{"run_id": runID, "trigger": trigger, "error": err.Error(), "agent_required": e.Cfg.VectorGraphAgentRequired}})
		}
		return err
	}
	now := time.Now().UTC()
	e.stateMu.Lock()
	e.lastVectorGraph = now
	e.stateMu.Unlock()
	stats := execution.Stats
	if e.Broker != nil {
		meta := withRunMutations(map[string]any{
			"run_id": runID, "trigger": trigger, "algorithm": "mutual-knn-local-scaling-v1", "orphan_algorithm": "orphan-knn-local-scaling-v1", "no_model_call": true,
			"indexed": stats.Indexed, "links": stats.Links, "reciprocal_links": stats.ReciprocalLinks, "candidate_pairs": stats.CandidatePairs, "exact_comparisons": stats.ExactComparisons,
			"orphan_pass_enabled": e.Cfg.VectorGraphOrphanPass, "orphan_focus": stats.OrphanFocus, "orphan_links": stats.OrphanLinks, "orphan_exact_comparisons": stats.OrphanStats.ExactComparisons, "orphan_candidate_pairs": stats.OrphanStats.CandidatePairs,
			"position_updates": stats.PositionUpdates, "layout_mode": execution.LayoutMode, "layout_applied": execution.LayoutDue && stats.PositionUpdates > 0, "periodic_refresh": reevaluateDue, "layout_refresh": layoutDue,
			"reevaluate_interval": e.Cfg.VectorGraphReevaluateInterval.String(), "layout_relax_interval": e.Cfg.VectorGraphLayoutRelaxInterval.String(), "learning_enabled": e.LearningEnabled(),
			"agent_offloaded": execution.Offloaded, "agent_id": execution.AgentID, "agent_compute_ms": execution.ComputeMS, "agent_fallback_reason": execution.FallbackReason, "speed_mode": e.SpeedModeEnabled(), "cpu_workers": e.vectorPrimaryConfig(execution.LayoutDue).Workers, "duration_ms": time.Since(started).Milliseconds(),
		}, execution.Mutations)
		e.Broker.Publish(model.Activity{Type: "vector.graph.rebuilt", Source: "brain", Phase: "semantic-linking", Message: fmt.Sprintf("Periodische Vektorpflege: %d Primär- + %d Orphan-Kanten aus %d Embeddings", stats.Links, stats.OrphanLinks, stats.Indexed), Strength: .58, Metadata: meta})
		e.Broker.Publish(model.Activity{Type: "vector.graph.maintenance.completed", Source: "brain", Phase: "semantic-linking", Message: fmt.Sprintf("Mathematische Nähepflege abgeschlossen · Layoutmodus %s · %d Positionsupdates", execution.LayoutMode, stats.PositionUpdates), Strength: .5, Metadata: meta})
	}
	return nil
}