package graph import ( "context" "database/sql" "encoding/json" "fmt" "math" "sort" "strings" "sync/atomic" "time" "github.com/local/glpi-neural-brain/internal/model" ) // MutationStats are monotonic, process-local counters for material changes to // the live graph. They deliberately distinguish creation, updates and removal // so the analysis dashboard can explain a run even when the net graph size did // not change. type MutationStats struct { NodesCreated uint64 `json:"nodes_created"` NodesUpdated uint64 `json:"nodes_updated"` NodesDeleted uint64 `json:"nodes_deleted"` EdgesCreated uint64 `json:"edges_created"` EdgesUpdated uint64 `json:"edges_updated"` EdgesDeleted uint64 `json:"edges_deleted"` VectorsCreated uint64 `json:"vectors_created"` VectorsUpdated uint64 `json:"vectors_updated"` VectorsDeleted uint64 `json:"vectors_deleted"` } func (m MutationStats) Delta(previous MutationStats) MutationStats { return MutationStats{ NodesCreated: safeDelta(m.NodesCreated, previous.NodesCreated), NodesUpdated: safeDelta(m.NodesUpdated, previous.NodesUpdated), NodesDeleted: safeDelta(m.NodesDeleted, previous.NodesDeleted), EdgesCreated: safeDelta(m.EdgesCreated, previous.EdgesCreated), EdgesUpdated: safeDelta(m.EdgesUpdated, previous.EdgesUpdated), EdgesDeleted: safeDelta(m.EdgesDeleted, previous.EdgesDeleted), VectorsCreated: safeDelta(m.VectorsCreated, previous.VectorsCreated), VectorsUpdated: safeDelta(m.VectorsUpdated, previous.VectorsUpdated), VectorsDeleted: safeDelta(m.VectorsDeleted, previous.VectorsDeleted), } } func (m *MutationStats) Add(other MutationStats) { m.NodesCreated += other.NodesCreated m.NodesUpdated += other.NodesUpdated m.NodesDeleted += other.NodesDeleted m.EdgesCreated += other.EdgesCreated m.EdgesUpdated += other.EdgesUpdated m.EdgesDeleted += other.EdgesDeleted m.VectorsCreated += other.VectorsCreated m.VectorsUpdated += other.VectorsUpdated m.VectorsDeleted += other.VectorsDeleted } func (m MutationStats) Empty() bool { return m.NodesCreated+m.NodesUpdated+m.NodesDeleted+ m.EdgesCreated+m.EdgesUpdated+m.EdgesDeleted+ m.VectorsCreated+m.VectorsUpdated+m.VectorsDeleted == 0 } func safeDelta(current, previous uint64) uint64 { if current < previous { return current } return current - previous } type GraphChange struct { ID int64 `json:"id,omitempty"` EventID string `json:"event_id,omitempty"` Timestamp time.Time `json:"timestamp"` ProcessID string `json:"process_id"` GraphVersion uint64 `json:"graph_version"` EntityKind string `json:"entity_kind"` Action string `json:"action"` EntityID string `json:"entity_id"` Label string `json:"label,omitempty"` RelationType string `json:"relation_type,omitempty"` Origin string `json:"origin,omitempty"` Details map[string]any `json:"details,omitempty"` } type AnalysisPoint struct { ProcessID string `json:"process_id"` GraphVersion uint64 `json:"graph_version"` NodeCount int `json:"node_count"` EdgeCount int `json:"edge_count"` VectorCount int `json:"vector_count"` Mutations MutationStats `json:"mutations"` Delta MutationStats `json:"delta"` } type AnalysisEventRecord struct { Activity model.Activity `json:"activity"` Point AnalysisPoint `json:"point"` ChangeCount int `json:"change_count"` ChangesTruncated int `json:"changes_truncated,omitempty"` } type AnalysisTimelineBucket struct { Start time.Time `json:"start"` Events int `json:"events"` Successes int `json:"successes"` Warnings int `json:"warnings"` Failures int `json:"failures"` Comparisons int64 `json:"comparisons"` SearchResults int64 `json:"search_results"` Mutations MutationStats `json:"mutations"` } type AnalysisSecurityLifecycle struct { InboxID string `json:"inbox_id"` RunID string `json:"run_id"` Title string `json:"title"` Status string `json:"status"` ProactiveState string `json:"proactive_state"` Outcome string `json:"outcome"` LastError string `json:"last_error,omitempty"` MaterializedNodeID string `json:"materialized_node_id,omitempty"` StartedAt time.Time `json:"started_at,omitempty"` CompletedAt time.Time `json:"completed_at,omitempty"` DurationMS int64 `json:"duration_ms,omitempty"` Confidence float64 `json:"confidence,omitempty"` Severity string `json:"severity,omitempty"` EventType string `json:"event_type,omitempty"` Mutations MutationStats `json:"mutations"` } type AnalysisRun struct { ID string `json:"id"` Kind string `json:"kind"` Title string `json:"title"` Status string `json:"status"` Verdict string `json:"verdict"` Explanation string `json:"explanation"` StartedAt time.Time `json:"started_at"` CompletedAt time.Time `json:"completed_at,omitempty"` DurationMS int64 `json:"duration_ms"` Trigger string `json:"trigger,omitempty"` Outcome string `json:"outcome,omitempty"` EventCount int `json:"event_count"` Mutations MutationStats `json:"mutations"` MutationsKnown bool `json:"mutations_known"` MutationAttribution string `json:"mutation_attribution,omitempty"` NodeIDs []string `json:"node_ids,omitempty"` EdgeIDs []string `json:"edge_ids,omitempty"` Metrics map[string]any `json:"metrics,omitempty"` Events []AnalysisEventRecord `json:"events,omitempty"` } type DetailedGraphAnalysis struct { Summary model.GraphAnalysis `json:"summary"` NodeKinds map[string]int `json:"node_kinds"` NodeStatuses map[string]int `json:"node_statuses"` NodeOrigins map[string]int `json:"node_origins"` NodeSources map[string]int `json:"node_sources"` EdgeTypes map[string]int `json:"edge_types"` EdgeStatuses map[string]int `json:"edge_statuses"` EdgeOrigins map[string]int `json:"edge_origins"` VectorRows int `json:"vector_rows"` VectorEligibleNodes int `json:"vector_eligible_nodes"` VectorCoverage float64 `json:"vector_coverage"` EmbeddingDimensions map[string]int `json:"embedding_dimensions"` AverageAIConfidence float64 `json:"average_ai_confidence"` AverageAISimilarity float64 `json:"average_ai_similarity"` SimilarityEdgeCount int `json:"similarity_edge_count"` NewestNodes []model.Node `json:"newest_nodes"` NewestEdges []model.Edge `json:"newest_edges"` CurrentProcessChange MutationStats `json:"current_process_changes"` } type AnalysisAuditStatus struct { QueueDepth int `json:"queue_depth"` QueueCapacity int `json:"queue_capacity"` DroppedEvents uint64 `json:"dropped_events"` QueueDroppedEvents uint64 `json:"queue_dropped_events"` PersistDroppedEvents uint64 `json:"persist_dropped_events"` LastDropReason string `json:"last_drop_reason,omitempty"` LastDropAt time.Time `json:"last_drop_at,omitempty"` LastPersistedAt time.Time `json:"last_persisted_at,omitempty"` LastError string `json:"last_error,omitempty"` } type AnalysisEventSelection struct { PersistedEvents int `json:"persisted_events"` ReturnedEvents int `json:"returned_events"` MeaningfulEvents int `json:"meaningful_events"` UnchangedScanRuns int `json:"unchanged_scan_runs"` LegacyScanRunsCompacted int `json:"legacy_scan_runs_compacted"` PersistedScanAggregates int `json:"persisted_scan_aggregates"` EquivalentScanRawEvents int `json:"equivalent_scan_raw_events"` EmbeddingBatchEventsCompacted int `json:"embedding_batch_events_compacted"` PersistedEmbeddingAggregates int `json:"persisted_embedding_aggregates"` EquivalentEmbeddingRawEvents int `json:"equivalent_embedding_raw_events"` AvoidedPersistedEvents int `json:"avoided_persisted_events"` DisplayLimitOmitted int `json:"display_limit_omitted"` OldestReturnedAt time.Time `json:"oldest_returned_at,omitempty"` NewestReturnedAt time.Time `json:"newest_returned_at,omitempty"` } type AnalysisDurationStats struct { Samples int `json:"samples"` TotalMS int64 `json:"total_ms"` AverageMS int64 `json:"average_ms"` P50MS int64 `json:"p50_ms"` P95MS int64 `json:"p95_ms"` MaxMS int64 `json:"max_ms"` } type AnalysisRunStats struct { Kind string `json:"kind"` Runs int `json:"runs"` Successes int `json:"successes"` Warnings int `json:"warnings"` Failures int `json:"failures"` Neutral int `json:"neutral"` Running int `json:"running"` EventCount int `json:"event_count"` MutationSamples int `json:"mutation_samples"` Duration AnalysisDurationStats `json:"duration"` Mutations MutationStats `json:"mutations"` } type AnalysisSecuritySummary struct { Materialized int `json:"materialized"` Rejected int `json:"rejected"` Failed int `json:"failed"` ResearchSupplements int `json:"research_supplements"` SupplementalSources int `json:"supplemental_sources"` AverageConfidence float64 `json:"average_confidence"` ConfidenceSamples int `json:"confidence_samples"` Severities map[string]int `json:"severities"` EventTypes map[string]int `json:"event_types"` AuthoritativeRecords int `json:"authoritative_records"` ReconciledRuns int `json:"reconciled_runs"` } type AnalysisArticleSummary struct { Created int `json:"created"` Rejected int `json:"rejected"` Failed int `json:"failed"` Duplicates int `json:"duplicates"` Skipped int `json:"skipped"` Reviews int `json:"reviews"` SupportedClaims int `json:"supported_claims"` PartialClaims int `json:"partially_supported_claims"` UnsupportedClaims int `json:"unsupported_claims"` ContradictedClaims int `json:"contradicted_claims"` InboxResearchHits int `json:"inbox_research_hits"` WebResearchFetches int `json:"web_research_fetches"` } type AnalysisPipelineSummary struct { Security AnalysisSecuritySummary `json:"security"` Articles AnalysisArticleSummary `json:"articles"` } type AnalysisHistory struct { GeneratedAt time.Time `json:"generated_at"` Since time.Time `json:"since"` Events []AnalysisEventRecord `json:"events"` Runs []AnalysisRun `json:"runs"` Timeline []AnalysisTimelineBucket `json:"timeline"` Changes []GraphChange `json:"changes"` Totals MutationStats `json:"totals"` EventCounts map[string]int `json:"event_counts"` StatusCounts map[string]int `json:"status_counts"` DroppedEvents uint64 `json:"dropped_events"` DroppedDetailedChanges uint64 `json:"dropped_detailed_changes"` DetailedChangesTruncated uint64 `json:"detailed_changes_truncated"` RawEventCount int `json:"raw_event_count"` ChangeCount int `json:"change_count"` Audit AnalysisAuditStatus `json:"audit"` EventSelection AnalysisEventSelection `json:"event_selection"` RunStats []AnalysisRunStats `json:"run_stats"` Pipelines AnalysisPipelineSummary `json:"pipelines"` } type analysisRecord struct { activity model.Activity point AnalysisPoint changes []GraphChange changesTruncated int } // analysisLearningScanAggregate keeps high-frequency no-op KB scans from // flooding the persisted analysis stream. A full scan may run every few // seconds, but when it changes absolutely nothing there is little diagnostic // value in storing a start/completed pair for every invocation. We retain the // exact number of runs and their timing as one compact summary event. type analysisLearningScanAggregate struct { Count int FirstAt time.Time LastAt time.Time TotalDurationMS int64 MinDurationMS int64 MaxDurationMS int64 KnowledgeElements int OllamaOK bool LastPoint AnalysisPoint } const ( analysisLearningScanAggregateWindow = 5 * time.Minute analysisLearningScanAggregateCount = 15 analysisEmbeddingAggregateWindow = 60 * time.Second analysisEmbeddingAggregateCount = 16 ) // analysisEmbeddingBatchAggregate preserves exact graph mutation accounting // while collapsing repetitive embedding progress events. Detailed vector // changes remain attached to the compact aggregate record. type analysisEmbeddingBatchAggregate struct { EventCount int ElementCount int FirstAt time.Time LastAt time.Time Model string LastPoint AnalysisPoint Delta MutationStats Changes []GraphChange ChangesTruncated int NodeIDs []string TotalDurationMS int64 MinDurationMS int64 MaxDurationMS int64 } var processCounter atomic.Uint64 func newProcessID() string { return fmt.Sprintf("process-%d-%d", time.Now().UTC().UnixNano(), processCounter.Add(1)) } func (s *Store) initAnalysisWriter() { s.analysisMu.Lock() defer s.analysisMu.Unlock() if s.analysisQueue != nil { return } if strings.TrimSpace(s.processID) == "" { s.processID = newProcessID() } s.analysisQueue = make(chan analysisRecord, 16384) s.analysisWG.Add(1) go func(queue <-chan analysisRecord) { defer s.analysisWG.Done() for { first, ok := <-queue if !ok { return } batch := []analysisRecord{first} timer := time.NewTimer(50 * time.Millisecond) collect: for len(batch) < 64 { select { case record, open := <-queue: if !open { if !timer.Stop() { <-timer.C } lost, err := s.persistAnalysisResilient(batch) s.recordAnalysisPersistResult(err, lost) return } batch = append(batch, record) case <-timer.C: break collect } } if !timer.Stop() { select { case <-timer.C: default: } } lost, err := s.persistAnalysisResilient(batch) s.recordAnalysisPersistResult(err, lost) } }(s.analysisQueue) } func (s *Store) recordAnalysisPersistResult(err error, lost uint64) { s.analysisMu.Lock() defer s.analysisMu.Unlock() if lost > 0 { s.analysisDropped += lost s.analysisPersistDropped += lost s.analysisLastDropAt = time.Now().UTC() if err != nil { s.analysisLastDropReason = err.Error() } else { s.analysisLastDropReason = fmt.Sprintf("analysis persistence lost %d event(s)", lost) } } if err != nil { s.analysisLastError = err.Error() return } s.analysisLastError = "" s.analysisLastPersisted = time.Now().UTC() } // persistAnalysisResilient keeps the audit journal append-only. A batch-level // failure falls back to individual writes so one malformed/duplicate event // cannot discard otherwise valid telemetry from the same 50ms batch. func (s *Store) persistAnalysisResilient(records []analysisRecord) (uint64, error) { // The writer has a dedicated WAL connection, so a longer deadline protects // audit durability without starving normal graph reads/writes. The old 30s // deadline could expire merely while waiting for the Store's sole DB slot. ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) err := s.persistAnalysisBatch(ctx, records) cancel() if err == nil || len(records) <= 1 { if err != nil { return uint64(len(records)), err } return 0, nil } var lost uint64 var lastErr error for _, record := range records { itemCtx, itemCancel := context.WithTimeout(context.Background(), 45*time.Second) itemErr := s.persistAnalysisRecord(itemCtx, record) itemCancel() if itemErr != nil { lost++ lastErr = itemErr } } if lost > 0 { return lost, fmt.Errorf("analysis persistence lost %d/%d record(s) after batch retry: %w", lost, len(records), lastErr) } return 0, nil } // RecordActivity captures a cheap O(1) graph checkpoint synchronously and // performs the SQLite write asynchronously. The expensive graph analysis is // only calculated when the dedicated dashboard is opened. // // High-frequency unchanged learning scans are deliberately compacted. They // still remain visible as exact run counts and duration statistics, but no // longer evict article/security/research events from the useful history. func (s *Store) RecordActivity(activity model.Activity) { if s == nil || s.db == nil || activity.Type == "brain.idle" { return } if activity.Timestamp.IsZero() { activity.Timestamp = time.Now().UTC() } // A learning scan normally finishes within a few seconds and its terminal // event already contains duration/result metadata. Persisting every start // marker doubled the audit noise without adding post-hoc information. if activity.Type == "learning.scan.started" { return } s.mu.Lock() nodes := len(s.nodes) edges := len(s.edges) point := AnalysisPoint{ ProcessID: s.processID, GraphVersion: s.version, NodeCount: nodes, EdgeCount: edges, VectorCount: len(s.vectors), Mutations: s.mutations, } changes := append([]GraphChange(nil), s.analysisPendingChanges...) changesTruncated := s.analysisPendingTruncated s.analysisPendingChanges = s.analysisPendingChanges[:0] s.analysisPendingTruncated = 0 s.mu.Unlock() record := analysisRecord{activity: activity, point: point, changes: changes, changesTruncated: changesTruncated} s.analysisMu.Lock() if s.analysisQueue == nil { s.analysisMu.Unlock() return } point.Delta = point.Mutations.Delta(s.analysisLastMutations) s.analysisLastMutations = point.Mutations record.point = point if activity.Type == "embedding.batch" { s.addEmbeddingBatchAggregateLocked(record) if flush := s.embeddingBatchAggregateReadyLocked(activity.Timestamp); flush != nil { s.enqueueAnalysisLocked(*flush) } s.analysisMu.Unlock() return } if isUnchangedLearningScan(activity, record) { if flush := s.flushEmbeddingBatchAggregateLocked(); flush != nil { s.enqueueAnalysisLocked(*flush) } s.addLearningScanAggregateLocked(record) flush := s.learningScanAggregateReadyLocked(activity.Timestamp) if flush != nil { s.enqueueAnalysisLocked(*flush) } s.analysisMu.Unlock() return } // Flush compact background telemetry before important work is persisted so // chronological exports remain easy to read even when aggregation windows // straddle a security/article event. if activity.Type != "learning.scan.unchanged.aggregate" { if flush := s.flushLearningScanAggregateLocked(); flush != nil { s.enqueueAnalysisLocked(*flush) } } if activity.Type != "embedding.batch.aggregate" { if flush := s.flushEmbeddingBatchAggregateLocked(); flush != nil { s.enqueueAnalysisLocked(*flush) } } queued := s.enqueueAnalysisLocked(record) s.analysisMu.Unlock() if !queued && len(changes)+changesTruncated > 0 { s.mu.Lock() s.analysisChangesDropped += uint64(len(changes) + changesTruncated) s.mu.Unlock() } } func isUnchangedLearningScan(activity model.Activity, record analysisRecord) bool { if activity.Type != "learning.scan.completed" || !strings.EqualFold(metadataString(activity.Metadata, "result"), "unchanged") { return false } if mutations, ok := explicitActivityMutations(activity); ok { return mutations.Empty() && len(record.changes) == 0 && record.changesTruncated == 0 } // Legacy fallback: old events had no causal counters. return record.point.Delta.Empty() && len(record.changes) == 0 && record.changesTruncated == 0 } func explicitActivityMutations(activity model.Activity) (MutationStats, bool) { if activity.Metadata == nil { return MutationStats{}, false } _, marker := activity.Metadata["mutation_attribution"] keys := []string{"run_nodes_created", "run_nodes_updated", "run_nodes_deleted", "run_edges_created", "run_edges_updated", "run_edges_deleted", "run_vectors_created", "run_vectors_updated", "run_vectors_deleted"} found := marker for _, key := range keys { if _, ok := activity.Metadata[key]; ok { found = true break } } if !found { return MutationStats{}, false } return MutationStats{ NodesCreated: uint64(metadataNumber(activity.Metadata, "run_nodes_created")), NodesUpdated: uint64(metadataNumber(activity.Metadata, "run_nodes_updated")), NodesDeleted: uint64(metadataNumber(activity.Metadata, "run_nodes_deleted")), EdgesCreated: uint64(metadataNumber(activity.Metadata, "run_edges_created")), EdgesUpdated: uint64(metadataNumber(activity.Metadata, "run_edges_updated")), EdgesDeleted: uint64(metadataNumber(activity.Metadata, "run_edges_deleted")), VectorsCreated: uint64(metadataNumber(activity.Metadata, "run_vectors_created")), VectorsUpdated: uint64(metadataNumber(activity.Metadata, "run_vectors_updated")), VectorsDeleted: uint64(metadataNumber(activity.Metadata, "run_vectors_deleted")), }, true } func (s *Store) addLearningScanAggregateLocked(record analysisRecord) { a := &s.analysisLearningScans if a.Count == 0 { a.FirstAt = record.activity.Timestamp a.MinDurationMS = -1 } a.Count++ a.LastAt = record.activity.Timestamp a.LastPoint = record.point duration := int64(metadataNumber(record.activity.Metadata, "duration_ms")) if duration > 0 { a.TotalDurationMS += duration if a.MinDurationMS < 0 || duration < a.MinDurationMS { a.MinDurationMS = duration } if duration > a.MaxDurationMS { a.MaxDurationMS = duration } } if count := int(metadataNumber(record.activity.Metadata, "knowledge_elements")); count > 0 { a.KnowledgeElements = count } if value, ok := record.activity.Metadata["ollama_ok"].(bool); ok { a.OllamaOK = value } } func (s *Store) learningScanAggregateReadyLocked(now time.Time) *analysisRecord { a := s.analysisLearningScans if a.Count == 0 { return nil } if a.Count < analysisLearningScanAggregateCount && now.Sub(a.FirstAt) < analysisLearningScanAggregateWindow { return nil } return s.flushLearningScanAggregateLocked() } func (s *Store) nextAnalysisAggregateIDLocked(kind string, at time.Time) string { s.analysisAggregateSeq++ return fmt.Sprintf("analysis-%s-%s-%d-%d", strings.TrimSpace(kind), s.processID, at.UnixNano(), s.analysisAggregateSeq) } func (s *Store) flushLearningScanAggregateLocked() *analysisRecord { a := s.analysisLearningScans if a.Count == 0 { return nil } avg := int64(0) if a.Count > 0 { avg = a.TotalDurationMS / int64(a.Count) } minDuration := a.MinDurationMS if minDuration < 0 { minDuration = 0 } activity := model.Activity{ ID: s.nextAnalysisAggregateIDLocked("learning-scan-aggregate", a.LastAt), Type: "learning.scan.unchanged.aggregate", Source: "brain", Phase: "indexed", Message: fmt.Sprintf("%d unveränderte KB-Lernläufe wurden im Analysejournal verdichtet", a.Count), Strength: .22, Timestamp: a.LastAt, Metadata: map[string]any{ "result": "unchanged_aggregated", "scan_count": a.Count, "equivalent_raw_events": a.Count * 2, "first_scan_at": a.FirstAt, "last_scan_at": a.LastAt, "total_duration_ms": a.TotalDurationMS, "avg_duration_ms": avg, "min_duration_ms": minDuration, "max_duration_ms": a.MaxDurationMS, "knowledge_elements": a.KnowledgeElements, "ollama_ok": a.OllamaOK, "mutation_attribution": "explicit", "run_nodes_created": 0, "run_nodes_updated": 0, "run_nodes_deleted": 0, "run_edges_created": 0, "run_edges_updated": 0, "run_edges_deleted": 0, "run_vectors_created": 0, "run_vectors_updated": 0, "run_vectors_deleted": 0, }, } point := a.LastPoint point.Delta = MutationStats{} s.analysisLearningScans = analysisLearningScanAggregate{} return &analysisRecord{activity: activity, point: point} } func (s *Store) addEmbeddingBatchAggregateLocked(record analysisRecord) { a := &s.analysisEmbeddingBatches if a.EventCount == 0 { a.FirstAt = record.activity.Timestamp a.MinDurationMS = -1 } a.EventCount++ a.LastAt = record.activity.Timestamp a.LastPoint = record.point if mutations, ok := explicitActivityMutations(record.activity); ok { a.Delta.Add(mutations) } a.ElementCount += int(metadataNumber(record.activity.Metadata, "batch_count")) if duration := int64(metadataNumber(record.activity.Metadata, "duration_ms")); duration >= 0 { a.TotalDurationMS += duration if a.MinDurationMS < 0 || duration < a.MinDurationMS { a.MinDurationMS = duration } if duration > a.MaxDurationMS { a.MaxDurationMS = duration } } if modelName := metadataString(record.activity.Metadata, "model"); modelName != "" { a.Model = modelName } a.NodeIDs = uniqueStrings(append(a.NodeIDs, record.activity.NodeIDs...)) remaining := analysisDetailedChangeLimit - len(a.Changes) if remaining > 0 { if len(record.changes) <= remaining { a.Changes = append(a.Changes, record.changes...) } else { a.Changes = append(a.Changes, record.changes[:remaining]...) a.ChangesTruncated += len(record.changes) - remaining } } else { a.ChangesTruncated += len(record.changes) } a.ChangesTruncated += record.changesTruncated } func (s *Store) embeddingBatchAggregateReadyLocked(now time.Time) *analysisRecord { a := s.analysisEmbeddingBatches if a.EventCount == 0 { return nil } if a.EventCount < analysisEmbeddingAggregateCount && now.Sub(a.FirstAt) < analysisEmbeddingAggregateWindow { return nil } return s.flushEmbeddingBatchAggregateLocked() } func (s *Store) flushEmbeddingBatchAggregateLocked() *analysisRecord { a := s.analysisEmbeddingBatches if a.EventCount == 0 { return nil } activity := model.Activity{ ID: s.nextAnalysisAggregateIDLocked("embedding-batch-aggregate", a.LastAt), Type: "embedding.batch.aggregate", Source: "ollama", Phase: "embedding", Message: fmt.Sprintf("%d Embedding-Batches mit %d Elementen wurden im Analysejournal verdichtet", a.EventCount, a.ElementCount), NodeIDs: append([]string(nil), a.NodeIDs...), Strength: .28, Timestamp: a.LastAt, Metadata: map[string]any{ "batch_events": a.EventCount, "batch_count": a.ElementCount, "equivalent_raw_events": a.EventCount, "first_batch_at": a.FirstAt, "last_batch_at": a.LastAt, "model": a.Model, "duration_ms": a.TotalDurationMS, "average_batch_duration_ms": func() int64 { if a.EventCount > 0 { return a.TotalDurationMS / int64(a.EventCount) } return 0 }(), "min_batch_duration_ms": a.MinDurationMS, "max_batch_duration_ms": a.MaxDurationMS, "mutation_attribution": "explicit", "run_nodes_created": a.Delta.NodesCreated, "run_nodes_updated": a.Delta.NodesUpdated, "run_nodes_deleted": a.Delta.NodesDeleted, "run_edges_created": a.Delta.EdgesCreated, "run_edges_updated": a.Delta.EdgesUpdated, "run_edges_deleted": a.Delta.EdgesDeleted, "run_vectors_created": a.Delta.VectorsCreated, "run_vectors_updated": a.Delta.VectorsUpdated, "run_vectors_deleted": a.Delta.VectorsDeleted, }, } point := a.LastPoint point.Delta = a.Delta record := analysisRecord{activity: activity, point: point, changes: append([]GraphChange(nil), a.Changes...), changesTruncated: a.ChangesTruncated} s.analysisEmbeddingBatches = analysisEmbeddingBatchAggregate{} return &record } func (s *Store) enqueueAnalysisLocked(record analysisRecord) bool { if s.analysisQueue == nil { return false } select { case s.analysisQueue <- record: return true default: s.analysisDropped++ s.analysisQueueDropped++ s.analysisLastDropAt = time.Now().UTC() s.analysisLastDropReason = "analysis audit queue full" return false } } func (s *Store) closeAnalysisWriter() { s.analysisMu.Lock() queue := s.analysisQueue if queue != nil { if flush := s.flushLearningScanAggregateLocked(); flush != nil { // Do not block while analysisMu is held; the writer records its persist // result under the same lock. The queue normally has ample capacity. s.enqueueAnalysisLocked(*flush) } if flush := s.flushEmbeddingBatchAggregateLocked(); flush != nil { s.enqueueAnalysisLocked(*flush) } s.analysisQueue = nil close(queue) } s.analysisMu.Unlock() if queue != nil { s.analysisWG.Wait() } } func (s *Store) persistAnalysisRecord(ctx context.Context, record analysisRecord) error { return s.persistAnalysisBatch(ctx, []analysisRecord{record}) } func (s *Store) persistAnalysisBatch(ctx context.Context, records []analysisRecord) error { if len(records) == 0 { return nil } db := s.analysisDB if db == nil { db = s.db } tx, err := db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() eventStatement, err := tx.PrepareContext(ctx, `INSERT INTO analysis_events(id,type,source,phase,query,message,node_ids_json,edge_ids_json,strength,metadata_json,timestamp_ns,process_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`) if err != nil { return err } defer eventStatement.Close() pointStatement, err := tx.PrepareContext(ctx, `INSERT INTO analysis_points(event_id,timestamp_ns,process_id,graph_version,node_count,edge_count,vector_count, node_created,node_updated,node_deleted,edge_created,edge_updated,edge_deleted,vector_created,vector_updated,vector_deleted, delta_node_created,delta_node_updated,delta_node_deleted,delta_edge_created,delta_edge_updated,delta_edge_deleted,delta_vector_created,delta_vector_updated,delta_vector_deleted, change_count,changes_truncated) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`) if err != nil { return err } defer pointStatement.Close() changeStatement, err := tx.PrepareContext(ctx, `INSERT INTO analysis_changes(event_id,timestamp_ns,process_id,graph_version,entity_kind,action,entity_id,label,relation_type,origin,details_json) VALUES(?,?,?,?,?,?,?,?,?,?,?)`) if err != nil { return err } defer changeStatement.Close() for _, record := range records { nodeIDs, _ := json.Marshal(record.activity.NodeIDs) edgeIDs, _ := json.Marshal(record.activity.EdgeIDs) metadata, _ := json.Marshal(record.activity.Metadata) if _, err := eventStatement.ExecContext(ctx, record.activity.ID, record.activity.Type, record.activity.Source, record.activity.Phase, record.activity.Query, record.activity.Message, string(nodeIDs), string(edgeIDs), record.activity.Strength, string(metadata), record.activity.Timestamp.UnixNano(), record.point.ProcessID); err != nil { return err } m, d := record.point.Mutations, record.point.Delta if _, err := pointStatement.ExecContext(ctx, record.activity.ID, record.activity.Timestamp.UnixNano(), record.point.ProcessID, record.point.GraphVersion, record.point.NodeCount, record.point.EdgeCount, record.point.VectorCount, m.NodesCreated, m.NodesUpdated, m.NodesDeleted, m.EdgesCreated, m.EdgesUpdated, m.EdgesDeleted, m.VectorsCreated, m.VectorsUpdated, m.VectorsDeleted, d.NodesCreated, d.NodesUpdated, d.NodesDeleted, d.EdgesCreated, d.EdgesUpdated, d.EdgesDeleted, d.VectorsCreated, d.VectorsUpdated, d.VectorsDeleted, len(record.changes), record.changesTruncated); err != nil { return err } for _, change := range record.changes { details, _ := json.Marshal(change.Details) timestamp := change.Timestamp if timestamp.IsZero() { timestamp = record.activity.Timestamp } if _, err := changeStatement.ExecContext(ctx, record.activity.ID, timestamp.UnixNano(), change.ProcessID, change.GraphVersion, change.EntityKind, change.Action, change.EntityID, change.Label, change.RelationType, change.Origin, string(details)); err != nil { return err } } } if err := tx.Commit(); err != nil { return err } last := records[len(records)-1] if last.activity.Timestamp.Unix()%997 == 0 { cutoff := time.Now().UTC().Add(-90 * 24 * time.Hour).UnixNano() _, _ = db.ExecContext(ctx, `DELETE FROM analysis_events WHERE timestamp_ns 0 { return s.analysisDetailCache } summary := s.Analyze() detail := DetailedGraphAnalysis{ Summary: summary, NodeKinds: map[string]int{}, NodeStatuses: map[string]int{}, NodeOrigins: map[string]int{}, NodeSources: map[string]int{}, EdgeTypes: map[string]int{}, EdgeStatuses: map[string]int{}, EdgeOrigins: map[string]int{}, EmbeddingDimensions: map[string]int{}, } s.mu.RLock() for _, node := range s.nodes { detail.NodeKinds[nonemptyAnalysis(node.Kind, "unknown")]++ detail.NodeStatuses[nonemptyAnalysis(node.Status, "unspecified")]++ detail.NodeOrigins[nonemptyAnalysis(node.Origin, "unknown")]++ source := explicitNodeSource(node) if source == "" { source = "(ohne source)" } detail.NodeSources[source]++ if node.Kind == "knowledge" || node.Kind == "ai-think" || node.Kind == "external" { detail.VectorEligibleNodes++ } detail.NewestNodes = insertNewestNode(detail.NewestNodes, node, 12) } var confidenceSum, similaritySum float64 for _, edge := range s.edges { if edge.Status == "rejected" { continue } detail.EdgeTypes[nonemptyAnalysis(edge.Type, "unknown")]++ detail.EdgeStatuses[nonemptyAnalysis(edge.Status, "unspecified")]++ detail.EdgeOrigins[nonemptyAnalysis(edge.Origin, "unknown")]++ if edge.Origin == "ai-inference" { confidenceSum += edge.Confidence if similarity, ok := numericMetadata(edge.Metadata, "semantic_similarity"); ok { similaritySum += similarity detail.SimilarityEdgeCount++ } } detail.NewestEdges = insertNewestEdge(detail.NewestEdges, edge, 12) } if summary.AIEdges > 0 { detail.AverageAIConfidence = confidenceSum / float64(summary.AIEdges) } if detail.SimilarityEdgeCount > 0 { detail.AverageAISimilarity = similaritySum / float64(detail.SimilarityEdgeCount) } detail.VectorRows = len(s.vectors) if detail.VectorEligibleNodes > 0 { detail.VectorCoverage = float64(detail.VectorRows) / float64(detail.VectorEligibleNodes) } for _, vector := range s.vectors { detail.EmbeddingDimensions[fmt.Sprint(len(vector))]++ } detail.CurrentProcessChange = s.mutations cachedVersion := s.version s.mu.RUnlock() s.analysisDetailVersion = cachedVersion s.analysisDetailCache = detail return detail } func insertNewestNode(nodes []model.Node, node model.Node, limit int) []model.Node { index := sort.Search(len(nodes), func(i int) bool { return !nodes[i].UpdatedAt.After(node.UpdatedAt) }) if index >= limit { return nodes } nodes = append(nodes, model.Node{}) copy(nodes[index+1:], nodes[index:]) nodes[index] = node if len(nodes) > limit { nodes = nodes[:limit] } return nodes } func insertNewestEdge(edges []model.Edge, edge model.Edge, limit int) []model.Edge { index := sort.Search(len(edges), func(i int) bool { return !edges[i].UpdatedAt.After(edge.UpdatedAt) }) if index >= limit { return edges } edges = append(edges, model.Edge{}) copy(edges[index+1:], edges[index:]) edges[index] = edge if len(edges) > limit { edges = edges[:limit] } return edges } func explicitNodeSource(node model.Node) string { if node.Metadata == nil { return "" } value, ok := node.Metadata["source"] if !ok || value == nil { return "" } text := strings.TrimSpace(fmt.Sprint(value)) if text == "" || strings.EqualFold(text, "") || strings.EqualFold(text, "null") { return "" } return text } func nonemptyAnalysis(value, fallback string) string { if strings.TrimSpace(value) == "" { return fallback } return value } func numericMetadata(metadata map[string]any, key string) (float64, bool) { if metadata == nil { return 0, false } value, ok := metadata[key] if !ok { return 0, false } switch typed := value.(type) { case float64: return typed, true case float32: return float64(typed), true case int: return float64(typed), true case int64: return float64(typed), true case json.Number: parsed, err := typed.Float64() return parsed, err == nil default: return 0, false } } type analysisScanCompactionStats struct { StartedRaw int RawUnchanged int AggregateRuns int AggregateRecords int EquivalentRaw int RawEmbeddingBatches int EmbeddingAggregateEvents int EmbeddingAggregateRecords int EquivalentEmbeddingRaw int AvoidedPersisted int } func (s *Store) AnalysisHistory(ctx context.Context, since time.Time, limit int) (AnalysisHistory, error) { if limit < 20 { limit = 200 } if limit > 1000 { limit = 1000 } if since.IsZero() { since = time.Now().UTC().Add(-24 * time.Hour) } // Fetch a larger pool of meaningful events than the UI displays. No-op scan // noise is excluded at SQL level, so article/security/research runs from many // hours ago are not displaced by a 20-second scan cadence. fetchLimit := limit * 20 if fetchLimit < 4000 { fetchLimit = 4000 } if fetchLimit > 30000 { fetchLimit = 30000 } meaningful, err := s.analysisMeaningfulEvents(ctx, since, fetchLimit) if err != nil { return AnalysisHistory{}, err } legacyScans, err := s.analysisLegacyLearningScanAggregates(ctx, since) if err != nil { return AnalysisHistory{}, err } legacyEmbeddings, err := s.analysisLegacyEmbeddingAggregates(ctx, since) if err != nil { return AnalysisHistory{}, err } combined := append(meaningful, legacyScans...) combined = append(combined, legacyEmbeddings...) sort.Slice(combined, func(i, j int) bool { return combined[i].Activity.Timestamp.After(combined[j].Activity.Timestamp) }) chronological := append([]AnalysisEventRecord(nil), combined...) sort.SliceStable(chronological, func(i, j int) bool { a, b := chronological[i].Activity, chronological[j].Activity if a.Timestamp.Equal(b.Timestamp) { pa, pb := analysisEventOrder(a), analysisEventOrder(b) if pa != pb { return pa < pb } return a.ID < b.ID } return a.Timestamp.Before(b.Timestamp) }) allRuns := buildAnalysisRuns(chronological) runStats := buildAnalysisRunStats(allRuns) runs := append([]AnalysisRun(nil), allRuns...) sort.Slice(runs, func(i, j int) bool { return runs[i].StartedAt.After(runs[j].StartedAt) }) if len(runs) > limit { runs = runs[:limit] } events := combined if len(events) > limit { events = events[:limit] } changeLimit := limit * 10 if changeLimit < 500 { changeLimit = 500 } if changeLimit > 10000 { changeLimit = 10000 } changes, changeCount, changeErr := s.analysisChanges(ctx, since, changeLimit) if changeErr != nil { return AnalysisHistory{}, changeErr } totals, rawCount, persistedTruncated, counts, aggregateErr := s.analysisAggregate(ctx, since) if aggregateErr != nil { return AnalysisHistory{}, aggregateErr } compaction, err := s.analysisLearningScanCompactionStats(ctx, since) if err != nil { return AnalysisHistory{}, err } statusCounts := map[string]int{} for _, event := range chronological { statusCounts[eventSeverity(event.Activity)]++ } timeline := buildTimeline(chronological, since) s.analysisMu.Lock() dropped := s.analysisDropped audit := AnalysisAuditStatus{ DroppedEvents: dropped, QueueDroppedEvents: s.analysisQueueDropped, PersistDroppedEvents: s.analysisPersistDropped, LastDropReason: s.analysisLastDropReason, LastDropAt: s.analysisLastDropAt, LastPersistedAt: s.analysisLastPersisted, LastError: s.analysisLastError, } if s.analysisQueue != nil { audit.QueueDepth = len(s.analysisQueue) audit.QueueCapacity = cap(s.analysisQueue) } s.analysisMu.Unlock() s.mu.RLock() droppedChanges := s.analysisChangesDropped s.mu.RUnlock() selection := AnalysisEventSelection{ PersistedEvents: rawCount, ReturnedEvents: len(events), MeaningfulEvents: maxIntAnalysis(0, rawCount-compaction.StartedRaw-compaction.RawUnchanged-compaction.RawEmbeddingBatches), UnchangedScanRuns: compaction.RawUnchanged + compaction.AggregateRuns, LegacyScanRunsCompacted: compaction.RawUnchanged, PersistedScanAggregates: compaction.AggregateRecords, EquivalentScanRawEvents: compaction.RawUnchanged*2 + compaction.EquivalentRaw, EmbeddingBatchEventsCompacted: compaction.RawEmbeddingBatches + compaction.EmbeddingAggregateEvents, PersistedEmbeddingAggregates: compaction.EmbeddingAggregateRecords, EquivalentEmbeddingRawEvents: compaction.RawEmbeddingBatches + compaction.EquivalentEmbeddingRaw, AvoidedPersistedEvents: compaction.AvoidedPersisted, } if len(events) > 0 { selection.NewestReturnedAt = events[0].Activity.Timestamp selection.OldestReturnedAt = events[len(events)-1].Activity.Timestamp } visibleMeaningful := selection.MeaningfulEvents + len(legacyScans) + len(legacyEmbeddings) if visibleMeaningful > len(events) { selection.DisplayLimitOmitted = visibleMeaningful - len(events) } return AnalysisHistory{ GeneratedAt: time.Now().UTC(), Since: since.UTC(), Events: events, Runs: runs, Timeline: timeline, Changes: changes, Totals: totals, EventCounts: counts, StatusCounts: statusCounts, DroppedEvents: dropped, DroppedDetailedChanges: uint64(persistedTruncated) + droppedChanges, DetailedChangesTruncated: uint64(persistedTruncated) + droppedChanges, RawEventCount: rawCount, ChangeCount: changeCount, Audit: audit, EventSelection: selection, RunStats: runStats, Pipelines: buildAnalysisPipelineSummary(chronological, counts), }, nil } func maxIntAnalysis(a, b int) int { if a > b { return a } return b } func (s *Store) analysisAggregate(ctx context.Context, since time.Time) (MutationStats, int, int, map[string]int, error) { var totals MutationStats var count, truncated int err := s.db.QueryRowContext(ctx, `SELECT COUNT(*), COALESCE(SUM(delta_node_created),0),COALESCE(SUM(delta_node_updated),0),COALESCE(SUM(delta_node_deleted),0), COALESCE(SUM(delta_edge_created),0),COALESCE(SUM(delta_edge_updated),0),COALESCE(SUM(delta_edge_deleted),0), COALESCE(SUM(delta_vector_created),0),COALESCE(SUM(delta_vector_updated),0),COALESCE(SUM(delta_vector_deleted),0), COALESCE(SUM(changes_truncated),0) FROM analysis_points WHERE timestamp_ns>=?`, since.UnixNano()).Scan(&count, &totals.NodesCreated, &totals.NodesUpdated, &totals.NodesDeleted, &totals.EdgesCreated, &totals.EdgesUpdated, &totals.EdgesDeleted, &totals.VectorsCreated, &totals.VectorsUpdated, &totals.VectorsDeleted, &truncated) if err != nil { return MutationStats{}, 0, 0, nil, err } rows, err := s.db.QueryContext(ctx, `SELECT type,COUNT(*) FROM analysis_events WHERE timestamp_ns>=? GROUP BY type`, since.UnixNano()) if err != nil { return MutationStats{}, 0, 0, nil, err } defer rows.Close() counts := map[string]int{} for rows.Next() { var eventType string var eventCount int if err := rows.Scan(&eventType, &eventCount); err != nil { return MutationStats{}, 0, 0, nil, err } counts[eventType] = eventCount } return totals, count, truncated, counts, rows.Err() } func (s *Store) analysisMeaningfulEvents(ctx context.Context, since time.Time, limit int) ([]AnalysisEventRecord, error) { rows, err := s.db.QueryContext(ctx, `SELECT e.id,e.type,e.source,e.phase,e.query,e.message,e.node_ids_json,e.edge_ids_json,e.strength,e.metadata_json,e.timestamp_ns, p.process_id,p.graph_version,p.node_count,p.edge_count,p.vector_count, p.node_created,p.node_updated,p.node_deleted,p.edge_created,p.edge_updated,p.edge_deleted,p.vector_created,p.vector_updated,p.vector_deleted, p.delta_node_created,p.delta_node_updated,p.delta_node_deleted,p.delta_edge_created,p.delta_edge_updated,p.delta_edge_deleted,p.delta_vector_created,p.delta_vector_updated,p.delta_vector_deleted, p.change_count,p.changes_truncated FROM analysis_events e JOIN analysis_points p ON p.event_id=e.id WHERE e.timestamp_ns>=? AND e.type<>'learning.scan.started' AND e.type<>'embedding.batch' AND NOT (e.type='learning.scan.completed' AND e.metadata_json LIKE '%"result":"unchanged"%') ORDER BY e.timestamp_ns DESC LIMIT ?`, since.UnixNano(), limit) if err != nil { return nil, err } defer rows.Close() return scanAnalysisEventRows(rows) } func scanAnalysisEventRows(rows *sql.Rows) ([]AnalysisEventRecord, error) { out := []AnalysisEventRecord{} for rows.Next() { var record AnalysisEventRecord var nodeIDs, edgeIDs, metadata string var timestamp int64 m, d := &record.Point.Mutations, &record.Point.Delta if err := rows.Scan(&record.Activity.ID, &record.Activity.Type, &record.Activity.Source, &record.Activity.Phase, &record.Activity.Query, &record.Activity.Message, &nodeIDs, &edgeIDs, &record.Activity.Strength, &metadata, ×tamp, &record.Point.ProcessID, &record.Point.GraphVersion, &record.Point.NodeCount, &record.Point.EdgeCount, &record.Point.VectorCount, &m.NodesCreated, &m.NodesUpdated, &m.NodesDeleted, &m.EdgesCreated, &m.EdgesUpdated, &m.EdgesDeleted, &m.VectorsCreated, &m.VectorsUpdated, &m.VectorsDeleted, &d.NodesCreated, &d.NodesUpdated, &d.NodesDeleted, &d.EdgesCreated, &d.EdgesUpdated, &d.EdgesDeleted, &d.VectorsCreated, &d.VectorsUpdated, &d.VectorsDeleted, &record.ChangeCount, &record.ChangesTruncated); err != nil { return nil, err } record.Activity.Timestamp = time.Unix(0, timestamp).UTC() _ = decodeJSON(nodeIDs, &record.Activity.NodeIDs) _ = decodeJSON(edgeIDs, &record.Activity.EdgeIDs) _ = decodeJSON(metadata, &record.Activity.Metadata) if record.Activity.Metadata == nil { record.Activity.Metadata = map[string]any{} } out = append(out, record) } return out, rows.Err() } func (s *Store) analysisLegacyLearningScanAggregates(ctx context.Context, since time.Time) ([]AnalysisEventRecord, error) { bucket := analysisTimelineBucket(time.Since(since)) bucketNS := int64(bucket) rows, err := s.db.QueryContext(ctx, `SELECT (e.timestamp_ns / ?) * ? AS bucket_ns,COUNT(*),MIN(e.timestamp_ns),MAX(e.timestamp_ns), MAX(p.graph_version),MAX(p.node_count),MAX(p.edge_count),MAX(p.vector_count) FROM analysis_events e JOIN analysis_points p ON p.event_id=e.id WHERE e.timestamp_ns>=? AND e.type='learning.scan.completed' AND e.metadata_json LIKE '%"result":"unchanged"%' GROUP BY bucket_ns ORDER BY bucket_ns DESC LIMIT 1000`, bucketNS, bucketNS, since.UnixNano()) if err != nil { return nil, err } defer rows.Close() out := []AnalysisEventRecord{} for rows.Next() { var bucketStart, first, last int64 var count int var version uint64 var nodes, edges, vectors int if err := rows.Scan(&bucketStart, &count, &first, &last, &version, &nodes, &edges, &vectors); err != nil { return nil, err } if count <= 0 { continue } lastAt := time.Unix(0, last).UTC() activity := model.Activity{ ID: fmt.Sprintf("legacy-learning-scan-aggregate-%d", bucketStart), Type: "learning.scan.unchanged.aggregate", Source: "brain", Phase: "indexed", Message: fmt.Sprintf("%d ältere unveränderte KB-Lernläufe im Analysefenster verdichtet", count), Strength: .18, Timestamp: lastAt, Metadata: map[string]any{ "result": "unchanged_aggregated", "scan_count": count, "equivalent_raw_events": count * 2, "first_scan_at": time.Unix(0, first).UTC(), "last_scan_at": lastAt, "legacy_compacted": true, "duration_unavailable": true, }, } out = append(out, AnalysisEventRecord{Activity: activity, Point: AnalysisPoint{ProcessID: "legacy-compaction", GraphVersion: version, NodeCount: nodes, EdgeCount: edges, VectorCount: vectors}}) } return out, rows.Err() } func (s *Store) analysisLegacyEmbeddingAggregates(ctx context.Context, since time.Time) ([]AnalysisEventRecord, error) { bucket := analysisTimelineBucket(time.Since(since)) bucketNS := int64(bucket) rows, err := s.db.QueryContext(ctx, `SELECT (e.timestamp_ns / ?) * ? AS bucket_ns,COUNT(*),MIN(e.timestamp_ns),MAX(e.timestamp_ns), MAX(p.graph_version),MAX(p.node_count),MAX(p.edge_count),MAX(p.vector_count), COALESCE(SUM(p.delta_node_created),0),COALESCE(SUM(p.delta_node_updated),0),COALESCE(SUM(p.delta_node_deleted),0), COALESCE(SUM(p.delta_edge_created),0),COALESCE(SUM(p.delta_edge_updated),0),COALESCE(SUM(p.delta_edge_deleted),0), COALESCE(SUM(p.delta_vector_created),0),COALESCE(SUM(p.delta_vector_updated),0),COALESCE(SUM(p.delta_vector_deleted),0), COALESCE(SUM(p.change_count),0),COALESCE(SUM(p.changes_truncated),0) FROM analysis_events e JOIN analysis_points p ON p.event_id=e.id WHERE e.timestamp_ns>=? AND e.type='embedding.batch' GROUP BY bucket_ns ORDER BY bucket_ns DESC LIMIT 1000`, bucketNS, bucketNS, since.UnixNano()) if err != nil { return nil, err } defer rows.Close() out := []AnalysisEventRecord{} for rows.Next() { var bucketStart, first, last int64 var count, nodes, edges, vectors, changeCount, truncated int var version uint64 var delta MutationStats if err := rows.Scan(&bucketStart, &count, &first, &last, &version, &nodes, &edges, &vectors, &delta.NodesCreated, &delta.NodesUpdated, &delta.NodesDeleted, &delta.EdgesCreated, &delta.EdgesUpdated, &delta.EdgesDeleted, &delta.VectorsCreated, &delta.VectorsUpdated, &delta.VectorsDeleted, &changeCount, &truncated); err != nil { return nil, err } if count <= 0 { continue } lastAt := time.Unix(0, last).UTC() changedVectors := delta.VectorsCreated + delta.VectorsUpdated + delta.VectorsDeleted activity := model.Activity{ ID: fmt.Sprintf("legacy-embedding-batch-aggregate-%d", bucketStart), Type: "embedding.batch.aggregate", Source: "ollama", Phase: "embedding", Message: fmt.Sprintf("%d ältere Embedding-Batches im Analysefenster verdichtet", count), Strength: .2, Timestamp: lastAt, Metadata: map[string]any{ "batch_events": count, "equivalent_raw_events": count, "vector_changes": changedVectors, "first_batch_at": time.Unix(0, first).UTC(), "last_batch_at": lastAt, "legacy_compacted": true, }, } out = append(out, AnalysisEventRecord{Activity: activity, Point: AnalysisPoint{ProcessID: "legacy-compaction", GraphVersion: version, NodeCount: nodes, EdgeCount: edges, VectorCount: vectors, Delta: delta}, ChangeCount: changeCount, ChangesTruncated: truncated}) } return out, rows.Err() } func analysisTimelineBucket(duration time.Duration) time.Duration { bucket := time.Hour if duration <= 3*time.Hour { bucket = 10 * time.Minute } else if duration <= 12*time.Hour { bucket = 30 * time.Minute } else if duration > 72*time.Hour { bucket = 6 * time.Hour } return bucket } func (s *Store) analysisLearningScanCompactionStats(ctx context.Context, since time.Time) (analysisScanCompactionStats, error) { var stats analysisScanCompactionStats if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(CASE WHEN type='learning.scan.started' THEN 1 ELSE 0 END),0), COALESCE(SUM(CASE WHEN type='learning.scan.completed' AND metadata_json LIKE '%"result":"unchanged"%' THEN 1 ELSE 0 END),0), COALESCE(SUM(CASE WHEN type='embedding.batch' THEN 1 ELSE 0 END),0) FROM analysis_events WHERE timestamp_ns>=?`, since.UnixNano()).Scan(&stats.StartedRaw, &stats.RawUnchanged, &stats.RawEmbeddingBatches); err != nil { return stats, err } rows, err := s.db.QueryContext(ctx, `SELECT metadata_json FROM analysis_events WHERE timestamp_ns>=? AND type='learning.scan.unchanged.aggregate'`, since.UnixNano()) if err != nil { return stats, err } defer rows.Close() for rows.Next() { var raw string if err := rows.Scan(&raw); err != nil { return stats, err } metadata := map[string]any{} _ = decodeJSON(raw, &metadata) runs := int(metadataNumber(metadata, "scan_count")) equivalent := int(metadataNumber(metadata, "equivalent_raw_events")) if equivalent == 0 && runs > 0 { equivalent = runs * 2 } stats.AggregateRecords++ stats.AggregateRuns += runs stats.EquivalentRaw += equivalent if equivalent > 1 { stats.AvoidedPersisted += equivalent - 1 } } if err := rows.Err(); err != nil { return stats, err } embeddingRows, err := s.db.QueryContext(ctx, `SELECT metadata_json FROM analysis_events WHERE timestamp_ns>=? AND type='embedding.batch.aggregate'`, since.UnixNano()) if err != nil { return stats, err } defer embeddingRows.Close() for embeddingRows.Next() { var raw string if err := embeddingRows.Scan(&raw); err != nil { return stats, err } metadata := map[string]any{} _ = decodeJSON(raw, &metadata) events := int(metadataNumber(metadata, "batch_events")) equivalent := int(metadataNumber(metadata, "equivalent_raw_events")) if equivalent == 0 { equivalent = events } stats.EmbeddingAggregateRecords++ stats.EmbeddingAggregateEvents += events stats.EquivalentEmbeddingRaw += equivalent if equivalent > 1 { stats.AvoidedPersisted += equivalent - 1 } } return stats, embeddingRows.Err() } func (s *Store) analysisChanges(ctx context.Context, since time.Time, limit int) ([]GraphChange, int, error) { var total int if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM analysis_changes WHERE timestamp_ns>=?`, since.UnixNano()).Scan(&total); err != nil { return nil, 0, err } rows, err := s.db.QueryContext(ctx, `SELECT id,event_id,timestamp_ns,process_id,graph_version,entity_kind,action,entity_id,label,relation_type,origin,details_json FROM analysis_changes WHERE timestamp_ns>=? ORDER BY timestamp_ns DESC,id DESC LIMIT ?`, since.UnixNano(), limit) if err != nil { return nil, 0, err } defer rows.Close() out := []GraphChange{} for rows.Next() { var change GraphChange var timestamp int64 var details string if err := rows.Scan(&change.ID, &change.EventID, ×tamp, &change.ProcessID, &change.GraphVersion, &change.EntityKind, &change.Action, &change.EntityID, &change.Label, &change.RelationType, &change.Origin, &details); err != nil { return nil, 0, err } change.Timestamp = time.Unix(0, timestamp).UTC() _ = decodeJSON(details, &change.Details) if change.Details == nil { change.Details = map[string]any{} } out = append(out, change) } return out, total, rows.Err() } func buildTimeline(events []AnalysisEventRecord, since time.Time) []AnalysisTimelineBucket { duration := time.Since(since) bucket := time.Hour if duration <= 3*time.Hour { bucket = 10 * time.Minute } else if duration <= 12*time.Hour { bucket = 30 * time.Minute } else if duration > 72*time.Hour { bucket = 6 * time.Hour } buckets := map[int64]*AnalysisTimelineBucket{} for _, event := range events { start := event.Activity.Timestamp.Truncate(bucket) key := start.UnixNano() entry := buckets[key] if entry == nil { entry = &AnalysisTimelineBucket{Start: start} buckets[key] = entry } entry.Events++ entry.Mutations.Add(event.Point.Delta) switch eventSeverity(event.Activity) { case "success": entry.Successes++ case "warning": entry.Warnings++ case "error": entry.Failures++ } entry.Comparisons += int64(metadataNumber(event.Activity.Metadata, "candidate_comparisons", "comparisons")) entry.SearchResults += int64(metadataNumber(event.Activity.Metadata, "result_count", "research_search_results")) } keys := make([]int64, 0, len(buckets)) for key := range buckets { keys = append(keys, key) } sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] }) out := make([]AnalysisTimelineBucket, 0, len(keys)) for _, key := range keys { out = append(out, *buckets[key]) } return out } func buildAnalysisRuns(events []AnalysisEventRecord) []AnalysisRun { active := map[string]*AnalysisRun{} activeOrder := []string{} completed := []AnalysisRun{} standalone := []AnalysisRun{} removeActive := func(key string) { delete(active, key) for i := len(activeOrder) - 1; i >= 0; i-- { if activeOrder[i] == key { activeOrder = append(activeOrder[:i], activeOrder[i+1:]...) break } } } for _, event := range events { kind, key, phase := classifyRunEvent(event.Activity) if strings.HasPrefix(key, "latest:") { wanted := strings.TrimPrefix(key, "latest:") key = "" for i := len(activeOrder) - 1; i >= 0; i-- { candidate := active[activeOrder[i]] if candidate != nil && candidate.Kind == wanted { key = activeOrder[i] break } } } if phase == "end" && key == "" && kind != "" { run := newAnalysisRun(kind, "terminal:"+event.Activity.ID, event) if duration := int64(metadataNumber(event.Activity.Metadata, "duration_ms")); duration > 0 { run.StartedAt = event.Activity.Timestamp.Add(-time.Duration(duration) * time.Millisecond) } finalizeAnalysisRun(&run, event.Activity) completed = append(completed, run) continue } if phase == "start" { // A duplicated start event for the same native run ID must never add // another active-order entry. Keep the original start timestamp and // attach the duplicate as an event so diagnostics stay lossless. if existing := active[key]; existing != nil { appendRunEvent(existing, event) continue } run := newAnalysisRun(kind, key, event) active[key] = &run activeOrder = append(activeOrder, key) continue } if key != "" { if run := active[key]; run != nil { appendRunEvent(run, event) if phase == "end" { finalizeAnalysisRun(run, event.Activity) completed = append(completed, *run) removeActive(key) } continue } // The analysis recorder intentionally suppresses high-frequency learning // start markers, and a selected time window can also begin after any // other run's start. Terminal events carry duration metadata, so create // a complete synthetic run instead of losing the operation entirely. if phase == "end" && kind != "" { run := newAnalysisRun(kind, key, event) if duration := int64(metadataNumber(event.Activity.Metadata, "duration_ms")); duration > 0 { run.StartedAt = event.Activity.Timestamp.Add(-time.Duration(duration) * time.Millisecond) } finalizeAnalysisRun(&run, event.Activity) completed = append(completed, run) continue } // An update carrying an explicit native run key must never be attached // to some other concurrently active workflow merely because its own // start marker was compacted or lies outside the selected window. if phase == "update" { continue } } if phase == "standalone" || isMeaningfulStandalone(event.Activity) { run := newAnalysisRun(nonemptyAnalysis(kind, kindForStandalone(event.Activity)), "standalone:"+event.Activity.ID, event) finalizeAnalysisRun(&run, event.Activity) standalone = append(standalone, run) continue } // Unresolved telemetry without a native/synthetic key is intentionally // not attached by temporal proximity. Pipeline summaries may still count // the raw event, but workflow cost/lifecycle views remain causal. continue } for _, key := range activeOrder { if run := active[key]; run != nil { // Old relation-research histories (written before research.completed // existed) can contain research.results but no terminal event whenever // no source was ingested. Do not display those historical records as // running forever. A grace period avoids prematurely closing a live // relation review that has just received its SearXNG results. if run.Kind == "research" && time.Since(run.CompletedAt) > 2*time.Minute && analysisRunHasEventType(run, "research.results") { run.Status = "success" run.Verdict = "abgeschlossen (Legacy)" run.Explanation = "Der ältere Lauf enthält SearXNG-Ergebnisse, aber kein explizites research.completed. Der Abschluss wurde aus dem letzten Ergebnis-Event rekonstruiert." completed = append(completed, *run) continue } run.Status = "running" run.Verdict = "läuft" run.Explanation = "Der Lauf ist noch nicht abgeschlossen oder sein Abschluss liegt außerhalb des gewählten Zeitfensters." completed = append(completed, *run) } } return append(completed, standalone...) } func buildAnalysisRunStats(runs []AnalysisRun) []AnalysisRunStats { type bucket struct { stats AnalysisRunStats durations []int64 } byKind := map[string]*bucket{} for _, run := range runs { kind := nonemptyAnalysis(run.Kind, "activity") entry := byKind[kind] if entry == nil { entry = &bucket{stats: AnalysisRunStats{Kind: kind}} byKind[kind] = entry } entry.stats.Runs++ entry.stats.EventCount += run.EventCount if run.MutationsKnown { entry.stats.Mutations.Add(run.Mutations) entry.stats.MutationSamples++ } switch run.Status { case "success": entry.stats.Successes++ case "warning": entry.stats.Warnings++ case "error": entry.stats.Failures++ case "running": entry.stats.Running++ default: entry.stats.Neutral++ } if run.Status != "running" && run.DurationMS > 0 { entry.durations = append(entry.durations, run.DurationMS) } } out := make([]AnalysisRunStats, 0, len(byKind)) for _, entry := range byKind { sort.Slice(entry.durations, func(i, j int) bool { return entry.durations[i] < entry.durations[j] }) if len(entry.durations) > 0 { var total int64 for _, duration := range entry.durations { total += duration } entry.stats.Duration = AnalysisDurationStats{ Samples: len(entry.durations), TotalMS: total, AverageMS: total / int64(len(entry.durations)), P50MS: percentileDuration(entry.durations, .50), P95MS: percentileDuration(entry.durations, .95), MaxMS: entry.durations[len(entry.durations)-1], } } out = append(out, entry.stats) } sort.Slice(out, func(i, j int) bool { if out[i].Duration.TotalMS == out[j].Duration.TotalMS { return out[i].Runs > out[j].Runs } return out[i].Duration.TotalMS > out[j].Duration.TotalMS }) return out } func percentileDuration(sortedValues []int64, percentile float64) int64 { if len(sortedValues) == 0 { return 0 } if percentile <= 0 { return sortedValues[0] } if percentile >= 1 { return sortedValues[len(sortedValues)-1] } index := int(math.Ceil(float64(len(sortedValues))*percentile)) - 1 if index < 0 { index = 0 } if index >= len(sortedValues) { index = len(sortedValues) - 1 } return sortedValues[index] } func buildAnalysisPipelineSummary(events []AnalysisEventRecord, counts map[string]int) AnalysisPipelineSummary { summary := AnalysisPipelineSummary{ Security: AnalysisSecuritySummary{Severities: map[string]int{}, EventTypes: map[string]int{}}, Articles: AnalysisArticleSummary{}, } summary.Security.Materialized = counts["source.security.materialized"] summary.Security.Rejected = counts["source.security.rejected"] summary.Security.Failed = counts["source.security.failed"] summary.Security.ResearchSupplements = counts["source.security.research"] summary.Articles.Created = counts["article.created"] summary.Articles.Rejected = counts["article.draft.rejected"] summary.Articles.Failed = counts["article.failed"] summary.Articles.Duplicates = counts["article.duplicate"] summary.Articles.Skipped = counts["article.skipped"] + counts["article.plan.skipped"] summary.Articles.Reviews = counts["article.review.completed"] summary.Articles.WebResearchFetches = counts["article.research.fetch.completed"] confidenceSum := 0.0 for _, event := range events { a := event.Activity switch a.Type { case "source.security.materialized": if confidence, ok := numericMetadata(a.Metadata, "confidence"); ok && confidence > 0 { confidenceSum += confidence summary.Security.ConfidenceSamples++ } severity := strings.ToLower(strings.TrimSpace(metadataString(a.Metadata, "severity"))) if severity == "" { severity = "unknown" } summary.Security.Severities[severity]++ eventType := strings.ToLower(strings.TrimSpace(metadataString(a.Metadata, "event_type"))) if eventType == "" { eventType = "unknown" } summary.Security.EventTypes[eventType]++ summary.Security.SupplementalSources += int(metadataNumber(a.Metadata, "supplemental_sources")) case "article.review.completed": summary.Articles.SupportedClaims += int(metadataNumber(a.Metadata, "supported_claims")) summary.Articles.PartialClaims += int(metadataNumber(a.Metadata, "partially_supported_claims")) summary.Articles.UnsupportedClaims += int(metadataNumber(a.Metadata, "unsupported_claims_count")) summary.Articles.ContradictedClaims += int(metadataNumber(a.Metadata, "contradicted_claims")) case "article.research.inbox": summary.Articles.InboxResearchHits += int(metadataNumber(a.Metadata, "results")) } } if summary.Security.ConfidenceSamples > 0 { summary.Security.AverageConfidence = confidenceSum / float64(summary.Security.ConfidenceSamples) } return summary } func analysisRunHasEventType(run *AnalysisRun, typeName string) bool { if run == nil { return false } for _, event := range run.Events { if event.Activity.Type == typeName { return true } } return false } func newAnalysisRun(kind, key string, event AnalysisEventRecord) AnalysisRun { if kind == "" { kind = kindForStandalone(event.Activity) } run := AnalysisRun{ ID: key, Kind: kind, Title: runTitle(kind, event.Activity), Status: "running", Verdict: "läuft", StartedAt: event.Activity.Timestamp, Metrics: map[string]any{}, } appendRunEvent(&run, event) return run } func appendRunEvent(run *AnalysisRun, event AnalysisEventRecord) { run.EventCount++ if mutations, ok := explicitActivityMutations(event.Activity); ok { if run.Kind == "learning" && event.Activity.Type == "learning.scan.completed" { // learning.scan.completed is the authoritative causal snapshot for the // whole scan. Intermediate graph.updated / embedding.identity_changed // events may carry subsets or the same cumulative stats and must not be // added a second time. run.Mutations = mutations run.MutationsKnown = true run.MutationAttribution = nonemptyAnalysis(metadataString(event.Activity.Metadata, "mutation_attribution"), "explicit-terminal") } else if !(run.Kind == "learning" && event.Activity.Type == "graph.updated") { run.Mutations.Add(mutations) run.MutationsKnown = true run.MutationAttribution = nonemptyAnalysis(metadataString(event.Activity.Metadata, "mutation_attribution"), "explicit") } } run.NodeIDs = uniqueStrings(append(run.NodeIDs, event.Activity.NodeIDs...)) run.EdgeIDs = uniqueStrings(append(run.EdgeIDs, event.Activity.EdgeIDs...)) if len(run.Events) < 80 { run.Events = append(run.Events, event) } if trigger := metadataString(event.Activity.Metadata, "trigger", "requested_by"); trigger != "" { run.Trigger = trigger } mergeRunMetrics(run.Metrics, event.Activity) if event.Activity.Timestamp.After(run.CompletedAt) { run.CompletedAt = event.Activity.Timestamp } } func finalizeAnalysisRun(run *AnalysisRun, terminal model.Activity) { run.CompletedAt = terminal.Timestamp run.DurationMS = run.CompletedAt.Sub(run.StartedAt).Milliseconds() if duration := int64(metadataNumber(terminal.Metadata, "duration_ms")); duration > 0 { run.DurationMS = duration } run.Outcome = metadataString(terminal.Metadata, "result", "outcome", "reason") run.Status = eventSeverity(terminal) outcome := strings.ToLower(strings.TrimSpace(run.Outcome)) if run.Status != "error" { switch outcome { case "no_candidate", "unchanged", "idle", "completed_no_change": run.Status = "neutral" case "skipped", "rejected", "no_useful_evidence", "deferred", "disabled": run.Status = "warning" } derived := verdictFromRun(run) if derived == "success" { run.Status = "success" } else if run.Status == "neutral" && derived == "warning" { run.Status = "warning" } } run.Verdict, run.Explanation = explainRun(run, terminal) } func analysisEventOrder(activity model.Activity) int { _, _, phase := classifyRunEvent(activity) switch phase { case "start": return 0 case "update": return 1 case "end": return 2 case "standalone": return 3 default: return 1 } } func classifyRunEvent(activity model.Activity) (kind, key, phase string) { typeName := activity.Type switch typeName { case "think.cycle.started": return "thinking", "thinking:" + activity.ID, "start" case "think.cycle.completed", "think.cycle.failed": return "thinking", latestSyntheticKey("thinking"), "end" case "learning.scan.started": return "learning", nonemptyAnalysis(metadataString(activity.Metadata, "run_id"), "learning:"+activity.ID), "start" case "learning.scan.completed", "learning.scan.failed": return "learning", nonemptyAnalysis(metadataString(activity.Metadata, "run_id"), latestSyntheticKey("learning")), "end" case "source.security.started": return "security-source", "security:" + nonemptyAnalysis(metadataString(activity.Metadata, "run_id", "inbox_id"), activity.ID), "start" case "source.security.materialized", "source.security.rejected", "source.security.failed": return "security-source", "security:" + nonemptyAnalysis(metadataString(activity.Metadata, "run_id", "inbox_id"), activity.ID), "end" case "article.plan.started": if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "article", "article:" + runID, "start" } return "article", "article:" + activity.ID, "start" case "article.created", "article.draft.rejected", "article.failed", "article.duplicate", "article.skipped", "article.plan.skipped": if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "article", "article:" + runID, "end" } return "article", latestSyntheticKey("article"), "end" case "autonomous.research.task.started": return "autonomous-research", "autonomous:" + nonemptyAnalysis(metadataString(activity.Metadata, "task_id"), activity.ID), "start" case "autonomous.research.task.completed", "autonomous.research.task.failed", "autonomous.research.task.cancelled": return "autonomous-research", "autonomous:" + nonemptyAnalysis(metadataString(activity.Metadata, "task_id"), activity.ID), "end" case "query.started": return "query", "query:" + nonemptyAnalysis(metadataString(activity.Metadata, "run_id"), activity.ID), "start" case "query.completed", "query.failed": if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "query", "query:" + runID, "end" } return "query", latestSyntheticKey("query"), "end" case "research.test.started": return "searxng-test", "research-test:" + activity.ID, "start" case "research.test.results", "research.test.failed": return "searxng-test", latestSyntheticKey("searxng-test"), "end" } if strings.HasPrefix(typeName, "source.security.") { return "security-source", "security:" + nonemptyAnalysis(metadataString(activity.Metadata, "run_id", "inbox_id"), activity.ID), "update" } if strings.HasPrefix(typeName, "article.research.") { // A concrete SearXNG research_id owns its own nested research run. // Routing/cache/grounding events without research_id still belong to // the native article lifecycle when one is present. if metadataString(activity.Metadata, "research_id") == "" { if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "article", "article:" + runID, "update" } } } if strings.HasPrefix(typeName, "article.") && !strings.HasPrefix(typeName, "article.research.") { if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "article", "article:" + runID, "update" } // Cluster scheduling/defer events happen outside the concrete article // lifecycle. Never attach them to whichever article happens to be open. if typeName == "article.cluster.started" || typeName == "article.cluster.deferred" { return "activity", "standalone:" + activity.ID, "standalone" } return "article", latestSyntheticKey("article"), "update" } if strings.HasPrefix(typeName, "think.") { return "thinking", latestSyntheticKey("thinking"), "update" } if typeName == "scan.started" { return "learning", latestSyntheticKey("learning"), "update" } if typeName == "graph.updated" { if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "learning", runID, "update" } return "activity", "standalone:" + activity.ID, "standalone" } if strings.HasPrefix(typeName, "query.") || typeName == "node.activated" || typeName == "edges.traversed" { if runID := metadataString(activity.Metadata, "run_id"); runID != "" { return "query", "query:" + runID, "update" } return "query", latestSyntheticKey("query"), "update" } if strings.HasPrefix(typeName, "embedding.") { return "embedding", "standalone:" + activity.ID, "standalone" } if strings.HasPrefix(typeName, "persistence.") || typeName == "agent.run" || strings.HasPrefix(typeName, "glpi.kb.") { return kindForStandalone(activity), "standalone:" + activity.ID, "standalone" } // Only the root lifecycle events open or close a research run. Nested // operations such as article.research.fetch.started/completed are updates of // the parent run, not independent processes. Treating every *.started as a // new run caused the same research_id to be appended to activeOrder once per // fetched page and made a single query appear five to seven times as // "running" in the Analysis Center. if strings.HasPrefix(typeName, "research.") || strings.HasPrefix(typeName, "article.research.") { id := metadataString(activity.Metadata, "research_id") if id != "" { switch typeName { case "research.started", "article.research.started": return "research", "research:" + id, "start" case "research.completed", "research.failed", "article.research.completed", "article.research.failed": return "research", "research:" + id, "end" default: return "research", "research:" + id, "update" } } } return "", "", "" } // latestSyntheticKey is a marker resolved by buildAnalysisRuns. Terminal events // without a native run ID are attached to the most recent active run of the // same kind. func latestSyntheticKey(kind string) string { return "latest:" + kind } func kindForStandalone(activity model.Activity) string { switch { case strings.HasPrefix(activity.Type, "glpi.kb."): return "glpi-sync" case strings.HasPrefix(activity.Type, "persistence."): return "persistence" case strings.HasPrefix(activity.Type, "article."): return "article" case activity.Type == "learning.scan.unchanged.aggregate": return "learning-summary" case strings.HasPrefix(activity.Type, "embedding."): return "embedding" case strings.HasPrefix(activity.Type, "autonomous.research.scan."): return "opportunity-scan" case activity.Type == "agent.run": return "agent" default: return "activity" } } func isMeaningfulStandalone(activity model.Activity) bool { if activity.Type == "brain.idle" || activity.Type == "think.queued" { return false } // graph.updated with a native learning run_id is only an intermediate // mutation snapshot. learning.scan.completed is the authoritative terminal // event and already carries the same causal stats. Never duplicate it as a // standalone learning run when the start marker was compacted away. if activity.Type == "graph.updated" && metadataString(activity.Metadata, "run_id") != "" { return false } return strings.Contains(activity.Type, "completed") || strings.Contains(activity.Type, "failed") || strings.Contains(activity.Type, "created") || strings.Contains(activity.Type, "synced") || strings.Contains(activity.Type, "flushed") || activity.Type == "agent.run" || activity.Type == "graph.updated" || activity.Type == "learning.scan.unchanged.aggregate" || activity.Type == "embedding.batch.aggregate" } func runTitle(kind string, activity model.Activity) string { switch kind { case "thinking": return "AI-THINK-Zyklus" case "learning": return "KB-Lernlauf" case "learning-summary": return "Unveränderte KB-Scans (verdichtet)" case "security-source": return nonemptyAnalysis(metadataString(activity.Metadata, "title"), "Proaktive Security-Meldung") case "autonomous-research": return nonemptyAnalysis(metadataString(activity.Metadata, "topic"), nonemptyAnalysis(activity.Message, "Autonome Recherche")) case "query": return nonemptyAnalysis(activity.Query, "Wissensabfrage") case "research", "searxng-test": return nonemptyAnalysis(metadataString(activity.Metadata, "research_query", "query"), nonemptyAnalysis(activity.Query, "SearXNG-Recherche")) case "article": return nonemptyAnalysis(metadataString(activity.Metadata, "title"), "Artikelsynthese") case "glpi-sync": return "GLPI-KB-Synchronisierung" case "persistence": return "Persistenz-Flush" case "embedding": return "Embedding-Lauf" case "opportunity-scan": return "Autonome Wissenslückensuche" case "agent": return "Agent-Lauf" default: return nonemptyAnalysis(activity.Message, activity.Type) } } func eventSeverity(activity model.Activity) string { typeName := strings.ToLower(activity.Type) message := strings.ToLower(activity.Message) if strings.Contains(typeName, "failed") || strings.Contains(typeName, "error") || strings.Contains(message, "fehlgeschlagen") { return "error" } if strings.Contains(typeName, "skipped") || strings.Contains(typeName, "rejected") || strings.Contains(typeName, "no_candidate") || strings.Contains(typeName, "paused") || strings.Contains(typeName, "deferred") || strings.Contains(message, "übersprungen") { return "warning" } if strings.Contains(typeName, "completed") || strings.Contains(typeName, "created") || strings.Contains(typeName, "materialized") || strings.Contains(typeName, "accepted") || strings.Contains(typeName, "ingested") || strings.Contains(typeName, "learned") || strings.Contains(typeName, "synced") || strings.Contains(typeName, "results") || strings.Contains(typeName, "flushed") { return "success" } return "neutral" } func verdictFromRun(run *AnalysisRun) string { if hasFailureEvent(run.Events) { return "error" } if !run.Mutations.Empty() { return "success" } if hasWarningEvent(run.Events) { return "warning" } return "neutral" } func explainRun(run *AnalysisRun, terminal model.Activity) (string, string) { status := run.Status if status == "error" { return "fehlgeschlagen", nonemptyAnalysis(metadataString(terminal.Metadata, "error"), nonemptyAnalysis(terminal.Message, "Der Lauf wurde mit einem Fehler beendet.")) } if run.Mutations.NodesCreated > 0 || run.Mutations.EdgesCreated > 0 || run.Mutations.VectorsCreated > 0 { parts := []string{} if run.Mutations.NodesCreated > 0 { parts = append(parts, fmt.Sprintf("%d neue Nodes", run.Mutations.NodesCreated)) } if run.Mutations.EdgesCreated > 0 { parts = append(parts, fmt.Sprintf("%d neue Edges", run.Mutations.EdgesCreated)) } if run.Mutations.VectorsCreated > 0 { parts = append(parts, fmt.Sprintf("%d neue Embeddings", run.Mutations.VectorsCreated)) } if run.Mutations.VectorsUpdated > 0 { parts = append(parts, fmt.Sprintf("%d neu berechnete Embeddings", run.Mutations.VectorsUpdated)) } return "positives Ergebnis", "Der Lauf hat den Wissenszustand materiell erweitert: " + strings.Join(parts, ", ") + "." } if run.Mutations.NodesDeleted+run.Mutations.EdgesDeleted+run.Mutations.VectorsDeleted > 0 { return "Bereinigung", fmt.Sprintf("Der Lauf hat veraltete Daten entfernt: %d Nodes, %d Edges und %d Embeddings.", run.Mutations.NodesDeleted, run.Mutations.EdgesDeleted, run.Mutations.VectorsDeleted) } if run.Mutations.NodesUpdated+run.Mutations.EdgesUpdated+run.Mutations.VectorsUpdated > 0 { return "aktualisiert", fmt.Sprintf("Bestehendes Wissen wurde verändert: %d Nodes, %d Edges und %d Embeddings wurden aktualisiert oder neu berechnet.", run.Mutations.NodesUpdated, run.Mutations.EdgesUpdated, run.Mutations.VectorsUpdated) } if status == "warning" { return "ohne Übernahme", nonemptyAnalysis(terminal.Message, "Der Lauf hat geprüft, aber wegen Qualitäts- oder Relevanzregeln nichts in den Graphen übernommen.") } return "ohne Graphänderung", nonemptyAnalysis(terminal.Message, "Der Lauf wurde beendet, hat aber keine Nodes, Edges oder Embeddings verändert.") } func hasFailureEvent(events []AnalysisEventRecord) bool { for _, event := range events { if eventSeverity(event.Activity) == "error" { return true } } return false } func hasWarningEvent(events []AnalysisEventRecord) bool { for _, event := range events { if eventSeverity(event.Activity) == "warning" { return true } } return false } func mergeRunMetrics(metrics map[string]any, activity model.Activity) { if metrics == nil { return } for _, keys := range [][]string{{"comparisons", "candidate_comparisons"}, {"exact_comparisons"}, {"coarse_comparisons"}, {"candidate_pool"}, {"checked"}, {"relations_created"}, {"articles_created"}, {"articles_skipped"}, {"research_search_results", "result_count"}, {"research_fetched", "pages_fetched"}, {"research_accepted", "evidence_count"}, {"research_rejected"}, {"queries_executed"}, {"batch_count"}, {"duration_ms"}} { name := keys[0] value := metadataNumber(activity.Metadata, keys...) if value == 0 { continue } previous, _ := metrics[name].(float64) metrics[name] = previous + value } if similarity := metadataNumber(activity.Metadata, "semantic_similarity"); similarity > 0 { metrics["last_semantic_similarity"] = similarity } if confidence := metadataNumber(activity.Metadata, "confidence"); confidence > 0 { metrics["last_confidence"] = confidence } if modelName := metadataString(activity.Metadata, "model"); modelName != "" { metrics["model"] = modelName } if mode := metadataString(activity.Metadata, "processing_mode"); mode != "" { metrics["processing_mode"] = mode } } func metadataNumber(metadata map[string]any, keys ...string) float64 { for _, key := range keys { if value, ok := numericMetadata(metadata, key); ok { return value } } return 0 } func metadataString(metadata map[string]any, keys ...string) string { for _, key := range keys { if metadata == nil { return "" } value, ok := metadata[key] if !ok || value == nil { continue } text := strings.TrimSpace(fmt.Sprint(value)) if text != "" && !strings.EqualFold(text, "") && !strings.EqualFold(text, "null") { return text } } return "" } func uniqueStrings(values []string) []string { seen := map[string]struct{}{} out := make([]string, 0, len(values)) for _, value := range values { value = strings.TrimSpace(value) if value == "" { continue } if _, exists := seen[value]; exists { continue } seen[value] = struct{}{} out = append(out, value) } return out } func (s *Store) countNodeCreatedLocked() { s.mutations.NodesCreated++ } func (s *Store) countNodeUpdatedLocked() { s.mutations.NodesUpdated++ } func (s *Store) countNodeDeletedLocked() { s.mutations.NodesDeleted++ } func (s *Store) countEdgeCreatedLocked() { s.mutations.EdgesCreated++ } func (s *Store) countEdgeUpdatedLocked() { s.mutations.EdgesUpdated++ } func (s *Store) countEdgeDeletedLocked() { s.mutations.EdgesDeleted++ } func (s *Store) countVectorCreatedLocked() { s.mutations.VectorsCreated++ } func (s *Store) countVectorUpdatedLocked() { s.mutations.VectorsUpdated++ } func (s *Store) countVectorDeletedLocked() { s.mutations.VectorsDeleted++ } const analysisDetailedChangeLimit = 2000 func (s *Store) recordChangeLocked(change GraphChange) { change.Timestamp = time.Now().UTC() change.ProcessID = s.processID change.GraphVersion = s.version if change.Details == nil { change.Details = map[string]any{} } if len(s.analysisPendingChanges) < analysisDetailedChangeLimit { s.analysisPendingChanges = append(s.analysisPendingChanges, change) return } s.analysisPendingTruncated++ } func nodeChange(node model.Node, action string) GraphChange { return GraphChange{EntityKind: "node", Action: action, EntityID: node.ID, Label: node.Label, Origin: node.Origin, Details: map[string]any{"kind": node.Kind, "status": node.Status, "source": explicitNodeSource(node)}} } func nodeUpdateChange(previous, current model.Node) GraphChange { change := nodeChange(current, "updated") change.Details["previous_label"] = previous.Label change.Details["previous_status"] = previous.Status change.Details["previous_source"] = explicitNodeSource(previous) return change } func edgeChange(edge model.Edge, action string) GraphChange { details := map[string]any{"source": edge.Source, "target": edge.Target, "status": edge.Status, "confidence": edge.Confidence} if similarity, ok := numericMetadata(edge.Metadata, "semantic_similarity"); ok { details["semantic_similarity"] = similarity } return GraphChange{EntityKind: "edge", Action: action, EntityID: edge.ID, RelationType: edge.Type, Origin: edge.Origin, Details: details} } func edgeUpdateChange(previous, current model.Edge) GraphChange { change := edgeChange(current, "updated") change.Details["previous_status"] = previous.Status change.Details["previous_confidence"] = previous.Confidence if similarity, ok := numericMetadata(previous.Metadata, "semantic_similarity"); ok { change.Details["previous_semantic_similarity"] = similarity } return change } func vectorRecalculatedChange(nodeID string, previousDimensions, dimensions int, label string) GraphChange { change := vectorChange(nodeID, "recalculated", dimensions, label) change.Details["previous_dimensions"] = previousDimensions return change } func vectorChange(nodeID, action string, dimensions int, label string) GraphChange { return GraphChange{EntityKind: "vector", Action: action, EntityID: nodeID, Label: label, Origin: "embedding", Details: map[string]any{"dimensions": dimensions}} } // ReconcileSecurityLifecycles uses the Source Inbox state as the authoritative // lifecycle for proactive Security work. Audit events remain valuable detail, // but a missing terminal event must never leave a completed inbox item shown as // "running". This also makes the dashboard resilient to historical event-ID // collisions and process interruptions. func ReconcileSecurityLifecycles(history *AnalysisHistory, records []AnalysisSecurityLifecycle) { if history == nil || len(records) == 0 { return } byID := make(map[string]int, len(history.Runs)) for i := range history.Runs { if history.Runs[i].Kind == "security-source" { byID[history.Runs[i].ID] = i } } reconciled := 0 for _, record := range records { keyPart := strings.TrimSpace(record.RunID) if keyPart == "" { keyPart = strings.TrimSpace(record.InboxID) } if keyPart == "" || record.StartedAt.IsZero() { continue } key := "security:" + keyPart idx, exists := byID[key] if !exists && record.InboxID != "" { legacyKey := "security:" + record.InboxID idx, exists = byID[legacyKey] } if !exists { run := AnalysisRun{ID: key, Kind: "security-source", Title: nonemptyAnalysis(record.Title, "Proaktive Security-Meldung"), StartedAt: record.StartedAt, Metrics: map[string]any{}, Status: "running", Verdict: "läuft"} history.Runs = append(history.Runs, run) idx = len(history.Runs) - 1 byID[key] = idx reconciled++ } run := &history.Runs[idx] if run.Metrics == nil { run.Metrics = map[string]any{} } run.Metrics["lifecycle_source"] = "source-inbox" run.Metrics["inbox_id"] = record.InboxID run.Metrics["proactive_state"] = record.ProactiveState if record.Confidence > 0 { run.Metrics["confidence"] = record.Confidence } if record.Severity != "" { run.Metrics["severity"] = record.Severity } if record.EventType != "" { run.Metrics["event_type"] = record.EventType } if record.MaterializedNodeID != "" { run.NodeIDs = uniqueStrings(append(run.NodeIDs, record.MaterializedNodeID)) } run.Mutations = record.Mutations run.MutationsKnown = true run.MutationAttribution = "source-inbox-store" if record.DurationMS > 0 { run.DurationMS = record.DurationMS } if !record.CompletedAt.IsZero() { run.CompletedAt = record.CompletedAt if run.DurationMS <= 0 { run.DurationMS = record.CompletedAt.Sub(record.StartedAt).Milliseconds() } } outcome := strings.ToLower(strings.TrimSpace(record.Outcome)) state := strings.ToLower(strings.TrimSpace(record.ProactiveState)) status := strings.ToLower(strings.TrimSpace(record.Status)) switch { case outcome == "materialized" || state == "done" || status == "materialized" || status == "used": if run.Status == "running" || run.Status == "neutral" || run.Status == "" { reconciled++ } run.Status = "success" run.Verdict = "positives Ergebnis" run.Outcome = "materialized" run.Explanation = "Der Abschluss wurde mit dem autoritativen Source-Inbox-State abgeglichen; der Security-Node ist materialisiert." case outcome == "rejected" || state == "rejected": if run.Status == "running" || run.Status == "neutral" || run.Status == "" { reconciled++ } run.Status = "warning" run.Verdict = "verworfen" run.Outcome = "rejected" run.Explanation = "Der Source-Inbox-State bestätigt, dass die proaktive Materialisierung fachlich verworfen wurde; der Candidate bleibt als passive Evidenz erhalten." case outcome == "failed_retry": if run.Status == "running" || run.Status == "neutral" || run.Status == "" { reconciled++ } run.Status = "warning" run.Verdict = "Retry eingeplant" run.Outcome = "failed_retry" run.Explanation = nonemptyAnalysis(record.LastError, "Der letzte Security-Versuch ist fehlgeschlagen und wurde mit Backoff erneut eingeplant.") case state == "processing": run.Status = "running" run.Verdict = "läuft" run.Explanation = "Der Source-Inbox-State bestätigt einen aktuell laufenden Security-Worker." case state == "queued": run.Status = "warning" run.Verdict = "wartet" run.Explanation = "Der Security-Candidate wartet in der proaktiven Queue auf seinen nächsten Versuch." } } sort.Slice(history.Runs, func(i, j int) bool { return history.Runs[i].StartedAt.After(history.Runs[j].StartedAt) }) history.RunStats = buildAnalysisRunStats(history.Runs) security := history.Pipelines.Security security.Materialized = 0 security.Rejected = 0 security.Failed = 0 security.Severities = map[string]int{} security.EventTypes = map[string]int{} security.AverageConfidence = 0 security.ConfidenceSamples = 0 confidenceTotal := 0.0 for _, record := range records { outcome := strings.ToLower(strings.TrimSpace(record.Outcome)) state := strings.ToLower(strings.TrimSpace(record.ProactiveState)) status := strings.ToLower(strings.TrimSpace(record.Status)) materialized := outcome == "materialized" || state == "done" || status == "materialized" || status == "used" if materialized { security.Materialized++ severity := strings.ToLower(strings.TrimSpace(record.Severity)) if severity == "" { severity = "unknown" } security.Severities[severity]++ eventType := strings.ToLower(strings.TrimSpace(record.EventType)) if eventType == "" { eventType = "unknown" } security.EventTypes[eventType]++ if record.Confidence > 0 { confidenceTotal += record.Confidence security.ConfidenceSamples++ } } else if outcome == "rejected" || state == "rejected" { security.Rejected++ } else if outcome == "failed_retry" { security.Failed++ } } if security.ConfidenceSamples > 0 { security.AverageConfidence = confidenceTotal / float64(security.ConfidenceSamples) } security.AuthoritativeRecords = len(records) security.ReconciledRuns = reconciled history.Pipelines.Security = security }