jbergner
440423c5b6
All checks were successful
release-tag / release-image (push) Successful in 2m43s
2408 lines
92 KiB
Go
2408 lines
92 KiB
Go
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<?`, cutoff)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) MutationStats() MutationStats {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return s.mutations
|
|
}
|
|
|
|
func (s *Store) DetailedAnalysis() DetailedGraphAnalysis {
|
|
s.mu.RLock()
|
|
requestedVersion := s.version
|
|
s.mu.RUnlock()
|
|
|
|
s.analysisDetailMu.Lock()
|
|
defer s.analysisDetailMu.Unlock()
|
|
if s.analysisDetailVersion == requestedVersion && s.analysisDetailCache.Summary.NodeCount > 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, "<nil>") || 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, "<nil>") && !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
|
|
}
|